概要

• 実際に Airflow に読ませるのは環境用意したりで大変
• まずはしょーもないミスが無いか手元でサクッと確認したいだけなんだ…

A simple way to Validate your Airflow DAGs on Pull Request

DagBag にロードしてみてエラーを確認するシンプルな仕組みのよう。

DagBag

airflow.models.dagbag - Airflow Documentation

準備

test/dag_validation.py

import os
import unittest
from airflow.models import DagBag

class TestDags(unittest.TestCase):
    def setUp(self):
        DAGS_DIR = os.getenv('INPUT_DAGPATHS', 'dags_')
        self.dagbag = DagBag(dag_folder = DAGS_DIR, include_examples = False)

    def test_import_dags(self):
        self.assertFalse(
            len(self.dagbag.import_errors),
            f"DAG import failures. Errors: {self.dagbag.import_errors}"
        )

suite = unittest.TestLoader().loadTestsFromTestCase(TestDags)

requirements.txt

pytest
apache-airflow

ローカルでテスト

$ pip install -r test/requirements.txt
$ INPUT_DAGPATHS=dags  pytest test/dag_validation.py -s -q

Github Actions

.github/workflows/validate.yml

name: dags_validate

on:
  push:
  workflow_dispatch:

permissions:
  contents: 'read'
  id-token: 'write'

jobs:
  test:
    runs-on: ubuntu-latest
    timeout-minutes: 5
    steps:
      - uses: actions/checkout@v3

      - uses: actions/setup-python@v4
        with:
          python-version: '3.10'
          cache: pip

      - name: install pip packages
        run: |
          pip install -r test/requirements.txt

      - name: "validate dags"
        run: INPUT_DAGPATHS=dags_  pytest test/dag_validation.py -s -q