apache · Lee-W · Aug 10, 2026
diff --git a/airflow-core/docs/migrations-ref.rst b/airflow-core/docs/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``76c46545c91e`` (head) | ``3c525f44bea8`` | ``3.4.0``         | Add new index for trigger.                                   |
+| ``b8f1c0a4d276`` (head) | ``76c46545c91e`` | ``3.4.0``         | Add timetable_partitioned_at_runtime to dag.                 |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``76c46545c91e``        | ``3c525f44bea8`` | ``3.4.0``         | Add new index for trigger.                                   |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``3c525f44bea8``        | ``b2f1a9c7d4e0`` | ``3.4.0``         | Add indexes on serialized_dag and dag_code.                  |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

@@ -19,14 +19,21 @@
 from typing import TYPE_CHECKING
 
 import structlog
+from sqlalchemy import select
 
 from airflow.exceptions import DeserializationError
+from airflow.models.asset import AssetPartitionDagRun
+from airflow.models.dagrun import DagRun
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.timetables.simple import PartitionedAssetTimetable
+from airflow.utils.state import DagRunState
 
 if TYPE_CHECKING:
+    from pendulum import DateTime
     from sqlalchemy.orm import Session
 
+    from airflow.timetables.base import Timetable
+
 
 log = structlog.get_logger(logger_name=__name__)
 
@@ -87,3 +94,73 @@ def load_partitioned_timetables(
         serdag.dag_id: _extract_partitioned_timetable(serdag)
         for serdag in SerializedDagModel.get_latest_serialized_dags(dag_ids=dag_ids, session=session)
     }
+
+
+def suggest_partition_key_for_dag(
+    *, dag_id: str, timetable: Timetable, now: DateTime, session: Session
+) -> str | None:
+    """
+    Suggest a partition key to pre-fill the manual-trigger form for *dag_id*.
+
+    This is a **guess** used only to pre-fill the UI form field; it is not a
+    validation and callers must still accept ``None`` from
+    ``validate_partition_key``. Returns ``None`` immediately for a timetable
+    that is neither ``partitioned`` nor ``partitioned_at_runtime`` (no query is
+    issued in that case). Otherwise, tries in order and returns the first
+    non-``None`` result:
+
+    1. The oldest pending ``AssetPartitionDagRun`` for *dag_id*
+       (``created_dag_run_id IS NULL``) — an asset event has already arrived
+       for this partition and is only waiting for the run to be created. Only
+       asset-driven timetables have any rows here. The ordering mirrors the
+       scheduler's FIFO claim order in
+       ``SchedulerJobRunner._create_partition_dag_runs``, so the suggestion
+       names the same partition the scheduler would create next rather than a
+       newer one that will not run until the backlog drains.
+    2. ``timetable.suggest_partition_key(now)`` — a purely time-based guess
+       that needs no history, so even a brand-new Dag with no runs or asset
+       events gets a suggestion (asset-driven timetables only; see
+       :meth:`~airflow.timetables.base.Timetable.suggest_partition_key`).
+    3. The ``partition_key`` of the most recent successful ``DagRun`` for
+       *dag_id*. This is the only source available to
+       ``partitioned_at_runtime`` timetables, which have no asset and no
+       temporal anchor of their own. Only the single most recent successful
+       run is consulted — if its ``partition_key`` is ``None`` (an
+       unpartitioned run), this returns ``None`` rather than searching
+       further back.
+
+    Source 3 can name a partition that has already run, and nothing rejects a
+    second ``DagRun`` for the same ``(dag_id, partition_key)`` — re-running a
+    partition is a legitimate re-materialization, so this deliberately neither
+    de-duplicates nor blocks it. The trigger form's help text warns the user
+    that the pre-filled key may already have run.
+
+    ``models.dag.get_last_dagrun`` is intentionally not reused here: it
+    filters ``logical_date.is_not(None)`` and excludes ``MANUAL`` runs by
+    default, which would drop exactly the partitioned runs this needs.
+    """
+    if not timetable.partitioned and not timetable.partitioned_at_runtime:
+        return None
+
+    pending_key = session.execute(
+        select(AssetPartitionDagRun.partition_key)
+        .where(
+            AssetPartitionDagRun.target_dag_id == dag_id,
+            AssetPartitionDagRun.created_dag_run_id.is_(None),
+        )
+        .order_by(AssetPartitionDagRun.created_at, AssetPartitionDagRun.id)
+        .limit(1)
+    ).scalar_one_or_none()
+    if pending_key is not None:
+        return pending_key
+
+    suggested_key = timetable.suggest_partition_key(now)
+    if suggested_key is not None:
+        return suggested_key
+
+    return session.execute(
+        select(DagRun.partition_key)
+        .where(DagRun.dag_id == dag_id, DagRun.state == DagRunState.SUCCESS)
+        .order_by(DagRun.id.desc())
+        .limit(1)
+    ).scalar_one_or_none()
@@ -71,6 +71,15 @@ def _get_file_token_serializer() -> URLSafeSerializer:
     "next_dagrun_run_after": "next_dagrun_create_after",
 }
 
+DAG_RESPONSE_ROUTE_SUPPLIED_FIELDS: frozenset[str] = frozenset(
+    {
+        # Computed per request by the route rather than read off ``DagModel``, so
+        # routes that build the response by reading attributes off a bare
+        # ``DagModel`` must skip these instead of raising ``AttributeError``.
+        "suggested_partition_key",
+    }
+)
+
 
 class DAGResponse(BaseModel):
     """Dag serializer for responses."""
@@ -96,6 +105,7 @@ class DAGResponse(BaseModel):
     timetable_summary: str | None
     timetable_description: str | None
     timetable_partitioned: bool
+    timetable_partitioned_at_runtime: bool
     timetable_periodic: bool
     tags: list[DagTagResponse]
     max_active_tasks: int
@@ -109,6 +119,7 @@ class DAGResponse(BaseModel):
     next_dagrun_run_after: datetime | None
     allowed_run_types: list[DagRunType] | None
     owners: list[str]
+    suggested_partition_key: str | None = None
 
     @field_serializer("tags")
     def serialize_tags(self, tags: list[DagTagResponse]) -> list[DagTagResponse]:

@@ -2929,6 +2929,9 @@ components:
         timetable_partitioned:
           type: boolean
           title: Timetable Partitioned
+        timetable_partitioned_at_runtime:
+          type: boolean
+          title: Timetable Partitioned At Runtime
         timetable_periodic:
           type: boolean
           title: Timetable Periodic
@@ -2990,6 +2993,11 @@ components:
             type: string
           type: array
           title: Owners
+        suggested_partition_key:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Suggested Partition Key
         asset_expression:
           anyOf:
           - oneOf:
@@ -3045,6 +3053,7 @@ components:
       - timetable_summary
       - timetable_description
       - timetable_partitioned
+      - timetable_partitioned_at_runtime
       - timetable_periodic
       - tags
       - max_active_tasks

@@ -12944,6 +12944,9 @@ components:
         timetable_partitioned:
           type: boolean
           title: Timetable Partitioned
+        timetable_partitioned_at_runtime:
+          type: boolean
+          title: Timetable Partitioned At Runtime
         timetable_periodic:
           type: boolean
           title: Timetable Periodic
@@ -13005,6 +13008,11 @@ components:
             type: string
           type: array
           title: Owners
+        suggested_partition_key:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Suggested Partition Key
         catchup:
           type: boolean
           title: Catchup
@@ -13146,6 +13154,7 @@ components:
       - timetable_summary
       - timetable_description
       - timetable_partitioned
+      - timetable_partitioned_at_runtime
       - timetable_periodic
       - tags
       - max_active_tasks
@@ -13258,6 +13267,9 @@ components:
         timetable_partitioned:
           type: boolean
           title: Timetable Partitioned
+        timetable_partitioned_at_runtime:
+          type: boolean
+          title: Timetable Partitioned At Runtime
         timetable_periodic:
           type: boolean
           title: Timetable Periodic
@@ -13319,6 +13331,11 @@ components:
             type: string
           type: array
           title: Owners
+        suggested_partition_key:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Suggested Partition Key
         is_backfillable:
           type: boolean
           title: Is Backfillable
@@ -13346,6 +13363,7 @@ components:
       - timetable_summary
       - timetable_description
       - timetable_partitioned
+      - timetable_partitioned_at_runtime
       - timetable_periodic
       - tags
       - max_active_tasks

@@ -17,13 +17,14 @@
 
 from __future__ import annotations
 
-from typing import Annotated
+from typing import TYPE_CHECKING, Annotated
 
 from fastapi import Depends, HTTPException, Query, Response, status
 from fastapi.exceptions import RequestValidationError
 from pydantic import ValidationError
 from sqlalchemy import delete, func, insert, select, update
 
+from airflow._shared.timezones import timezone
 from airflow.api.common import delete_dag as delete_dag_module
 from airflow.api_fastapi.common.dagbag import DagBagDep, get_latest_version_of_dag
 from airflow.api_fastapi.common.db.common import SessionDep, apply_filters_to_select, paginated_select
@@ -56,6 +57,7 @@
     datetime_range_filter_factory,
     filter_param_factory,
 )
+from airflow.api_fastapi.common.partition_helpers import suggest_partition_key_for_dag
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.compat import HTTP_422_UNPROCESSABLE_CONTENT
 from airflow.api_fastapi.core_api.datamodels.dags import (
@@ -79,9 +81,38 @@
 from airflow.models.dagrun import DagRun
 from airflow.utils.state import DagRunState
 
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+    from airflow.serialization.serialized_objects import SerializedDAG
+
+
 dags_router = AirflowRouter(tags=["DAG"], prefix="/dags")
 
 
+def attach_suggested_partition_key(dag_model: DagModel, dag: SerializedDAG, *, session: Session) -> None:
+    """
+    Attach the manual-trigger partition key suggestion to *dag_model* for serialization.
+
+    ``suggested_partition_key`` is computed per request from live pending
+    partitions and run history, so it cannot be a ``DagModel`` column. Setting a
+    non-mapped attribute leaves the SQLAlchemy identity map untouched — nothing
+    is persisted by a later ``flush()``. Only single-Dag routes carry it; the
+    collection routes leave it at its ``None`` default rather than issue a query
+    per Dag.
+    """
+    setattr(
+        dag_model,
+        "suggested_partition_key",
+        suggest_partition_key_for_dag(
+            dag_id=dag_model.dag_id,
+            timetable=dag.timetable,
+            now=timezone.coerce_datetime(timezone.utcnow()),
+            session=session,
+        ),
+    )
+
+
 @dags_router.get("", dependencies=[Depends(requires_access_dag(method="GET"))])
 def get_dags(
     limit: QueryLimit,
@@ -209,6 +240,8 @@ def get_dag(
         if not key.startswith("_") and not hasattr(dag_model, key):
             setattr(dag_model, key, value)
 
+    attach_suggested_partition_key(dag_model, dag, session=session)
+
     return dag_model
 
 
@@ -236,6 +269,8 @@ def get_dag_details(
         if not key.startswith("_") and not hasattr(dag_model, key):
             setattr(dag_model, key, value)
 
+    attach_suggested_partition_key(dag_model, dag, session=session)
+
     # Check if this Dag is marked as favorite by the current user
     user_id = str(user.get_id())
     is_favorite = (

@@ -58,7 +58,11 @@
     filter_param_factory,
 )
 from airflow.api_fastapi.common.router import AirflowRouter
-from airflow.api_fastapi.core_api.datamodels.dags import DAG_ALIAS_MAPPING, DAGResponse
+from airflow.api_fastapi.core_api.datamodels.dags import (
+    DAG_ALIAS_MAPPING,
+    DAG_RESPONSE_ROUTE_SUPPLIED_FIELDS,
+    DAGResponse,
+)
 from airflow.api_fastapi.core_api.datamodels.ui.dag_runs import DAGRunLightResponse
 from airflow.api_fastapi.core_api.datamodels.ui.dags import (
     DAGRunStateCountsResponse,
@@ -255,13 +259,17 @@ def get_dags(
     # aggregate rows by dag_id
     # Build the dict dynamically from DAGResponse.model_fields so that new fields
     # added to DAGResponse are picked up automatically without code changes here.
+    # ``getattr`` is deliberately left without a default so a field that is neither
+    # a ``DagModel`` attribute nor declared route-supplied fails loudly here
+    # instead of silently serializing as ``None``.
     dag_runs_by_dag_id: dict[str, DAGWithLatestDagRunsResponse] = {}
     for dag in dags:
         dag_data = {
             DAG_ALIAS_MAPPING.get(field_name, field_name): getattr(
                 dag, DAG_ALIAS_MAPPING.get(field_name, field_name)
             )
             for field_name in DAGResponse.model_fields
+            if field_name not in DAG_RESPONSE_ROUTE_SUPPLIED_FIELDS
         }
         dag_data.update(
             {

@@ -389,6 +389,7 @@ def _get_dagbag_dag_details(dag: DAG) -> dict:
         "timetable_summary": core_timetable.summary,
         "timetable_description": core_timetable.description,
         "timetable_partitioned": core_timetable.partitioned,
+        "timetable_partitioned_at_runtime": core_timetable.partitioned_at_runtime,
         "timetable_periodic": core_timetable.periodic,
         "tags": dag.tags,
         "max_active_tasks": dag.max_active_tasks,
@@ -403,6 +404,8 @@ def _get_dagbag_dag_details(dag: DAG) -> dict:
         "next_dagrun_logical_date": None,
         "next_dagrun_run_after": None,
         "allowed_run_types": dag.allowed_run_types,
+        # No DB session here, so the trigger-form suggestion sources are unavailable.
+        "suggested_partition_key": None,
         "is_backfillable": core_timetable.periodic
         and (dag.allowed_run_types is None or DagRunType.BACKFILL_JOB in dag.allowed_run_types),
     }

@@ -682,6 +682,7 @@ def update_dags(
             dm.timetable_summary = dag.timetable.summary
             dm.timetable_description = dag.timetable.description
             dm.timetable_partitioned = dag.timetable.partitioned
+            dm.timetable_partitioned_at_runtime = dag.timetable.partitioned_at_runtime
             dm.timetable_periodic = dag.timetable.periodic
             dm.partition_mapper_info = dag.timetable.partition_mapper_info
             dm.fail_fast = dag.fail_fast if dag.fail_fast is not None else False