Repository: apache/airflow
File: airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py (Python)

Code before:
from airflow.api_fastapi.core_api.services.ui.dependencies import (
    extract_single_connected_component,
    get_data_dependencies,
    get_scheduling_dependencies,
)

dependencies_router = AirflowRouter(tags=["Dependencies"])


@dependencies_router.get(
    "/dependencies",
    responses=create_openapi_http_exception_doc(
        [
            status.HTTP_404_NOT_FOUND,
        ]
    ),
    dependencies=[Depends(requires_access_dag("GET", DagAccessEntity.DEPENDENCIES))],
)
def get_dependencies(
    session: SessionDep,
    user: GetUserDep,
    node_id: str | None = None,
    dependency_type: Literal["scheduling", "data"] = "scheduling",
) -> BaseGraphResponse:
    """Dependencies graph."""
    auth_manager = get_auth_manager()
    readable_dag_ids = auth_manager.get_authorized_dag_ids(method="GET", user=user)

    if dependency_type == "data":
        if node_id is None or not node_id.startswith("asset:"):
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, "Data dependencies require an asset node_id (e.g., 'asset:123')"
            )

        try:
            asset_id = int(node_id.replace("asset:", ""))
        except ValueError:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid asset node_id: {node_id}")

        data = get_data_dependencies(asset_id, session, readable_dag_ids)
        return BaseGraphResponse(**data)

    data = get_scheduling_dependencies(readable_dag_ids)

    if node_id is not None:
        try:
            data = extract_single_connected_component(node_id, data["nodes"], data["edges"])
        except ValueError as e:
            raise HTTPException(404, str(e))

    return BaseGraphResponse(**data)

Review comment (security):
You can also use `AuthManagerDep` to get the auth manager

Code after:
    extract_single_connected_component,
    get_data_dependencies,
    get_scheduling_dependencies,
)

dependencies_router = AirflowRouter(tags=["Dependencies"])


@dependencies_router.get(
    "/dependencies",
    responses=create_openapi_http_exception_doc(
        [
            status.HTTP_404_NOT_FOUND,
        ]
    ),
    dependencies=[Depends(requires_access_dag("GET", DagAccessEntity.DEPENDENCIES))],
)
def get_dependencies(
    session: SessionDep,
    readable_dags_filter: ReadableDagsFilterDep,
    node_id: str | None = None,
    dependency_type: Literal["scheduling", "data"] = "scheduling",
) -> BaseGraphResponse:
    """Dependencies graph."""
    if dependency_type == "data":
        if node_id is None or not node_id.startswith("asset:"):
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, "Data dependencies require an asset node_id (e.g., 'asset:123')"
            )

        try:
            asset_id = int(node_id.replace("asset:", ""))
        except ValueError:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid asset node_id: {node_id}")

        data = get_data_dependencies(asset_id, session, readable_dags_filter.value)
        return BaseGraphResponse(**data)

    data = get_scheduling_dependencies(readable_dags_filter.value)

    if node_id is not None:
        try:
            data = extract_single_connected_component(node_id, data["nodes"], data["edges"])
        except ValueError as e:
            raise HTTPException(404, str(e))

    return BaseGraphResponse(**data)