Replace DagBag from global app state to with FastAPI dependency

airflow-core/src/airflow/api_fastapi/app.py

@@ -31,7 +31,6 @@
     init_middlewares,
     init_views,
 )
-from airflow.api_fastapi.core_api.init_dagbag import get_dag_bag
 from airflow.api_fastapi.execution_api.app import create_task_execution_api_app
 from airflow.configuration import conf
 from airflow.exceptions import AirflowConfigException
@@ -80,16 +79,12 @@ def create_app(apps: str = "all") -> FastAPI:
         version="2",
     )
 
-    dag_bag = get_dag_bag()
-
     if "execution" in apps_list or "all" in apps_list:
         task_exec_api_app = create_task_execution_api_app()
-        task_exec_api_app.state.dag_bag = dag_bag
         init_error_handlers(task_exec_api_app)
         app.mount("/execution", task_exec_api_app)
 
     if "core" in apps_list or "all" in apps_list:
-        app.state.dag_bag = dag_bag
         init_plugins(app)
         init_auth_manager(app)
         init_flask_plugins(app)

airflow-core/src/airflow/api_fastapi/core_api/init_dagbag.py → airflow-core/src/airflow/api_fastapi/common/deps.py

@@ -14,16 +14,22 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+
 from __future__ import annotations
 
 import os
+from typing import Annotated
+
+from fastapi import Depends
 
-from airflow.models import DagBag
+from airflow.models.dagbag import DagBag
 from airflow.settings import DAGS_FOLDER
 
 
-def get_dag_bag() -> DagBag:
-    """Instantiate the appropriate DagBag based on the ``SKIP_DAGS_PARSING`` environment variable."""
+def _get_dag_bag() -> DagBag:
     if os.environ.get("SKIP_DAGS_PARSING") == "True":
         return DagBag(os.devnull, include_examples=False)
     return DagBag(DAGS_FOLDER, read_dags_from_db=True)
+
+
+DagBagDep = Annotated[DagBag, Depends(_get_dag_bag)]

airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py

@@ -20,11 +20,12 @@
 from datetime import datetime
 from typing import TYPE_CHECKING, Annotated
 
-from fastapi import Depends, HTTPException, Request, status
+from fastapi import Depends, HTTPException, status
 from sqlalchemy import and_, delete, func, select
 from sqlalchemy.orm import joinedload, subqueryload
 
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.deps import DagBagDep
 from airflow.api_fastapi.common.parameters import (
     BaseParam,
     FilterParam,
@@ -345,7 +346,7 @@ def create_asset_event(
 )
 def materialize_asset(
     asset_id: int,
-    request: Request,
+    dag_bag: DagBagDep,
     session: SessionDep,
 ) -> DAGRunResponse:
     """Materialize an asset by triggering a DAG run that produces it."""
@@ -367,7 +368,7 @@ def materialize_asset(
         )
 
     dag: DAG | None
-    if not (dag := request.app.state.dag_bag.get_dag(dag_id)):
+    if not (dag := dag_bag.get_dag(dag_id)):
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"DAG with ID `{dag_id}` was not found")
 
     return dag.create_dagrun(

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py

@@ -20,7 +20,7 @@
 from typing import Annotated, Literal, cast
 
 import structlog
-from fastapi import Depends, HTTPException, Query, Request, status
+from fastapi import Depends, HTTPException, Query, status
 from fastapi.exceptions import RequestValidationError
 from pydantic import ValidationError
 from sqlalchemy import select
@@ -33,6 +33,7 @@
 )
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.deps import DagBagDep
 from airflow.api_fastapi.common.parameters import (
     FilterOptionEnum,
     FilterParam,
@@ -147,7 +148,7 @@ def patch_dag_run(
     dag_run_id: str,
     patch_body: DAGRunPatchBody,
     session: SessionDep,
-    request: Request,
+    dag_bag: DagBagDep,
     user: GetUserDep,
     update_mask: list[str] | None = Query(None),
 ) -> DAGRunResponse:
@@ -161,7 +162,7 @@ def patch_dag_run(
             f"The DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
         )
 
-    dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+    dag: DAG = dag_bag.get_dag(dag_id)
 
     if not dag:
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"Dag with id {dag_id} was not found")
@@ -255,7 +256,7 @@ def clear_dag_run(
     dag_id: str,
     dag_run_id: str,
     body: DAGRunClearBody,
-    request: Request,
+    dag_bag: DagBagDep,
     session: SessionDep,
 ) -> TaskInstanceCollectionResponse | DAGRunResponse:
     dag_run = session.scalar(
@@ -267,7 +268,7 @@ def clear_dag_run(
             f"The DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
         )
 
-    dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+    dag: DAG = dag_bag.get_dag(dag_id)
 
     if body.dry_run:
         task_instances = dag.clear(
@@ -331,7 +332,7 @@ def get_dag_runs(
     ],
     readable_dag_runs_filter: ReadableDagRunsFilterDep,
     session: SessionDep,
-    request: Request,
+    dag_bag: DagBagDep,
 ) -> DAGRunCollectionResponse:
     """
     Get all DAG Runs.
@@ -341,7 +342,7 @@ def get_dag_runs(
     query = select(DagRun)
 
     if dag_id != "~":
-        dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+        dag: DAG = dag_bag.get_dag(dag_id)
         if not dag:
             raise HTTPException(status.HTTP_404_NOT_FOUND, f"The DAG with dag_id: `{dag_id}` was not found")
 
@@ -389,7 +390,7 @@ def get_dag_runs(
 def trigger_dag_run(
     dag_id,
     body: TriggerDAGRunPostBody,
-    request: Request,
+    dag_bag: DagBagDep,
     user: GetUserDep,
     session: SessionDep,
 ) -> DAGRunResponse:
@@ -405,7 +406,7 @@ def trigger_dag_run(
         )
 
     try:
-        dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+        dag: DAG = dag_bag.get_dag(dag_id)
         params = body.validate_context(dag)
 
         dag_run = dag.create_dagrun(

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_versions.py

@@ -18,11 +18,12 @@
 
 from typing import Annotated
 
-from fastapi import Depends, HTTPException, Request, status
+from fastapi import Depends, HTTPException, status
 from sqlalchemy import select
 
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.deps import DagBagDep
 from airflow.api_fastapi.common.parameters import (
     FilterParam,
     QueryLimit,
@@ -80,10 +81,9 @@ def get_dag_version(
 )
 def get_dag_versions(
     dag_id: str,
+    session: SessionDep,
     limit: QueryLimit,
     offset: QueryOffset,
-    session: SessionDep,
-    request: Request,
     version_number: Annotated[
         FilterParam[int], Depends(filter_param_factory(DagVersion.version_number, int))
     ],
@@ -97,6 +97,7 @@ def get_dag_versions(
             SortParam(["id", "version_number", "bundle_name", "bundle_version"], DagVersion).dynamic_depends()
         ),
     ],
+    dag_bag: DagBagDep,
 ) -> DAGVersionCollectionResponse:
     """
     Get all DAG Versions.
@@ -106,7 +107,7 @@ def get_dag_versions(
     query = select(DagVersion)
 
     if dag_id != "~":
-        dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+        dag: DAG = dag_bag.get_dag(dag_id)
         if not dag:
             raise HTTPException(status.HTTP_404_NOT_FOUND, f"The DAG with dag_id: `{dag_id}` was not found")
 

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py

@@ -19,7 +19,7 @@
 
 from typing import Annotated
 
-from fastapi import Depends, HTTPException, Query, Request, Response, status
+from fastapi import Depends, HTTPException, Query, Response, status
 from fastapi.exceptions import RequestValidationError
 from pydantic import ValidationError
 from sqlalchemy import select, update
@@ -30,6 +30,7 @@
     paginated_select,
 )
 from airflow.api_fastapi.common.db.dags import generate_dag_with_latest_run_query
+from airflow.api_fastapi.common.deps import DagBagDep
 from airflow.api_fastapi.common.parameters import (
     FilterOptionEnum,
     FilterParam,
@@ -165,9 +166,13 @@ def get_dags(
     ),
     dependencies=[Depends(requires_access_dag(method="GET"))],
 )
-def get_dag(dag_id: str, session: SessionDep, request: Request) -> DAGResponse:
+def get_dag(
+    dag_id: str,
+    session: SessionDep,
+    dag_bag: DagBagDep,
+) -> DAGResponse:
     """Get basic information about a DAG."""
-    dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+    dag: DAG = dag_bag.get_dag(dag_id)
     if not dag:
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"Dag with id {dag_id} was not found")
 
@@ -192,9 +197,9 @@ def get_dag(dag_id: str, session: SessionDep, request: Request) -> DAGResponse:
     ),
     dependencies=[Depends(requires_access_dag(method="GET"))],
 )
-def get_dag_details(dag_id: str, session: SessionDep, request: Request) -> DAGDetailsResponse:
+def get_dag_details(dag_id: str, session: SessionDep, dag_bag: DagBagDep) -> DAGDetailsResponse:
     """Get details of DAG."""
-    dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+    dag: DAG = dag_bag.get_dag(dag_id)
     if not dag:
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"Dag with id {dag_id} was not found")
 

airflow-core/src/airflow/api_fastapi/core_api/routes/public/extra_links.py

@@ -19,10 +19,11 @@
 
 from typing import TYPE_CHECKING
 
-from fastapi import Depends, HTTPException, Request, status
+from fastapi import Depends, HTTPException, status
 from sqlalchemy.sql import select
 
 from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.deps import DagBagDep
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.datamodels.extra_links import ExtraLinkCollectionResponse
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
@@ -49,13 +50,13 @@ def get_extra_links(
     dag_run_id: str,
     task_id: str,
     session: SessionDep,
-    request: Request,
+    dag_bag: DagBagDep,
     map_index: int = -1,
 ) -> ExtraLinkCollectionResponse:
     """Get extra links for task instance."""
     from airflow.models.taskinstance import TaskInstance
 
-    dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+    dag: DAG = dag_bag.get_dag(dag_id)
     if not dag:
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"DAG with ID = {dag_id} not found")
 

airflow-core/src/airflow/api_fastapi/core_api/routes/public/log.py

@@ -27,6 +27,7 @@
 from sqlalchemy.sql import select
 
 from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.deps import DagBagDep
 from airflow.api_fastapi.common.headers import HeaderAcceptJsonOrText
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.common.types import Mimetype
@@ -76,6 +77,7 @@ def get_log(
     try_number: PositiveInt,
     accept: HeaderAcceptJsonOrText,
     request: Request,
+    dag_bag: DagBagDep,
     session: SessionDep,
     full_content: bool = False,
     map_index: int = -1,
@@ -129,7 +131,7 @@ def get_log(
         metadata["end_of_log"] = True
         raise HTTPException(status.HTTP_404_NOT_FOUND, "TaskInstance not found")
 
-    dag = request.app.state.dag_bag.get_dag(dag_id)
+    dag = dag_bag.get_dag(dag_id)
     if dag:
         with contextlib.suppress(TaskNotFound):
             ti.task = dag.get_task(ti.task_id)