Migrate API endpoints from sync to async DB sessions #67799

Description

Incrementally migrate Airflow's FastAPI endpoints to async database I/O. Each conversion should let database waits yield the event loop while preserving the endpoint's behavior. This is a tracking issue; individual endpoints or small cohesive groups should land in separate PRs.

The async engine, session factory, AsyncSessionDep, create_session_async, and async pagination helpers already exist. The work is to adopt them throughout each request path, including dependencies and domain helpers. Changing a route to async def while it still calls blocking helpers directly would make matters worse.

Motivation and limits

Synchronous FastAPI routes occupy a threadpool worker during blocking database calls. Native async database calls allow other requests to progress while those calls wait. That is the intended scheduling improvement; a production throughput increase must be measured separately.

#36504 explored async database access in the Airflow 2 architecture. Its compatibility and driver-testing concerns remain useful. In Airflow 3, worker and trigger execution code should use the Task SDK/Execution API boundary, rather than gaining direct metadata database access. SDK async connection lookup and async Execution API secrets transport already exist; server-side secret resolution still needs work.

The earlier experiments are exploratory evidence, not a production performance guarantee. High-concurrency results were inconclusive. Keep benchmark implementation and reports separate from endpoint migration PRs, and do not infer a CPU, GIL, or pool bottleneck from throughput and active-connection counts alone.

Definition of done per endpoint

For the contributors' convenience, the migration steps are encoded as an agent SKILL in #73405

1. Trace the complete request path: route, auth/team dependencies, domain helpers, external I/O, and serialization. Use AsyncSessionDep and await database operations. Handle remaining blocking I/O explicitly; document any thread-offloaded compatibility bridge.
2. Preserve the HTTP contract, permissions, team scope, query/count/order semantics, locking, rowcount branches, and transaction behavior. Let the session owner commit/roll back; do not add commits to route handlers. A pure implementation conversion does not require an Execution API version bump.
3. Preserve existing contract assertions and run supported API-version tests. Update coroutine mocks and fixture setup as needed. Add focused regressions for risks introduced by the conversion, including relevant failure paths; do not require test source to remain byte-identical.
4. Match async-engine lifetime to test event-loop lifetime. Reuse an appropriate shared fixture where available. The heartbeat's engine-reconfiguration fixture is a local workaround, not a rule to replicate unconditionally. Broader test-loop changes belong in separate work, as raised in the review discussion.
5. Exercise database-dependent behavior on SQLite, PostgreSQL, and MySQL using current drivers. Record exact backend/driver and results. SQLite does not verify row locks or all production-driver semantics. Where SQL, driver, or connection configuration changes, also validate affected overrides and PgBouncer configurations.
6. Pass applicable formatting, lint, typing, and static checks. Keep unrelated endpoints, synthetic routers, and benchmark infrastructure out of the conversion.

Cross-cutting work

• Server-side async secrets resolution: support native async metastore reads, compatible offloading for existing synchronous custom/provider backends, then migrate the Variable and Connection value-read endpoints. Preserve configured backend order, legacy overrides, team propagation, cache hits/misses, and authoritative access denials. Native async cloud clients can follow independently; every provider need not migrate before core support lands. Coordinate with #72329, which proposes SDK async Variable methods.
• Test engine lifecycle: evaluate a reusable engine/loop strategy as migration grows. Do not mask lifecycle failures or change global loop scope merely to make one route pass.
• Deployment configuration: verify actual sync and async pool settings and their multiplication across workers/replicas. Budget configured overflow and other DB clients as well as steady pool sizes. Avoid a universal connection-count prescription based on the POC.
• Performance follow-up: if benchmarking adoption, use a controlled workload, record client/server resources and driver/pool settings, and profile unexplained results. Performance experiments are not prerequisites to parity-focused conversions.

Execution API migration inventory

Paths below are relative to /execution. Checked means the conversion is merged, not merely present in a local branch. Snapshot checked against 45bad9f54d plus the local variable-keys WIP; recheck current code and open PRs before claiming a route. These are candidates, not promises that every handler is a mechanical session swap. Core API endpoints need their own call-path inventory when selected.

The inventory accounts for all 57 production route declarations in this checkout: 55 are listed below, including the older-version Dag-run route. The two health routes are excluded from migration: GET /health returns a static response without database access, and GET /health/ping already awaits service pings. The six experimental benchmark routes are excluded as development-only code.

Task instances — routes/task_instances.py

• PATCH /task-instances/{task_instance_id}/run — ti_run
• PATCH /task-instances/{task_instance_id}/state — ti_update_state
• PATCH /task-instances/{task_instance_id}/skip-downstream — ti_skip_downstream
• PUT /task-instances/{task_instance_id}/heartbeat — ti_heartbeat (#67800)
• PUT /task-instances/{task_instance_id}/rtif — ti_put_rtif
• PATCH /task-instances/{task_instance_id}/rendered-map-index — ti_patch_rendered_map_index
• GET /task-instances/{task_instance_id}/previous-successful-dagrun — get_previous_successful_dagrun (#73403)
• GET /task-instances/count — get_task_instance_count (#73966)
• GET /task-instances/previous/{dag_id}/{task_id} — get_previous_task_instance (#73403)
• GET /task-instances/states — get_task_instance_states (#73966)
• GET /task-instances/breadcrumbs — get_task_instance_breadcrumbs (#73403)
• GET /task-instances/{task_instance_id}/validate-inlets-and-outlets — validate_inlets_and_outlets

Dag runs — routes/dag_runs.py

• GET /dag-runs/{dag_id}/previous — get_previous_dagrun_compat (older API versions)
• GET /dag-runs/{dag_id}/{run_id} — get_dag_run
• POST /dag-runs/{dag_id}/{run_id} — trigger_dag_run
• POST /dag-runs/{dag_id}/{run_id}/clear — clear_dag_run
• GET /dag-runs/{dag_id}/{run_id}/state — get_dagrun_state (#73403)
• GET /dag-runs/count — get_dr_count (#73403)
• GET /dag-runs/previous — get_previous_dagrun

XComs — routes/xcoms.py

• GET /xcoms/{dag_id}/{run_id}/{task_id}/{key:path}/item/{offset} — get_mapped_xcom_by_index (#75396)
• GET /xcoms/{dag_id}/{run_id}/{task_id}/{key:path}/slice — get_mapped_xcom_by_slice (#75396)
• HEAD /xcoms/{dag_id}/{run_id}/{task_id}/{key:path} — head_xcom (#75396)
• GET /xcoms/{dag_id}/{run_id}/{task_id}/{key:path} — get_xcom (#75396)
• POST /xcoms/{dag_id}/{run_id}/{task_id}/{key:path} — set_xcom (#75396)
• DELETE /xcoms/{dag_id}/{run_id}/{task_id}/{key:path} — delete_xcom (#75396)

Asset state store — routes/asset_state_store.py

• GET /store/asset/by-name/value — get_asset_state_store_by_name (#74540)
• PUT /store/asset/by-name/value — set_asset_state_store_by_name (#74540)
• DELETE /store/asset/by-name/value — delete_asset_state_store_by_name (#74540)
• DELETE /store/asset/by-name/clear — clear_asset_state_store_by_name (#74540)
• GET /store/asset/by-uri/value — get_asset_state_store_by_uri (#74540)
• PUT /store/asset/by-uri/value — set_asset_state_store_by_uri (#74540)
• DELETE /store/asset/by-uri/value — delete_asset_state_store_by_uri (#74540)
• DELETE /store/asset/by-uri/clear — clear_asset_state_store_by_uri (#74540)

Assets and asset events

• GET /asset-events/by-asset — get_asset_event_by_asset_name_uri
• GET /asset-events/by-asset-alias — get_asset_event_by_asset_alias
• GET /assets/by-name — get_asset_by_name (#73403)
• GET /assets/by-uri — get_asset_by_uri (#73403)
• GET /assets/by-alias — get_assets_by_alias

Task state store — routes/task_state_store.py

• GET /store/ti/{task_instance_id}/{key:path} — get_task_state_store (#75381)
• PUT /store/ti/{task_instance_id}/{key:path} — set_task_state_store (#75381)
• DELETE /store/ti/{task_instance_id}/{key:path} — delete_task_state_store (#75381)
• DELETE /store/ti/{task_instance_id} — clear_task_state_store (#75381)

Human-in-the-loop, Dags, reschedules, and variables

• POST /hitlDetails/{task_instance_id} — upsert_hitl_detail
• PATCH /hitlDetails/{task_instance_id} — update_hitl_detail
• GET /hitlDetails/{task_instance_id} — get_hitl_detail (#73403)
• GET /dags/{dag_id} — get_dag
• GET /task-reschedules/{task_instance_id}/start_date — get_start_date (#73403)
• GET /variables/keys — get_variable_keys (WIP, not merged) (#73407)

Connection tests and callbacks

• GET /connection-tests/{connection_test_id}/connection — get_connection_test_connection (locks and changes state despite using GET)
• PATCH /connection-tests/{connection_test_id} — patch_connection_test
• PATCH /callbacks/{callback_id}/run — run_callback (single-use token exchange)

Value resolution and Variable writes

These use model/domain helpers rather than a route-level session swap:

• GET /variables/{variable_key:path} — async secrets resolution, then get_variable
• GET /connections/{connection_id} — async secrets resolution, then get_connection
• PUT /variables/{variable_key:path} — put_variable: secrets conflict checks, database writes, and cache invalidation require a separate transaction-aware conversion
• DELETE /variables/{variable_key:path} — delete_variable: database deletion and cache invalidation; does not require external async secrets lookup

References

• Final heartbeat conversion and discussions
• Historical async SQLAlchemy proposal and discussions
• Execution API source

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct