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

Code before:
from airflow.api_fastapi.core_api.security import (
    GetUserDep,
    ReadableDagRunsFilterDep,
    requires_access_asset,
    requires_access_dag,
)
from airflow.api_fastapi.core_api.services.public.dag_run import DagRunWaiter
from airflow.api_fastapi.logging.decorators import action_logging
from airflow.configuration import conf
from airflow.listeners.listener import get_listener_manager
from airflow.models import DagModel, DagRun
from airflow.models.asset import AssetEvent
from airflow.models.dag_version import DagVersion
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType

log = structlog.get_logger(__name__)

dag_run_router = AirflowRouter(tags=["DagRun"], prefix="/dags/{dag_id}/dagRuns")


@dag_run_router.get(
    "/recent-configurations",
    responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
    dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.RUN))],
)
def get_recent_dag_run_configurations(
    dag_id: str,
    session: SessionDep,
    dag_bag: DagBagDep,
    limit: Annotated[int, Query(ge=1, le=50, description="Maximum number of recent configurations to return")] = 5,
) -> RecentConfigurationsResponse:
    """Get recent manual DAG run configurations for a specific DAG."""
    # Check if the DAG exists
    get_latest_version_of_dag(dag_bag, dag_id, session)
    
    # Get the configured limit from settings
    configured_limit = conf.getint("webserver", "num_recent_configurations_for_trigger", fallback=5)
    actual_limit = min(limit, configured_limit)
    
    # Query recent manual DAG runs with configurations
    query = (
        select(DagRun.run_id, DagRun.conf, DagRun.logical_date, DagRun.start_date)
        .where(
            DagRun.dag_id == dag_id,
            DagRun.run_type == DagRunType.MANUAL,
            DagRun.conf.is_not(None)
        )
        .order_by(DagRun.start_date.desc())
        .limit(actual_limit)
    )

Review comment (suggestion):
`airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dag_run.py` might be a better place for this route.

Code after:
    GetUserDep,
    ReadableDagRunsFilterDep,
    requires_access_asset,
    requires_access_dag,
)
from airflow.api_fastapi.core_api.services.public.dag_run import DagRunWaiter
from airflow.api_fastapi.logging.decorators import action_logging
from airflow.listeners.listener import get_listener_manager
from airflow.models import DagModel, DagRun
from airflow.models.asset import AssetEvent
from airflow.models.dag_version import DagVersion
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType

log = structlog.get_logger(__name__)

dag_run_router = AirflowRouter(tags=["DagRun"], prefix="/dags/{dag_id}/dagRuns")


@dag_run_router.get(
    "/{dag_run_id}",
    responses=create_openapi_http_exception_doc(
        [
            status.HTTP_404_NOT_FOUND,
        ]
    ),
    dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.RUN))],
)
def get_dag_run(dag_id: str, dag_run_id: str, session: SessionDep) -> DAGRunResponse:
    dag_run = session.scalar(
        select(DagRun).filter_by(dag_id=dag_id, run_id=dag_run_id).options(joinedload(DagRun.dag_model))
    )
    if dag_run is None:
        raise HTTPException(
            status.HTTP_404_NOT_FOUND,
            f"The DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
        )

    return dag_run


@dag_run_router.delete(
    "/{dag_run_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    responses=create_openapi_http_exception_doc(
        [
            status.HTTP_400_BAD_REQUEST,
            status.HTTP_404_NOT_FOUND,
        ],
    ),
    dependencies=[