Подключение к базе данных Yandex Managed Service for Apache Airflow™

Вы можете программно подключиться к базе данных Managed Service for Apache Airflow™ и обновить данные подключений с помощью направленного ациклического графа (DAG).

Подготовьте DAG-файл и запустите графПодготовьте DAG-файл и запустите граф

1. Создайте локально файл с именем update_connections.py и скопируйте в него скрипт:

   import json
   from airflow.decorators import dag, task
   from airflow.settings import Session
   from airflow.models import Connection
   
   
   @dag(schedule=None)
   def update_connections():
       @task
       def update_connections_task():
           with Session() as session:
               connections = session.query(Connection)
               for conn in connections:
                   extra = conn.extra_dejson
                   print(f"extra: {extra}")
                   update_count = extra.get('update_count', 0)
                   extra['update_count'] = update_count + 1
                   conn.set_extra(json.dumps(extra))
                   session.add(conn)
               session.commit()
   
       update_connections_task()
   
   
   update_connections()
   

2. Загрузите DAG-файл update_connections.py в созданный ранее бакет. В результате одноименный граф появится в веб-интерфейсе Apache Airflow™ автоматически.

3. Откройте веб-интерфейс Apache Airflow™.

4. Убедитесь, что в разделе DAGs появился новый граф update_connections.

   Загрузка DAG-файла из бакета может занять несколько минут.

5. Чтобы запустить граф, в строке с его именем нажмите кнопку .

Проверьте результатПроверьте результат

Чтобы проверить результат в веб-интерфейсе Apache Airflow™:

1. В разделе DAGs откройте граф update_connections.
2. Перейдите в раздел Graph.
3. Выберите задание update_connections_task.
4. Перейдите в раздел Logs.
5. Убедитесь, что в логах есть список обновленных подключений. Это значит, что запрос выполнен успешно.