Pre-populate a suggested partition key when manually triggering cron-partitioned Dags

airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py

@@ -71,6 +71,11 @@ def _get_file_token_serializer() -> URLSafeSerializer:
     "next_dagrun_run_after": "next_dagrun_create_after",
 }
 
+# Fields the route layer attaches per request instead of reading off DagModel. Callers that
+# build a response by iterating DAGResponse.model_fields over a bare DagModel must skip these,
+# or getattr raises AttributeError and the endpoint returns 500.
+DAG_RESPONSE_ROUTE_SUPPLIED_FIELDS: frozenset[str] = frozenset({"suggested_partition_key"})
+
 
 class DAGResponse(BaseModel):
     """Dag serializer for responses."""
@@ -96,6 +101,7 @@ class DAGResponse(BaseModel):
     timetable_summary: str | None
     timetable_description: str | None
     timetable_partitioned: bool
+    suggested_partition_key: str | None = None
     timetable_periodic: bool
     tags: list[DagTagResponse]
     max_active_tasks: int

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -2929,6 +2929,11 @@ components:
         timetable_partitioned:
           type: boolean
           title: Timetable Partitioned
+        suggested_partition_key:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Suggested Partition Key
         timetable_periodic:
           type: boolean
           title: Timetable Periodic

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -12944,6 +12944,11 @@ components:
         timetable_partitioned:
           type: boolean
           title: Timetable Partitioned
+        suggested_partition_key:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Suggested Partition Key
         timetable_periodic:
           type: boolean
           title: Timetable Periodic
@@ -13258,6 +13263,11 @@ components:
         timetable_partitioned:
           type: boolean
           title: Timetable Partitioned
+        suggested_partition_key:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Suggested Partition Key
         timetable_periodic:
           type: boolean
           title: Timetable Periodic

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

@@ -24,6 +24,7 @@
 from pydantic import ValidationError
 from sqlalchemy import delete, func, insert, select, update
 
+from airflow._shared.timezones.timezone import coerce_datetime, utcnow
 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
@@ -209,6 +210,10 @@ def get_dag(
         if not key.startswith("_") and not hasattr(dag_model, key):
             setattr(dag_model, key, value)
 
+    setattr(
+        dag_model, "suggested_partition_key", dag.timetable.suggest_partition_key(coerce_datetime(utcnow()))
+    )
+
     return dag_model
 
 
@@ -258,6 +263,9 @@ def get_dag_details(
     # Add is_favorite and active_runs_count fields to the Dag model
     setattr(dag_model, "is_favorite", is_favorite)
     setattr(dag_model, "active_runs_count", active_runs_count)
+    setattr(
+        dag_model, "suggested_partition_key", dag.timetable.suggest_partition_key(coerce_datetime(utcnow()))
+    )
 
     return DAGDetailsResponse.model_validate(dag_model)
 

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

@@ -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,16 @@ 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.
+    # Route-supplied fields are skipped: they have no DagModel attribute to read, and
+    # this list view does not compute them, so they fall back to their model default.
     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(
             {

airflow-core/src/airflow/timetables/base.py

@@ -338,6 +338,35 @@ def _decode_partition_date(self, partition_key: str) -> datetime.datetime | None
         """
         return None
 
+    def suggest_partition_key(self, now: DateTime) -> str | None:
+        """
+        Suggest a partition key for a manual trigger happening at *now*.
+
+        Used to pre-populate the partition key field when a user manually triggers a Dag
+        run through the UI, so they are not required to know the exact key format by
+        heart. Returns ``None`` when this timetable is not ``partitioned``, when it defers
+        partition selection to runtime (``partitioned_at_runtime``), or when the concrete
+        timetable does not implement a suggestion (the default).
+
+        :param now: The instant to compute the suggestion for — normally "right now", the
+            moment the trigger form is rendered.
+        :returns: A partition key string ready to prefill the trigger form, or ``None`` if
+            no suggestion is available.
+        """
+        if not self.partitioned or self.partitioned_at_runtime:
+            return None
+        return self._suggest_partition_key(now)
+
+    def _suggest_partition_key(self, now: DateTime) -> str | None:
+        """
+        Compute a suggested partition key for *now*.
+
+        Called by :meth:`suggest_partition_key` only after the partitioned-state guards
+        pass. The default returns ``None``; partitioned timetables that can derive a
+        "current" partition from a point in time override this.
+        """
+        return None
+
     @property
     def partition_mapper_info(self) -> list[PartitionMapperInfo]:
         """

airflow-core/src/airflow/timetables/trigger.py

@@ -467,6 +467,18 @@ def _get_partition_info(self, run_date: DateTime) -> tuple[DateTime, str]:
         partition_key = self._format_key(partition_date)
         return partition_date, partition_key
 
+    def _suggest_partition_key(self, now: DateTime) -> str:
+        """
+        Suggest the key of the most recently elapsed partition as of *now*.
+
+        The tick is derived from the cron expression alone, so the suggestion is not
+        clamped by the Dag's ``start_date`` — a Dag that has not started scheduling yet
+        still gets the tick preceding *now*, which is a valid key for a manual run.
+        """
+        aligned = self._align_to_prev(coerce_datetime(now))
+        _, partition_key = self._get_partition_info(run_date=aligned)
+        return partition_key
+
     def iter_partition_dagrun_infos(
         self,
         *,

airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -2951,6 +2951,17 @@ export const $DAGDetailsResponse = {
             type: 'boolean',
             title: 'Timetable Partitioned'
         },
+        suggested_partition_key: {
+            anyOf: [
+                {
+                    type: 'string'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Suggested Partition Key'
+        },
         timetable_periodic: {
             type: 'boolean',
             title: 'Timetable Periodic'
@@ -3446,6 +3457,17 @@ export const $DAGResponse = {
             type: 'boolean',
             title: 'Timetable Partitioned'
         },
+        suggested_partition_key: {
+            anyOf: [
+                {
+                    type: 'string'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Suggested Partition Key'
+        },
         timetable_periodic: {
             type: 'boolean',
             title: 'Timetable Periodic'
@@ -9335,6 +9357,17 @@ export const $DAGWithLatestDagRunsResponse = {
             type: 'boolean',
             title: 'Timetable Partitioned'
         },
+        suggested_partition_key: {
+            anyOf: [
+                {
+                    type: 'string'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Suggested Partition Key'
+        },
         timetable_periodic: {
             type: 'boolean',
             title: 'Timetable Periodic'

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -865,6 +865,7 @@ export type DAGDetailsResponse = {
     timetable_summary: string | null;
     timetable_description: string | null;
     timetable_partitioned: boolean;
+    suggested_partition_key?: string | null;
     timetable_periodic: boolean;
     tags: Array<DagTagResponse>;
     max_active_tasks: number;
@@ -949,6 +950,7 @@ export type DAGResponse = {
     timetable_summary: string | null;
     timetable_description: string | null;
     timetable_partitioned: boolean;
+    suggested_partition_key?: string | null;
     timetable_periodic: boolean;
     tags: Array<DagTagResponse>;
     max_active_tasks: number;
@@ -2366,6 +2368,7 @@ export type DAGWithLatestDagRunsResponse = {
     timetable_summary: string | null;
     timetable_description: string | null;
     timetable_partitioned: boolean;
+    suggested_partition_key?: string | null;
     timetable_periodic: boolean;
     tags: Array<DagTagResponse>;
     max_active_tasks: number;

airflow-core/src/airflow/ui/src/components/TriggerDag/TriggerDAGForm.test.tsx

@@ -214,4 +214,151 @@ describe("TriggerDAGForm", () => {
     await waitFor(() => expect(screen.getByText("dagRun.partitionKey")).toBeInTheDocument());
     expect(screen.getByText("components:triggerDag.partitionKeyHelp")).toBeInTheDocument();
   });
+
+  it("pre-fills the partition key field with the suggested value", async () => {
+    const { container } = render(
+      <TriggerDAGForm
+        dagDisplayName="Partitioned Dag"
+        dagId="example_partitioned_dag"
+        error={undefined}
+        hasSchedule={false}
+        isPartitioned
+        isPaused={false}
+        isPending={false}
+        onSubmitTrigger={vi.fn()}
+        open
+        suggestedPartitionKey="2026-02-18T00:00:00"
+      />,
+      { wrapper: Wrapper },
+    );
+
+    fireEvent.click(screen.getByText("Advanced Options"));
+
+    await waitFor(() =>
+      expect(container.querySelector<HTMLInputElement>('input[name="partitionKey"]')).toBeInTheDocument(),
+    );
+    const partitionKeyField = container.querySelector<HTMLInputElement>('input[name="partitionKey"]');
+
+    expect(partitionKeyField?.value).toBe("2026-02-18T00:00:00");
+  });
+
+  it("leaves the partition key field blank when no suggestion is provided", async () => {
+    const { container } = render(
+      <TriggerDAGForm
+        dagDisplayName="Partitioned Dag"
+        dagId="example_partitioned_dag"
+        error={undefined}
+        hasSchedule={false}
+        isPartitioned
+        isPaused={false}
+        isPending={false}
+        onSubmitTrigger={vi.fn()}
+        open
+      />,
+      { wrapper: Wrapper },
+    );
+
+    fireEvent.click(screen.getByText("Advanced Options"));
+
+    await waitFor(() =>
+      expect(container.querySelector<HTMLInputElement>('input[name="partitionKey"]')).toBeInTheDocument(),
+    );
+    const partitionKeyField = container.querySelector<HTMLInputElement>('input[name="partitionKey"]');
+
+    expect(partitionKeyField?.value).toBe("");
+  });
+
+  it("applies a fresher suggestion arriving after the form is shown", async () => {
+    const { container, rerender } = render(
+      <TriggerDAGForm
+        dagDisplayName="Partitioned Dag"
+        dagId="example_partitioned_dag"
+        error={undefined}
+        hasSchedule={false}
+        isPartitioned
+        isPaused={false}
+        isPending={false}
+        onSubmitTrigger={vi.fn()}
+        open
+        suggestedPartitionKey="2026-02-18T00:00:00"
+      />,
+      { wrapper: Wrapper },
+    );
+
+    fireEvent.click(screen.getByText("Advanced Options"));
+
+    await waitFor(() =>
+      expect(container.querySelector<HTMLInputElement>('input[name="partitionKey"]')).toBeInTheDocument(),
+    );
+
+    rerender(
+      <TriggerDAGForm
+        dagDisplayName="Partitioned Dag"
+        dagId="example_partitioned_dag"
+        error={undefined}
+        hasSchedule={false}
+        isPartitioned
+        isPaused={false}
+        isPending={false}
+        onSubmitTrigger={vi.fn()}
+        open
+        suggestedPartitionKey="2026-02-18T01:00:00"
+      />,
+    );
+
+    await waitFor(() =>
+      expect(container.querySelector<HTMLInputElement>('input[name="partitionKey"]')?.value).toBe(
+        "2026-02-18T01:00:00",
+      ),
+    );
+  });
+
+  it("keeps an edited partition key when a fresher suggestion arrives", async () => {
+    const { container, rerender } = render(
+      <TriggerDAGForm
+        dagDisplayName="Partitioned Dag"
+        dagId="example_partitioned_dag"
+        error={undefined}
+        hasSchedule={false}
+        isPartitioned
+        isPaused={false}
+        isPending={false}
+        onSubmitTrigger={vi.fn()}
+        open
+        suggestedPartitionKey="2026-02-18T00:00:00"
+      />,
+      { wrapper: Wrapper },
+    );
+
+    fireEvent.click(screen.getByText("Advanced Options"));
+
+    await waitFor(() =>
+      expect(container.querySelector<HTMLInputElement>('input[name="partitionKey"]')).toBeInTheDocument(),
+    );
+
+    const partitionKeyField = container.querySelector<HTMLInputElement>('input[name="partitionKey"]');
+
+    fireEvent.change(partitionKeyField as HTMLInputElement, { target: { value: "2026-01-01T00:00:00" } });
+
+    rerender(
+      <TriggerDAGForm
+        dagDisplayName="Partitioned Dag"
+        dagId="example_partitioned_dag"
+        error={undefined}
+        hasSchedule={false}
+        isPartitioned
+        isPaused={false}
+        isPending={false}
+        onSubmitTrigger={vi.fn()}
+        open
+        suggestedPartitionKey="2026-02-18T01:00:00"
+      />,
+    );
+
+    await waitFor(() =>
+      expect(container.querySelector<HTMLInputElement>('input[name="partitionKey"]')?.value).toBe(
+        "2026-01-01T00:00:00",
+      ),
+    );
+  });
 });