从Airflow连接ID获取SQLAlchemy引擎的最佳方法是什么？
当前，我正在创建一个钩子，检索其URI，然后使用它来创建SQLAlchemy引擎。
postgres_hook = PostgresHook(self.postgres_conn_id)
engine = create_engine(postgres_hook.get_uri())
这可行，但是两个命令都建立了到数据库的连接。
当我在连接上具有“额外”参数时，需要第三个连接来检索那些参数（请参阅从Airflow Postgres钩子检索完整的连接URI）
有没有更短，更直接的方法？
需要明确的是，实际上您的命令将建立两个数据库连接，但连接到两个单独的数据库（除非您尝试连接到Postgres Airflow数据库）。初始化挂钩的第一行不应进行任何连接。只有第二行首先从Airflow数据库中获取连接详细信息（我认为您无法避免），然后使用它来连接到Postgres数据库（我认为这是重点）。
您可以使用以下方法使它稍微简单一些：
postgres_hook = PostgresHook(self.postgres_conn_id)
engine = postgres_hook.get_sqlalchemy_engine()
看起来很干净，但是如果您想不通过而获得更直接的指示PostgresHook，则可以通过查询Airflow的数据库直接获取。但是，这意味着您将最终复制代码以从连接对象构建URI。如果要继续进行此操作，则get_connection（）的基础实现是一个很好的示例。
from airflow.settings import Session
conn = session.query(Connection).filter(Connection.conn_id == self.postgres_conn_id).one()
... # build uri from connection
create_engine(uri)
此外，如果您希望能够访问extras而无需单独的数据库中获取超越get_uri()或get_sqlalchemy_engine()不，是可以覆盖BaseHook.get_connection（）来连接对象保存到重用一个实例变量。这将需要在之上创建您自己的钩子PostgresHook，因此我知道这可能不是理想的选择。
class CustomPostgresHook(PostgresHook):
@classmethod
def get_connection(cls, conn_id): # type: (str) -> Connection
conn = super().get_connection(conn_id)
self.conn_obj = conn # can't use self.conn because PostgresHook will overriden in https://github.com/apache/airflow/blob/1.10.10/airflow/hooks/postgres_hook.py#L93 by a different type of connection
return conn
postgres_hook = CustomPostgresHook(self.postgres_conn_id)
uri = postgres_hook.get_uri()
# do something with postgres_hook.conn_obj.extras_dejson
某些内置的Airflow挂钩已经具有此行为（grpc，samba，tableau），但绝对不是标准化的。
本文收集自互联网，转载请注明来源。
如有侵权，请联系[email protected] 删除。
我来说两句