Add Airflow 3.3 to CI test matrix

.github/workflows/cicd.yaml

@@ -63,7 +63,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
-        airflow-version: ["2.9", "2.10", "2.11", "3.0", "3.1", "3.2"]
+        airflow-version: ["2.9", "2.10", "2.11", "3.0", "3.1", "3.2", "3.3"]
         # Third matrix axis: optional pin for apache-airflow-providers-standard.
         #
         # Purpose: regression coverage for issue #679. On Airflow 2.x with
@@ -166,7 +166,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
-        airflow-version: ["2.9", "2.10", "2.11", "3.0", "3.1", "3.2"]
+        airflow-version: ["2.9", "2.10", "2.11", "3.0", "3.1", "3.2", "3.3"]
         exclude:
           - python-version: "3.13"
             airflow-version: "2.9"

dev/tests/dags/test_dag_example.py

@@ -1,12 +1,23 @@
 """Example DAGs test. This test ensures that all Dags have tags, retries set to two, and no import errors. This is an example pytest and may not be fit the context of your DAGs. Feel free to add and remove tests."""
 
+import inspect
 import logging
 import os
 from contextlib import contextmanager
 
 import pytest
 from airflow.models import DagBag
 
+# Airflow 3.3 dropped the `include_examples` kwarg from DagBag.__init__ (examples
+# are no longer loaded implicitly), so only pass it on versions that still support it.
+_DAGBAG_SUPPORTS_INCLUDE_EXAMPLES = "include_examples" in inspect.signature(DagBag.__init__).parameters
+
+
+def _new_dag_bag():
+    if _DAGBAG_SUPPORTS_INCLUDE_EXAMPLES:
+        return DagBag(include_examples=False)
+    return DagBag()
+
 
 @contextmanager
 def suppress_logging(namespace):
@@ -24,7 +35,7 @@ def get_import_errors():
     Generate a tuple for import errors in the dag bag
     """
     with suppress_logging("airflow"):
-        dag_bag = DagBag(include_examples=False)
+        dag_bag = _new_dag_bag()
 
         def strip_path_prefix(path):
             return os.path.relpath(path, os.environ.get("AIRFLOW_HOME"))
@@ -38,7 +49,7 @@ def get_dags():
     Generate a tuple of dag_id, <DAG objects> in the DagBag
     """
     with suppress_logging("airflow"):
-        dag_bag = DagBag(include_examples=False)
+        dag_bag = _new_dag_bag()
 
     def strip_path_prefix(path):
         return os.path.relpath(path, os.environ.get("AIRFLOW_HOME"))

pyproject.toml

@@ -113,15 +113,15 @@ pre-install-commands = ["sh scripts/test/pre-install-airflow.sh {matrix:airflow}
 
 [[tool.hatch.envs.tests.matrix]]
 python = ["3.10", "3.11", "3.12"]
-airflow = ["2.9", "2.10", "2.11", "3.0", "3.1", "3.2"]
+airflow = ["2.9", "2.10", "2.11", "3.0", "3.1", "3.2", "3.3"]
 
 [[tool.hatch.envs.tests.matrix]]
 python = ["3.13"]
-airflow = ["3.1", "3.2"]
+airflow = ["3.1", "3.2", "3.3"]
 
 [[tool.hatch.envs.tests.matrix]]
 python = ["3.14"]
-airflow = ["3.2"]
+airflow = ["3.2", "3.3"]
 
 # Regression coverage for issue #679: on Airflow 2.x with apache-airflow-providers-standard
 # installed, the core and providers.standard PythonOperator classes diverge.

scripts/test/pre-install-airflow.sh

@@ -39,6 +39,8 @@ elif [ "$AIRFLOW_VERSION" = "3.1" ]; then
   uv pip install "apache-airflow~=3.1.0" "apache-airflow-providers-http>=4.0.0"  "apache-airflow-providers-cncf-kubernetes>=4.4.0" "apache-airflow-providers-common-sql>=1.2.0" "apache-airflow-providers-slack" --constraint /tmp/constraint.txt
 elif [ "$AIRFLOW_VERSION" = "3.2" ]; then
   uv pip install "apache-airflow~=3.2.0" "apache-airflow-providers-http>=4.0.0"  "apache-airflow-providers-cncf-kubernetes>=4.4.0" "apache-airflow-providers-common-sql>=1.2.0" "apache-airflow-providers-slack" --constraint /tmp/constraint.txt
+elif [ "$AIRFLOW_VERSION" = "3.3" ]; then
+  uv pip install "apache-airflow==3.3.0" "apache-airflow-providers-http>=4.0.0"  "apache-airflow-providers-cncf-kubernetes>=4.4.0" "apache-airflow-providers-common-sql>=1.2.0" "apache-airflow-providers-slack" --constraint /tmp/constraint.txt
 else
   uv pip install "apache-airflow==$AIRFLOW_VERSION" "apache-airflow-providers-http>=4.0.0" "apache-airflow-providers-cncf-kubernetes>=4.4.0" "apache-airflow-providers-common-sql>=1.2.0" "apache-airflow-providers-slack" --constraint /tmp/constraint.txt
 fi;

tests/test__main__.py

@@ -46,7 +46,14 @@ def test_help_output_when_no_command():
     result = runner.invoke(app, [])
     assert result.exit_code == 0
     assert "DAG Factory" in result.output
-    assert "dagfactory [OPTIONS] COMMAND [ARGS]" in result.output
+    # typer's exact "Usage:" formatting (brackets/ellipsis around COMMAND/ARGS, and whether
+    # ANSI color codes split "Usage:" from "dagfactory") varies by version and environment,
+    # so just check the pieces that are stable across both.
+    assert "Usage" in result.output
+    assert "dagfactory" in result.output
+    assert "OPTIONS" in result.output
+    assert "COMMAND" in result.output
+    assert "ARGS" in result.output
 
 
 def test_help_option():

tests/test_example_dags.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import inspect
 from functools import cache
 from pathlib import Path
 
@@ -15,6 +16,9 @@
 EXAMPLE_DAGS_DIR = Path(__file__).parent.parent / "dev/dags"
 AIRFLOW_IGNORE_FILE = EXAMPLE_DAGS_DIR / ".airflowignore"
 AIRFLOW_VERSION = Version(airflow.__version__)
+# Airflow 3.3 dropped the `include_examples` kwarg from DagBag.__init__ (examples
+# are no longer loaded implicitly), so only pass it on versions that still support it.
+_DAGBAG_SUPPORTS_INCLUDE_EXAMPLES = "include_examples" in inspect.signature(DagBag.__init__).parameters
 # TODO: Enable asset_triggered_dags.py once https://github.com/apache/airflow/issues/51644 is solved
 IGNORED_DAG_FILES = ["example_callbacks.py", "example_http_operator_task.py", "asset_triggered_dags.py", "kpo.py"]
 
@@ -27,7 +31,7 @@
 
 @provide_session
 def get_session(session=None):
-    create_default_connections(session)
+    create_default_connections(session=session)
     return session
 
 
@@ -68,7 +72,10 @@ def get_dag_bag() -> DagBag:
     # Print the contents of the .airflowignore file, and build the DagBag
     print(".airflowignore contents: ")
     print(AIRFLOW_IGNORE_FILE.read_text())
-    db = DagBag(EXAMPLE_DAGS_DIR, include_examples=False)
+    if _DAGBAG_SUPPORTS_INCLUDE_EXAMPLES:
+        db = DagBag(EXAMPLE_DAGS_DIR, include_examples=False)
+    else:
+        db = DagBag(EXAMPLE_DAGS_DIR)
 
     assert db.dags
     assert not db.import_errors