Pythonコードの環境変数として気流接続を使用する方法

Question

AIRFLOW_CONN_を使用して気流環境変数にアクセスし、Pythonコードで使用する方法を知っている人はいませんか。パスワードを取得するためにフックを使うことはできますが、pythonでAIRFLOW_CONNを使用してデータベースに接続しようとしています。 Airflow UIとそのドキュメントに接続を保存しました。彼らはconn_idにAIRFLOW_CONN_接頭辞を使用して使用すると述べました。 os.environ ['AIRFLOW_CONN_REDSHIFT']を使用してPythonコードで使用しましたが、環境変数を識別しません。助けてください。

Answer 1

データベースへの接続を保存し、AIRFLOW_CONN_環境変数を設定することは、接続を追加する2つの方法です。接続IDの下に格納しない限り、一方向のみを選択してください。

PythonOperatorのような演算子でPythonコードを実行していると仮定すると、BaseHookのように接続をフェッチすることができます。

は、データベースに格納された：

環境変数に格納され

@classmethod 
def _get_connections_from_db(cls, conn_id): 
    session = settings.Session() 
    db = (
     session.query(Connection) 
     .filter(Connection.conn_id == conn_id) 
     .all() 
    ) 
    session.expunge_all() 
    session.close() 
    if not db: 
     raise AirflowException(
      "The conn_id `{0}` isn't defined".format(conn_id)) 
    return db 

：私はこのコードの重複を避けるためにフックを経由して、それを取り出す推薦が

@classmethod 
def _get_connection_from_env(cls, conn_id): 
    environment_uri = os.environ.get(CONN_ENV_PREFIX + conn_id.upper()) 
    conn = None 
    if environment_uri: 
     conn = Connection(conn_id=conn_id, uri=environment_uri) 
    return conn 

！