Есть ли способ создавать/изменять соединения через API Airflow

Пройдя через Admin → Connections, у нас есть возможность создавать/изменять параметры подключения, но мне интересно, могу ли я сделать то же самое через API, чтобы я мог программно устанавливать соединения

airflow.models.Connection похоже, что он имеет дело только с фактическим подключением к экземпляру, а не с сохранением его в списке. Это похоже на функцию, которая должна была быть реализована, но я не уверен, где я могу найти документы для этой конкретной функции.

Ответы

Ответ 1

Соединение на самом деле является моделью, которую вы можете использовать для запроса и вставки нового соединения

from airflow import settings
from airflow.models import Connection
conn = Connection(
        conn_id=conn_id,
        conn_type=conn_type,
        host=host,
        login=login,
        password=password,
        port=port
) #create a connection object
session = settings.Session() # get the session
session.add(conn)
session.commit() # it will insert the connection object programmatically.

Ответ 2

Вы также можете добавлять, удалять и перечислять соединения из CLI Airflow, если вам нужно сделать это вне кода Python/Airflow, через bash, в Dockerfile и т.д.

airflow connections --add ...

Использование:

airflow connections [-h] [-l] [-a] [-d] [--conn_id CONN_ID]
                    [--conn_uri CONN_URI] [--conn_extra CONN_EXTRA]
                    [--conn_type CONN_TYPE] [--conn_host CONN_HOST]
                    [--conn_login CONN_LOGIN] [--conn_password CONN_PASSWORD]
                    [--conn_schema CONN_SCHEMA] [--conn_port CONN_PORT]

https://airflow.apache.org/cli.html#connections

Это не похоже на то, что CLI в настоящее время поддерживает изменение существующего соединения, но есть проблема Jira для него с активным открытым PR на GitHub.

Ответ 3

код ниже непрерывно продолжает создавать соединения (бесконечно), как это остановить?

from airflow import settings
from airflow.models import Connection
conn = Connection(
        conn_id=conn_id,
        conn_type=conn_type,
        host=host,
        login=login,
        password=password,
        port=port
) #create a connection object
session = settings.Session() # get the session
session.add(conn)
session.commit() # it will insert the connection object programmatically.

Обратите внимание, я изменил параметр в connection() в моей системе.