Migrate API endpoints from sync to async DB sessions

[Dev-iL]

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
• GET /xcoms/{dag_id}/{run_id}/{task_id}/{key:path}/slice — get_mapped_xcom_by_slice
• HEAD /xcoms/{dag_id}/{run_id}/{task_id}/{key:path} — head_xcom
• GET /xcoms/{dag_id}/{run_id}/{task_id}/{key:path} — get_xcom
• POST /xcoms/{dag_id}/{run_id}/{task_id}/{key:path} — set_xcom
• DELETE /xcoms/{dag_id}/{run_id}/{task_id}/{key:path} — delete_xcom

Asset state store — routes/asset_state_store.py

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

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
• PUT /store/ti/{task_instance_id}/{key:path} — set_task_state_store
• DELETE /store/ti/{task_instance_id}/{key:path} — delete_task_state_store
• DELETE /store/ti/{task_instance_id} — clear_task_state_store

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

[Dev-iL]

cc @hussein-awala @eladkal

[maheshsv]

can i take this issue

[Dev-iL]

@maheshsv I suggest waiting until #67800 is merged, then using its final form as an example to convert other endpoints.

[zach-overflow]

@Dev-iL I was looking into the call pathways for potentially migrating /task-instances/states (get_task_instance_states), but I'm curious about this point in the SKILL.md in https://github.com/apache/airflow/pull/73405/changes :

Do not pass an AsyncSession to @provide_session or synchronous model methods.
AsyncSession.run_sync adapts SQLAlchemy work through a greenlet; it does not make
arbitrary network, filesystem, or secrets-backend calls nonblocking.

Is the intent here to reduce the scope of alterations for async onboarding? Or are there other reasons to use AsyncSession.run_sync instead of defining an async version of the @provide_session-decorated methods? I found it was not difficult to define a @provide_session_async decorator, and modify / add coroutines as needed without having to do any sync thread offloading calls.

If there is a need for the @provide_session_async decorator I can contribute that as a standalone PR. Let me know if I'm misunderstanding the constraints though.

[Dev-iL]

@zach-overflow Thank you for your willingness to help!

Is the intent here to reduce the scope of alterations for async onboarding? Or are there other reasons to use AsyncSession.run_sync instead of defining an async version of the @provide_session-decorated methods?

It isn't meant to limit scope. The line guards against two specific traps:

• Passing an AsyncSession to a sync @provide_session method doesn't fail at the call site. A body that calls session.scalar(...) gets back an un-awaited coroutine. Bodies that chain .scalars(...).all() or use .query fail with AttributeError instead.
• run_sync is fine for a unit that is only SQL: the event loop stays free while the database works, same as with a native await (checked this on all db backends). It doesn't help if the unit also makes a network, filesystem or secrets-backend call, since that still blocks the loop.

The skill doesn't prefer run_sync; native async is the first choice. You're right that it doesn't say async twins are welcome, or that run_sync is OK for pure-SQL units. I'll clarify that in #73405.

I found it was not difficult to define a @provide_session_async decorator, and modify / add coroutines as needed without having to do any sync thread offloading calls.

If there is a need for the @provide_session_async decorator I can contribute that as a standalone PR. Let me know if I'm misunderstanding the constraints though.

I asked the AI to test this approach. TL;DR: it agrees that the decorator is easy, but /task-instances/states doesn't need it, so the recommendation is to wait with it until something does.

Full analysis report by Claude

I wrote a ~13-line prototype on top of find_session_idx. It passes the session through when given and creates one when omitted, so it works as you described.

For /task-instances/states, every call on the path already passes session. Only the task_group_id branch goes beyond the route: _get_group_tasks → get_latest_version_of_dag → DBDagBag.get_latest_version_of_dag → SerializedDagModel.get. latest_item_select_object is already a plain select and the data columns aren't deferred, so the async version can be:

serdag = await session.scalar(SerializedDagModel.latest_item_select_object(dag_id))
dag = dag_bag._read_dag(serdag)  # existing sync, session-free

Against a seeded 500-task Dag on SQLite this returns the same Dag as the sync path, with no MissingGreenlet. No twin, decorator or run_sync is needed.

One open point: _read_dag deserializes the whole Dag on every call and never hits the cache on this path. A 500-task Dag took ~32ms, which would now run on the event loop instead of a threadpool worker. Probably fine, but worth a decision (offload or accept) in the PR.

My one worry about the decorator: if a caller forgets session, it silently opens a second connection. With a pool of one, I reproduced a route timing out waiting on its own pool. Routes should pass the AsyncSessionDep session down; dag.clear(...) in clear_dag_run currently doesn't. The legitimate users are entry points with no caller-owned session, like secrets backends. My async-secrets WIP hand-rolls if session is None: async with create_session_async() twice.

So I'd convert /states without the decorator. If you want to propose it, I'd tie it to its first consumers (async secrets) so reviewers can see it in use, and get a committer's view first, since create_session_async is :meta private: today.

[zach-overflow]

Sounds good and thanks for the clarification. I’ll take on the /task-instances/(states|count) routes since those are effectively calling much of the same code, without the decorator helper addition.

[Prab-27]

I'm working on Asset state store — routes/asset_state_store.py

[Dev-iL]

@Prab-27 Please make sure to link back to this issue once the PR is open