airflow-v3-migrator 0.2.0

Автоматическая миграция DAG-файлов Airflow 2.x -> 3.x с интеграцией GitLab и Airflow API

airflow-v3-migrator

Инструмент автоматической миграции DAG-файлов Airflow 2.x → Airflow 3.x.

Делает три вещи:

1. Диагностика — находит устаревшие модули, операторы и параметры (provide_context, @apply_defaults, schedule_interval, DummyOperator, старые импорты).
2. Авто-исправление — заменяет импорты на новые SDK-пути, удаляет устаревшие аргументы, переименовывает операторы.
3. Интеграция с GitLab — читает файлы DAG из исходной ветки (prod), применяет правки и коммитит результат в целевую ветку (v3-prod). Зависимые файлы (.sql, .json и т.д.) переносятся без изменений.

Установка

pip install .
# или режим разработки:
pip install -e .

Для шаринга можно собрать wheel и передать коллегам:

pip install build
python -m build          # создаст dist/airflow_v3_migrator-0.1.0-py3-none-any.whl
# коллега ставит:
pip install airflow_v3_migrator-0.1.0-py3-none-any.whl

Настройка

Секреты не зашиты в код. Скопируйте .env.example в .env и заполните:

cp .env.example .env

Переменная	Описание
AIRFLOW_BASE_URL	URL Airflow
AIRFLOW_USERNAME / AIRFLOW_PASSWORD	доступ к Airflow REST API
GITLAB_URL	URL GitLab
GITLAB_TOKEN	Personal Access Token (read_api, write_repository)
GITLAB_PROJECT_PATH	путь проекта, по умолчанию stark/airflow
GITLAB_SOURCE_BRANCH / GITLAB_TARGET_BRANCH	ветки, по умолчанию prod → v3-prod
CHANNEL_MAP	опц. подмена ID, формат OLD:NEW,OLD2:NEW2

.env загружается автоматически. Можно также передавать переменные через окружение.

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

# Список активных DAG из Airflow
airflow-v3-migrate list-dags

# Мигрировать DAG по имени (путь к папке берётся из Airflow)
airflow-v3-migrate migrate-dag exchange_prices

# Мигрировать конкретную папку репозитория
airflow-v3-migrate migrate-folder dags/for_bison/moq

# Без интерактивного подтверждения
airflow-v3-migrate migrate-dag exchange_prices --yes

# Со своим сообщением коммита
airflow-v3-migrate migrate-dag exchange_prices -m "CDAT-2654 migrate exchange_prices"

# Применить правила к локальным файлам (без GitLab/Airflow)
airflow-v3-migrate fix-local ./dags
airflow-v3-migrate fix-local ./dags --diagnose   # только диагностика

Использование как класса (рекомендуется)

Все токены и доступы передаются один раз при инициализации, дальше — только методы. Полный пример: example.py.

from airflow_v3_migrator import AirflowV3Migrator

migrator = AirflowV3Migrator(
    airflow_base_url="https://airflow-cdat.autodoc.dev",
    airflow_username="admin",
    airflow_password="***",
    gitlab_url="https://gitlab.autodoc.dev",
    gitlab_token="***",                       # права: read_api, write_repository
    gitlab_project_path="stark/airflow",
    source_branch="prod",
    target_branch="v3-prod",
    channel_map={"PROD_CHANNEL_ID": "TEST_CHANNEL_ID"},   # опц. подмена ID
)

# либо собрать из .env / окружения:
# migrator = AirflowV3Migrator.from_env()

dags = migrator.get_active_dags()             # список активных DAG из Airflow
url = migrator.migrate_dag("exchange_prices") # возвращает ссылку на коммит

migrator.analyze_code(code)                   # список проблем в коде
migrator.migrate_code(code)                   # мигрированный код
migrator.migrate_dag("exchange_prices")       # миграция DAG по имени + коммит
migrator.migrate_folder("dags/for_bison/moq") # миграция папки + коммит

# сообщение коммита: при инициализации (commit_message=...),
# на лету в методе, или сменить атрибут для всех вызовов
migrator.migrate_dag("exchange_prices", commit_message="CDAT-2654 migrate")
migrator.commit_message = "CDAT-2654 auto dag move"
migrator.fix_local("./dags", diagnose=True)   # локально, без GitLab/Airflow

Методы класса

Метод	Что делает
get_active_dags(only_active=True)	список DAG из Airflow
find_dag(name)	найти DAG по имени
analyze_code(code)	список проблем Airflow 2.x
migrate_code(code)	вернуть мигрированный код
migrate_dag(name, confirm=True)	мигрировать DAG по имени и закоммитить
migrate_folder(folder, confirm=True)	мигрировать папку репозитория
fix_local(path, diagnose=False)	применить правила к локальным файлам

Перезапись существующей папки (update вместо create)

По умолчанию, если папка уже есть в целевой ветке, перенос отменяется. Чтобы вместо этого перезаписать файлы, передайте overwrite=True — тогда существующие файлы коммитятся с action=update, а новые с action=create (выбирается автоматически по каждому файлу):

migrator.migrate_dag("exchange_prices", overwrite=True)
migrator.migrate_folder("dags/for_bison/moq", overwrite=True)

airflow-v3-migrate migrate-dag exchange_prices --overwrite

migrate_dag / migrate_folder / push_commit возвращают ссылку на созданный коммит (web_url) или None, если коммитить нечего. Автор коммита по умолчанию — Adilkhan Zaidulla <[e-mail]> (меняется параметрами author_name / author_email или переменными GITLAB_AUTHOR_NAME / GITLAB_AUTHOR_EMAIL).

Логирование

Уровень логов управляется через setup_logging или параметр log_level:

from airflow_v3_migrator import setup_logging
import logging

setup_logging(logging.DEBUG)   # подробно: HTTP-запросы, diff каждого файла
setup_logging("WARNING")       # только предупреждения и ошибки

В CLI — флаг --log-level:

airflow-v3-migrate --log-level DEBUG migrate-dag exchange_prices

Публикация в PyPI

pip install build twine
python -m build                       # соберёт dist/*.whl и dist/*.tar.gz
twine check dist/*                     # проверка метаданных
twine upload dist/*                    # публикация (спросит токен PyPI)

Токен PyPI берётся из ~/.pypirc или переменной TWINE_PASSWORD (username __token__). Для проверки можно сначала загрузить в TestPyPI: twine upload -r testpypi dist/*.

⚠️ Имя airflow-v3-migrator на PyPI может быть занято — проверьте на pypi.org и при необходимости поменяйте name в pyproject.toml.

Безопасность

• Никогда не коммитьте .env и токены в репозиторий (.gitignore это учитывает).
• Если токен/пароль уже попадал в общий ноутбук — отзовите его и выпустите новый.