Airflow 3 tasks can't use the metadata database any more - replacing session queries with the REST API client

#What changed

From the Airflow 3 upgrade guide:

"In Airflow 3, operators cannot access the Airflow metadata database directly using database sessions."

Code that did this in Airflow 2 (for example with create_session() as session: session.query(DagRun)...) fails in Airflow 3 at runtime. Typical uses:

• Checking the state of another DAG's latest run
• Counting failed task instances for a report
• Clearing or marking tasks from a "maintenance" DAG
• Reading variables or connections by querying tables
• Cleanup DAGs that delete old rows

#1. Find it

grep -rnE "create_session|provide_session|settings\.Session|from airflow\.models import (DagRun|TaskInstance)|session\.query" dags/ plugins/

ruff's AIR rules flag many deprecated imports too:

ruff check dags/ plugins/ --select AIR301,AIR302

#2. Pick the replacement

#3. Use the REST API client

pip install apache-airflow-client

Get a token from the API server's /auth/token endpoint (the exact flow depends on your auth manager; with the default Simple Auth Manager it's a POST with username and password), store it as a secret, and use it in the task:

from airflow.sdk import task, Variable
import airflow_client.client as client
from airflow_client.client.api import dag_run_api

@task
def latest_run_state(dag_id: str) -> str:
    cfg = client.Configuration(host=Variable.get("airflow_api_host"))   # e.g. http://airflow-api-server:8080
    cfg.access_token = Variable.get("airflow_api_token")
    with client.ApiClient(cfg) as api:
        runs = dag_run_api.DagRunApi(api).get_dag_runs(
            dag_id=dag_id, order_by="-logical_date", limit=1)
        return runs.dag_runs[0].state if runs.dag_runs else "none"

Method and parameter names follow the generated client for /api/v2; check the client's README for your installed version.

Give the API identity only the permissions it needs (read DAG runs, or clear tasks for specific DAGs), not admin.

#4. The workaround to avoid

The upgrade guide mentions that PostgresHook, MySqlHook or SQLExecuteQueryOperator can still query the metadata DB as an ordinary database connection, and warns that this will break in Airflow 3.2+ because the schema isn't a public API. If you must do it temporarily:

• Use a read-only database user.
• Keep the queries in one place so they're easy to remove.
• Put a removal date on it.

Since current Airflow is 3.3, treat this as unsupported.

#5. Test

Run the rewritten tasks in staging with airflow dags test <dag_id> and confirm they work with the API identity's permissions, and fail clearly (not silently) when the token expires.

#Checklist

• All session/ORM use in DAGs and plugins found
• Each use mapped to context, Variable/Connection, XCom, assets or the REST API
• REST API identity with least privilege and a stored token
• Token rotation planned
• No direct metadata DB queries left (or dated, read-only exceptions)

#Sources

• Airflow: Upgrading to Airflow 3 - Restricted Metadata Database Access