Migrating Airflow 2 To 3

Airflow 2 to 3 Migration

This skill helps migrate Airflow 2.x DAG code to Airflow 3.x, focusing on code changes (imports, operators, hooks, context, API usage).

Important: Before migrating to Airflow 3, strongly recommend upgrading to Airflow 2.11 first, then to at least Airflow 3.0.11 (ideally directly to 3.1). Other upgrade paths would make rollbacks impossible. See: https://www.astronomer.io/docs/astro/airflow3/upgrade-af3#upgrade-your-airflow-2-deployment-to-airflow-3. Additionally, early 3.0 versions have many bugs - 3.1 provides a much better experience.

Migration at a Glance

1. Run Ruff's Airflow migration rules to auto-fix detectable issues (AIR30/AIR301/AIR302/AIR31/AIR311/AIR312).

• ruff check --preview --select AIR --fix --unsafe-fixes .

2. Scan for remaining issues using the manual search checklist in reference/migration-checklist.md.

• Focus on: direct metadata DB access, legacy imports, scheduling/context keys, XCom pickling, datasets-to-assets, REST API/auth, plugins, and file paths.
• Hard behavior/config gotchas to explicitly review:
• Cron scheduling semantics: consider AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVAL=True if you need Airflow 2-style cron data intervals.
• .airflowignore syntax changed from regexp to glob; set AIRFLOW__CORE__DAG_IGNORE_FILE_SYNTAX=regexp if you must keep regexp behavior.
• OAuth callback URLs add an /auth/ prefix (e.g. /auth/oauth-authorized/google).
• Shared utility imports: Bare imports like import common from dags/common/ no longer work on Astro. Use fully qualified imports: import dags.common.

3. Plan changes per file and issue type:

• Fix imports - update operators/hooks/providers - refactor metadata access to using the Airflow client instead of direct access - fix use of outdated context variables - fix scheduling logic.

4. Implement changes incrementally, re-running Ruff and code searches after each major change. 5. Explain changes to the user and caution them to test any updated logic such as refactored metadata, scheduling logic and use of the Airflow context.

---

Architecture & Metadata DB Access

Airflow 3 changes how components talk to the metadata database:

• Workers no longer connect directly to the metadata DB.
• Task code runs via the Task Execution API exposed by the API server.
• The DAG processor runs as an independent process separate from the scheduler.
• The Triggerer uses the task execution mechanism via an in-process API server.

Trigger implementation gotcha: If a trigger calls hooks synchronously inside the asyncio event loop, it may fail or block. Prefer calling hooks via sync_to_async(...) (or otherwise ensure hook calls are async-safe).

Key code impact: Task code can still import ORM sessions/models, but any attempt to use them to talk to the metadata DB will fail with:

RuntimeError: Direct database access via the ORM is not allowed in Airflow 3.x

Patterns to search for

When scanning DAGs, custom operators, and @task functions, look for:

• Session helpers: provide_session, create_session, @provide_session
• Sessions from settings: from airflow.settings import Session
• Engine access: from airflow.settings import engine
• ORM usage with models: session.query(DagModel)..., session.query(DagRun)...

Replacement: Airflow Python client

Preferred for rich metadata access patterns. Add to requirements.txt:

apache-airflow-client==<your-airflow-runtime-version>

Example usage:

import os
from airflow.sdk import BaseOperator
import airflow_client.client
from airflow_client.client.api.dag_api import DAGApi

_HOST = os.getenv("AIRFLOW__API__BASE_URL", "https://<your-org>.astronomer.run/<deployment>/")
_TOKEN = os.getenv("DEPLOYMENT_API_TOKEN")

class ListDagsOperator(BaseOperator):
    def execute(self, context):
        config = airflow_client.client.Configuration(host=_HOST, access_token=_TOKEN)
        with airflow_client.client.ApiClient(config) as api_client:
            dag_api = DAGApi(api_client)
            dags = dag_api.get_dags(limit=10)
            self.log.info("Found %d DAGs", len(dags.dags))

Replacement: Direct REST API calls

For simple cases, call the REST API directly using requests:

from airflow.sdk import task
import os
import requests

_HOST = os.getenv("AIRFLOW__API__BASE_URL", "https://<your-org>.astronomer.run/<deployment>/")
_TOKEN = os.getenv("DEPLOYMENT_API_TOKEN")

@task
def list_dags_via_api() -> None:
    response = requests.get(
        f"{_HOST}/api/v2/dags",
        headers={"Accept": "application/json", "Authorization": f"Bearer {_TOKEN}"},
        params={"limit": 10}
    )
    response.raise_for_status()
    print(response.json())

---

Ruff Airflow Migration Rules

Use Ruff's Airflow rules to detect and fix many breaking changes automatically.

• AIR30 / AIR301 / AIR302: Removed code and imports in Airflow 3 - must be fixed.
• AIR31 / AIR311 / AIR312: Deprecated code and imports - still work but will be removed in future versions; should be fixed.

Commands to run (via uv) against the project root:

# Auto-fix all detectable Airflow issues (safe + unsafe)
ruff check --preview --select AIR --fix --unsafe-fixes .

# Check remaining Airflow issues without fixing
ruff check --preview --select AIR .

---

Reference Files

For detailed code examples and migration patterns, see:

• [reference/config-changes.md](reference/config-changes.md) - airflow.cfg section moves, renames, and removals
• [reference/migration-patterns.md](reference/migration-patterns.md) - Code examples for imports, scheduling, XCom, Assets, DAG bundles, runtime behavior changes
• [reference/removed-methods.md](reference/removed-methods.md) - Removed model methods with SDK/API migration paths
• [reference/migration-checklist.md](reference/migration-checklist.md) - Search patterns and fixes for issues Ruff doesn't catch

---

Quick Reference Tables

Key Import Changes

Context Key Changes

Asset-triggered runs: logical_date may be None; use context["dag_run"].logical_date defensively.

Cannot trigger with future `logical_date`: Use logical_date=None and rely on run_id instead.

Cron note: for scheduled runs using cron, logical_date semantics differ under CronTriggerTimetable (aligning logical_date with run_after). If you need Airflow 2-style cron data intervals, consider AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVAL=True.

Default Behavior Changes

Callback Behavior Changes

• on_success_callback no longer runs on skip; use on_skipped_callback if needed.
• @teardown with TriggerRule.ALWAYS not allowed; teardowns now execute even if DAG run terminated early.

---

Resources

• Astronomer Airflow 3 Upgrade Guide
• Airflow 3 Release Notes
• Ruff Airflow Rules

---

Related Skills

• testing-dags: For testing DAGs after migration
• debugging-dags: For troubleshooting migration issues
• deploying-airflow: For deploying migrated DAGs to production

Configuration File Changes (airflow.cfg)

Airflow 3 reorganized many configuration options across sections. These changes affect airflow.cfg, environment variables (AIRFLOW__SECTION__KEY), and Helm chart overrides.

---

Options Moved from [core] to [database]

• sql_alchemy_conn
• sql_engine_encoding
• sql_engine_collation_for_ids
• sql_alchemy_pool_enabled
• sql_alchemy_pool_size
• sql_alchemy_max_overflow
• sql_alchemy_pool_recycle
• sql_alchemy_pool_pre_ping
• sql_alchemy_schema
• sql_alchemy_connect_args
• load_default_connections
• max_db_retries

Example: AIRFLOW__CORE__SQL_ALCHEMY_CONN → AIRFLOW__DATABASE__SQL_ALCHEMY_CONN

---

Options Moved from [core] to [logging]

• base_log_folder
• remote_logging
• remote_log_conn_id
• remote_base_log_folder
• encrypt_s3_logs
• logging_level
• fab_logging_level
• logging_config_class
• colored_console_log
• colored_log_format
• colored_formatter_class
• log_format
• simple_log_format
• task_log_prefix_template
• log_filename_template
• log_processor_filename_template
• dag_processor_manager_log_location
• task_log_reader
• interleave_timestamp_parser

Example: AIRFLOW__CORE__REMOTE_LOGGING → AIRFLOW__LOGGING__REMOTE_LOGGING

---

Renamed Options

---

Removed Options

Migration Checklist

After running Ruff's AIR rules, use this manual search checklist to find remaining issues.

1. Direct metadata DB access

Search for:

• provide_session
• create_session
• @provide_session
• Session(
• engine
• with Session()
• engine.connect(
• Session(bind=engine)
• from airflow.settings import Session
• from airflow.settings import engine
• from sqlalchemy.orm.session import Session

Fix: Refactor to use Airflow Python client or REST API

---

2. Legacy imports

Search for:

• from airflow.contrib
• from airflow.operators.
• from airflow.hooks.

Fix: Map to provider imports (see migration-patterns.md)

---

3. Removed/renamed DAG arguments

Search for:

• schedule_interval=
• timetable=
• days_ago(
• fail_stop=
• concurrency= (on DAG constructor)
• sla=
• sla_miss_callback
• task_concurrency=

Fix:

• schedule_interval and timetable → use schedule=
• days_ago → use pendulum.today("UTC").add(days=-N)
• fail_stop → renamed to fail_fast
• concurrency (DAG) → renamed to max_active_tasks
• sla and sla_miss_callback → removed; use Astro Alerts or OSS Deadline Alerts (Airflow 3.1+ experimental)
• task_concurrency → renamed to max_active_tis_per_dag

Additional parameter removals

Search for:

• execution_date on TriggerDagRunOperator → removed; use logical_date or run_id

---

4. Deprecated context keys

Search for:

• execution_date
• prev_ds
• next_ds
• yesterday_ds
• tomorrow_ds
• templates_dict

Fix:

• execution_date → use context["dag_run"].logical_date
• tomorrow_ds / yesterday_ds → use ds with date math: macros.ds_add(ds, 1) / macros.ds_add(ds, -1)
• prev_ds / next_ds → use prev_start_date_success or timetable API
• templates_dict → use params via context["params"]

---

5. XCom pickling

Search for:

• ENABLE_XCOM_PICKLING
• .xcom_pull( without task_ids=

Fix: Use JSON-serializable data or custom backend

---

6. Datasets to Assets

Search for:

• airflow.datasets
• triggering_dataset_events
• DatasetOrTimeSchedule
• on_dataset_created
• on_dataset_changed
• outlet_events["
• inlet_events["

Fix: Switch to airflow.sdk.Asset, AssetOrTimeSchedule, on_asset_created/on_asset_changed. Use Asset(name=...) objects as keys in outlet_events/inlet_events (not strings)

---

7. Removed operators

Search for:

• SubDagOperator
• SimpleHttpOperator
• DagParam
• DummyOperator

Fix: Use TaskGroups, HttpOperator, Param, EmptyOperator

---

8. Email changes

Search for:

• airflow.operators.email.EmailOperator
• airflow.utils.email
• email= (task parameter for email on failure/retry)

Fix: Use SMTP provider (apache-airflow-providers-smtp). Replace legacy email behavior with SMTP-provider callbacks such as send_smtp_notification(...) or SmtpNotifier.

---

9. REST API v1

Search for:

• /api/v1
• auth=(
• execution_date (in API params)
• dataset_triggered or dataset_expression (in API responses/requests)
• schedule_interval (in API responses)
• /api/v1/roles, /api/v1/permissions, /api/v1/users

Fix: Update to /api/v2 with Bearer tokens. Replace execution_date params with logical_date. Dataset endpoints now under asset resources.

Endpoint renames:

Field renames in API responses:

---

10. File paths and shared utility imports

Search for:

• open("include/
• open("data/
• template_searchpath=
• relative paths
• import common or from common (bare imports from dags/common/ or similar)
• import utils or from utils (bare imports from dags/utils/ or similar)
• sys.path.append or sys.path.insert (custom path manipulation)

Fix:

• Use __file__ or AIRFLOW_HOME anchoring for file paths
• Note: triggers cannot be in DAG bundle; must be elsewhere on sys.path
• Shared utility imports: Bare imports like import common no longer work. Use fully qualified imports: import dags.common or from dags.common.utils import helper_function

---

11. FAB-based plugins

Search for:

• appbuilder_views
• appbuilder_menu_items
• flask_blueprints
• AirflowPlugin

Fix: Flask-AppBuilder removed from core. FAB plugins need manual migration to new system (React apps, FastAPI, listeners). Do not auto-migrate; recommend separate PR

---

12. Configuration file (airflow.cfg)

Search for:

• AIRFLOW__CORE__SQL_ALCHEMY (database settings moved to [database])
• AIRFLOW__CORE__REMOTE_LOGGING or AIRFLOW__CORE__BASE_LOG_FOLDER (logging settings moved to [logging])
• AIRFLOW__CORE__DAG_CONCURRENCY (renamed to max_active_tasks_per_dag)
• AIRFLOW__SCHEDULER__DEACTIVATE_STALE_DAGS_INTERVAL (renamed to parsing_cleanup_interval)
• AIRFLOW__WEBSERVER__BASE_URL (moved to [api]base_url)
• AIRFLOW__KUBERNETES__ (section replaced by [kubernetes_executor])

Fix: Update environment variables and config references to new section/key names. See config-changes.md for full mapping.

---

13. airflow_local_settings.py: rename policy → task_policy

---

14. Callback and behavior changes

Search for:

• on_success_callback
• @teardown
• templates_dict
• expanded_ti_count
• external_trigger
• test_mode
• trigger_rule="dummy" or TriggerRule.DUMMY
• trigger_rule="none_failed_or_skipped" or NONE_FAILED_OR_SKIPPED

Fix:

• on_success_callback no longer runs on skip; use on_skipped_callback if needed
• @teardown with trigger rule always not allowed; teardowns now execute even if DAG run terminated early
• templates_dict removed → use params via context["params"]
• expanded_ti_count removed → use REST API "Get Mapped Task Instances"
• dag_run.external_trigger removed → infer from dag_run.run_type
• test_mode removed; avoid relying on this flag
• dummy trigger rule removed → use always (or TriggerRule.ALWAYS)
• none_failed_or_skipped trigger rule removed → use none_failed_min_one_success (or TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)

Migration Patterns Reference

Detailed code examples for Airflow 2 to 3 migration.

Table of Contents

• Removed Modules & Import Reorganizations
• Task SDK & Param Usage
• SubDAGs, SLAs, and Removed Features
• Scheduling & Context Changes
• XCom Pickling Removal
• Datasets to Assets
• DAG Bundles & File Paths
• CLI Argument Changes
• Runtime Behavioral Changes

---

Removed Modules & Import Reorganizations

airflow.contrib.* removed

The entire airflow.contrib.* namespace is removed in Airflow 3.

Before (Airflow 2.x, removed in Airflow 3):

from airflow.contrib.operators.dummy_operator import DummyOperator

After (Airflow 3):

from airflow.providers.standard.operators.empty import EmptyOperator

Use EmptyOperator instead of the removed DummyOperator.

Core operators moved to provider packages

Many commonly used core operators moved to the standard provider.

Example for BashOperator and PythonOperator:

# Airflow 2 legacy imports (removed in Airflow 3, AIR30/AIR301)
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator

# Airflow 2/3 deprecated imports (still work but deprecated, AIR31/AIR311)
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

# Recommended in Airflow 3: Standard provider
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator

Operators moved to the apache-airflow-providers-standard package include (non-exhaustive):

• BashOperator
• BranchDateTimeOperator
• BranchDayOfWeekOperator
• LatestOnlyOperator
• PythonOperator
• PythonVirtualenvOperator
• ExternalPythonOperator
• BranchPythonOperator
• BranchPythonVirtualenvOperator
• BranchExternalPythonOperator
• ShortCircuitOperator
• TriggerDagRunOperator

This provider is installed on Astro Runtime by default.

Hook and sensor imports moved to providers

Most hooks and sensors live in provider packages in Airflow 3. Look for very old imports:

from airflow.hooks.http_hook import HttpHook
from airflow.hooks.base_hook import BaseHook

Replace with provider imports:

from airflow.providers.http.hooks.http import HttpHook
from airflow.sdk import BaseHook  # base hook from task SDK where appropriate

EmailOperator moved to SMTP provider

In Airflow 3, EmailOperator is provided by the SMTP provider, not the standard provider.

from airflow.providers.smtp.operators.smtp import EmailOperator

EmailOperator(
    task_id="send_email",
    conn_id="smtp_default",
    to="receiver@example.com",
    subject="Test Email",
    html_content="This is a test email",
)

Ensure apache-airflow-providers-smtp is added to any project that uses email features or notifications so that email-related code is compatible with Airflow 3.2 and later.

Replacing legacy email notifications: Move towards SMTP-provider based callbacks (and eventually SmtpNotifier) instead of relying on legacy task-level email behavior:

from airflow.providers.smtp.notifications.smtp import send_smtp_notification

BashOperator(
    task_id="my_task",
    bash_command="exit 1",
    on_failure_callback=[
        send_smtp_notification(
            from_email="airflow@my_domain.com",
            to="my_name@my_domain.ch",
            subject="[Error] The Task {{ ti.task_id }} failed",
            html_content="debug logs",
        )
    ],
)

Astro users: Consider Astro Alerts for critical notifications (works independently of Airflow components).

---

Task SDK & Param Usage

In Airflow 3, most classes and decorators used by DAG authors are available via the Task SDK (airflow.sdk). Using these imports makes it easier to evolve your code with future Airflow versions.

Key Task SDK imports

Prefer these imports in new code:

from airflow.sdk import (
    dag,
    task,
    setup,
    teardown,
    DAG,
    TaskGroup,
    BaseOperator,
    BaseSensorOperator,
    Param,
    ParamsDict,
    Variable,
    Connection,
    Context,
    Asset,
    AssetAlias,
    AssetAll,
    AssetAny,
    DagRunState,
    TaskInstanceState,
    TriggerRule,
    WeightRule,
    BaseHook,
    BaseNotifier,
    XComArg,
    chain,
    chain_linear,
    cross_downstream,
    get_current_context,
)

Import mappings from legacy to Task SDK

---

SubDAGs, SLAs, and Removed Features

SubDAGs removed

Search for:

• SubDagOperator(
• from airflow.operators.subdag_operator import SubDagOperator
• from airflow.operators.subdag import SubDagOperator

Migration guidance:

• Use TaskGroup or @task_group for logical grouping within a single DAG.
• For workflows that were previously split via SubDAGs, consider:
• Refactoring into smaller DAGs.
• Using Assets (formerly Datasets) for cross-DAG dependencies.

SLAs removed

Search for:

• sla=
• sla_miss_callback
• SLAMiss

Code changes:

• Remove SLA-related parameters from tasks and DAGs.
• Remove SLA-based callbacks from DAG definitions.
• On Astro, use Astro Alerts for DAG/task-level SLAs.

Other removed or renamed code features

• DagParam removed — use Param from airflow.sdk.
• SimpleHttpOperator removed - use HttpOperator from the HTTP provider.
• Trigger rules:
• dummy - use TriggerRule.ALWAYS.
• none_failed_or_skipped - use TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS.
• .xcom_pull behavior:
• In Airflow 3, calling xcom_pull(key="...") without task_ids always returns None; always specify task_ids explicitly.
• fail_stop DAG parameter renamed to fail_fast.
• max_active_tasks now limits active task instances per DAG run instead of across all DAG runs.
• on_success_callback no longer runs on skip; use on_skipped_callback if needed.
• @teardown with TriggerRule.ALWAYS not allowed; teardowns now execute even if DAG run terminated early.
• templates_dict removed - use params via context["params"].
• expanded_ti_count removed - use REST API "Get Mapped Task Instances" endpoint.
• dag_run.external_trigger removed - infer from dag_run.run_type.
• test_mode removed; avoid relying on this flag.
• Cannot trigger a DAG with a logical_date in the future; use logical_date=None and rely on run_id instead.

Executor removals

• SequentialExecutor removed — use LocalExecutor instead (can use SQLite for local dev).
• CeleryKubernetesExecutor removed — use Multiple Executor Configuration instead.
• LocalKubernetesExecutor removed — use Multiple Executor Configuration instead.
• Executor registration via plugins removed — treat executors as plain Python classes.

---

Scheduling & Context Changes

Default scheduling behavior

Airflow 3 changes default DAG scheduling:

• schedule=None instead of timedelta(days=1).
• catchup=False instead of True.

Code impact:

• If a DAG relied on implicit daily scheduling, explicitly set schedule.
• If a DAG relied on catchup by default, explicitly set catchup=True.

Removed context keys and replacements

Note: These replacements are not always drop-in; logic changes may be required.

Asset-triggered runs: logical_date may be None. Use defensive access: context["dag_run"].logical_date or context["run_id"].

days_ago removed

The helper days_ago from airflow.utils.dates is removed. Replace with explicit datetimes:

# WRONG - Removed in Airflow 3
from airflow.utils.dates import days_ago
start_date=days_ago(2)

# CORRECT - Use pendulum
import pendulum
start_date=pendulum.today("UTC").add(days=-2)

---

XCom Pickling Removal

In Airflow 3:

• AIRFLOW__CORE__ENABLE_XCOM_PICKLING is removed.
• The default XCom backend requires values to be serializable (for most users this means JSON-serializable values).

If tasks need to pass complex objects (e.g. NumPy arrays), you must use a custom XCom backend.

Example custom backend for NumPy arrays:

from airflow.sdk.bases.xcom import BaseXCom
import json
import numpy as np

class NumpyXComBackend(BaseXCom):
    @staticmethod
    def serialize_value(value, **kwargs):
        if isinstance(value, np.ndarray):
            return json.dumps({"type": "ndarray", "data": value.tolist(), "dtype": str(value.dtype)}).encode()
        return BaseXCom.serialize_value(value)

    @staticmethod
    def deserialize_value(result):
        if isinstance(result.value, bytes):
            d = json.loads(result.value.decode("utf-8"))
            if d.get("type") == "ndarray":
                return np.array(d["data"], dtype=d["dtype"])
        return BaseXCom.deserialize_value(result)

Reference: https://www.astronomer.io/docs/learn/custom-xcom-backend-strategies

---

Datasets to Assets

Datasets were renamed to Assets in Airflow 3; the old APIs are deprecated.

Mappings:

When working with asset events in the task context, do not use plain strings as keys in outlet_events or inlet_events:

# WRONG
outlet_events["myasset"]

# CORRECT
from airflow.sdk import Asset
outlet_events[Asset(name="myasset")]

Reading asset event data:

from airflow.sdk import task

@task
def read_triggering_assets(**context):
    events = context.get("triggering_asset_events") or {}
    for asset, asset_events in events.items():
        first_event = asset_events[0]
        print(asset, first_event.source_run_id)

Cosmos/dbt note: Asset URIs changed from dots to slashes (schema.table → schema/table). Upgrade astronomer-cosmos to >= 1.10.0 for Airflow 3 compatibility (and >= 1.11.0 if you need dbt Docs hosting in the Airflow UI).

---

DAG Bundles & File Paths

On Astro Runtime, Airflow 3 uses a versioned DAG bundle, so file paths and imports behave differently.

Shared utility imports

If you import shared utility code from dags/common/ or similar directories, bare imports no longer work in Airflow 3 on Astro. This is because DAG bundles place the bundle root on sys.path, but not <bundle_root>/dags. Additionally, bare imports are unsafe with DAG bundles due to Python's global import cache conflicting with concurrent bundle versions.

Use fully qualified imports instead:

# Airflow 2 (no longer works)
import common
from common.utils import helper_function

# Airflow 3
import dags.common
from dags.common.utils import helper_function

Each bundle has its own dags package rooted at its bundle directory, which keeps imports scoped to the correct bundle version.

File path handling

On Astro Runtime, Airflow 3 uses a versioned DAG bundle, so file paths behave differently:

For files inside `dags/` folder:

import os
dag_dir = os.path.dirname(__file__)
with open(os.path.join(dag_dir, "my_file.txt"), "r") as f:
    contents = f.read()

For files in `include/` or other mounted folders:

import os
with open(f"{os.getenv('AIRFLOW_HOME')}/include/my_file.txt", 'r') as f:
    contents = f.read()

For `template_searchpath`:

import os
from airflow.sdk import dag

@dag(template_searchpath=[f"{os.getenv('AIRFLOW_HOME')}/include/sql"])
def my_dag():
    ...

Note: Triggers cannot be in the DAG bundle; they must be elsewhere on sys.path.

---

CLI Argument Changes

Several Airflow CLI arguments were renamed or removed in Airflow 3. Update any scripts, CI pipelines, or documentation that invoke these commands.

---

Runtime Behavioral Changes

These changes affect DAG execution behavior without changing imports or API signatures. They can cause silent bugs if not addressed.

Connection validation

Connection.extra must now be valid JSON. Airflow 3 enforces this at save time. Connections with non-JSON extra fields will fail validation.

DAG tags type change

DAG.tags changed from list to MutableSet. This means:

• Duplicate tags are silently removed
• Tag ordering is not guaranteed
• Code that relies on tags[0] or list-specific operations will break

Param serialization

All Param values must be JSON serializable. Parsed date/time Param values are now RFC 3339 compliant. Non-serializable default values will raise errors at DAG parse time.

Dataset/Asset hashability

Dataset (now Asset) and DatasetAlias (now AssetAlias) are no longer hashable. Code that uses them as dictionary keys or in sets will raise TypeError. Additionally, Dataset equality now considers the extra dict.

Cron scheduling semantics

Cron schedules now default to CronTriggerTimetable instead of CronDataIntervalTimetable. Under the new timetable, logical_date equals run_after (not data_interval_start). Set AIRFLOW__CORE__CREATE_CRON_DATA_INTERVALS=True to revert to Airflow 2 behavior.

logical_date can be None

For asset-triggered or manually-triggered DAG runs, logical_date can be None. Code that assumes logical_date is always a datetime will raise AttributeError. Use defensive access patterns.

XCom pickling disabled

XCom pickling is disabled by default for security. The default XCom backend requires JSON-serializable values. Use a custom XCom backend for complex objects.

Removed & Inaccessible Model Methods

Airflow 3 enforces a strict boundary between the public SDK (airflow.sdk) used in DAG code and internal ORM models (airflow.models.*) used by the scheduler and API. Task code runs in isolated subprocesses with no database access (AIP-72).

If your DAG code calls any of the methods below, the fix is not just a rename — you need to replace direct model access with SDK patterns or the Airflow Client API.

---

DAG

In Airflow 3, use from airflow.sdk import DAG. The SDK DAG is a definition object — it does not expose runtime state methods.

---

TaskInstance

In task code, access task instance state through the context dict or SDK IPC methods. Direct ORM queries on TaskInstance are blocked at runtime.

---

DagRun

Access DagRun properties through the task context. Direct ORM queries on DagRun are blocked at runtime.

---

Connection

Connection resolution happens through hooks. Do not query Connection objects directly.

---

Dataset → Asset Renames