Merge branch 'main' into feat/databricks-sql-query-tags

Author: SkastVnT

.github/skills/airflow-translations/locales/de.md

@@ -74,6 +74,7 @@ that **must be used consistently**:
 | Dag Processor         | Dag Prozessor                         |                                                |
 | Heartbeat             | Lebenszeichen                         | e.g., "Letztes Lebenszeichen"                  |
 | Upstream / Downstream | Vorgelagert / Nachgelagert            |                                                |
+| Deadline              | Frist                                 |                                                |
 
 ## 3. Task/Run States
 

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

@@ -291,6 +291,7 @@ def create_backfill_dry_run(
             to_date=to_date,
             reverse=body.run_backwards,
             reprocess_behavior=body.reprocess_behavior,
+            dag_run_conf=body.dag_run_conf,
             session=session,
         )
         backfills = [

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/variable.py

@@ -34,3 +34,10 @@ class VariablePostBody(StrictBaseModel):
 
     value: str | None = Field(alias="val")
     description: str | None = Field(default=None)
+
+
+class VariableKeysResponse(StrictBaseModel):
+    """Variable keys schema for list responses."""
+
+    keys: list[str]
+    total_entries: int

airflow-core/src/airflow/api_fastapi/execution_api/routes/variables.py

@@ -20,9 +20,12 @@
 import logging
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, HTTPException, Path, Request, status
+from fastapi import APIRouter, Depends, HTTPException, Path, Query, Request, status
+from sqlalchemy import func, select
 
+from airflow.api_fastapi.common.db.common import SessionDep
 from airflow.api_fastapi.execution_api.datamodels.variable import (
+    VariableKeysResponse,
     VariablePostBody,
     VariableResponse,
 )
@@ -54,17 +57,55 @@ async def has_variable_access(
     return True
 
 
-router = APIRouter(
-    responses={status.HTTP_404_NOT_FOUND: {"description": "Variable not found"}},
-    dependencies=[Depends(has_variable_access)],
-)
+router = APIRouter()
 
 log = logging.getLogger(__name__)
 
 
+# /keys must be declared before /{variable_key:path} so the static path is
+# matched first; otherwise the catch-all path param would swallow it.
+# has_variable_access is applied per-route below (not at router level) because
+# it requires a variable_key path parameter that /keys does not have.
+@router.get(
+    "/keys",
+    responses={
+        status.HTTP_401_UNAUTHORIZED: {"description": "Unauthorized"},
+    },
+)
+def get_variable_keys(
+    session: SessionDep,
+    team_name: Annotated[str | None, Depends(get_team_name_dep)] = None,
+    prefix: Annotated[str | None, Query()] = None,
+    limit: Annotated[int, Query(ge=1, le=10_000)] = 1000,
+    offset: Annotated[int, Query(ge=0)] = 0,
+) -> VariableKeysResponse:
+    """
+    Get Airflow Variable keys, optionally filtered by prefix.
+
+    .. note::
+        This endpoint deliberately bypasses the per-variable ``has_variable_access``
+        check, since access scoping requires a specific variable key. Any authenticated
+        task within a team can therefore enumerate every variable key in that team —
+        including keys for variables it would not be allowed to read. This is consistent
+        with Airflow's security model (workers within a deployment trust each other),
+        but the asymmetry between key enumeration and value access is intentional.
+    """
+    stmt = select(Variable.key).order_by(Variable.key)
+    if prefix is not None:
+        stmt = stmt.where(Variable.key.startswith(prefix, autoescape=True))
+    if team_name is not None:
+        stmt = stmt.where(Variable.team_name == team_name)
+
+    total_entries = session.scalar(select(func.count()).select_from(stmt.subquery())) or 0
+    keys = session.scalars(stmt.offset(offset).limit(limit)).all()
+    return VariableKeysResponse(keys=list(keys), total_entries=total_entries)
+
+
 @router.get(
     "/{variable_key:path}",
+    dependencies=[Depends(has_variable_access)],
     responses={
+        status.HTTP_404_NOT_FOUND: {"description": "Variable not found"},
         status.HTTP_401_UNAUTHORIZED: {"description": "Unauthorized"},
         status.HTTP_403_FORBIDDEN: {"description": "Task does not have access to the variable"},
     },
@@ -90,8 +131,10 @@ def get_variable(
 
 @router.put(
     "/{variable_key:path}",
+    dependencies=[Depends(has_variable_access)],
     status_code=status.HTTP_201_CREATED,
     responses={
+        status.HTTP_404_NOT_FOUND: {"description": "Variable not found"},
         status.HTTP_401_UNAUTHORIZED: {"description": "Unauthorized"},
         status.HTTP_403_FORBIDDEN: {"description": "Task does not have access to the variable"},
     },
@@ -108,8 +151,10 @@ def put_variable(
 
 @router.delete(
     "/{variable_key:path}",
+    dependencies=[Depends(has_variable_access)],
     status_code=status.HTTP_204_NO_CONTENT,
     responses={
+        status.HTTP_404_NOT_FOUND: {"description": "Variable not found"},
         status.HTTP_401_UNAUTHORIZED: {"description": "Unauthorized"},
         status.HTTP_403_FORBIDDEN: {"description": "Task does not have access to the variable"},
     },

airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py

@@ -46,9 +46,11 @@
     AddStateEndpoints,
     AddTeamNameField,
 )
+from airflow.api_fastapi.execution_api.versions.v2026_06_30 import AddVariableKeysEndpoint
 
 bundle = VersionBundle(
     HeadVersion(),
+    Version("2026-06-30", AddVariableKeysEndpoint),
     Version(
         "2026-06-16",
         AddRetryPolicyFields,

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py

@@ -0,0 +1,28 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from cadwyn import VersionChange, endpoint
+
+
+class AddVariableKeysEndpoint(VersionChange):
+    """Add GET /variables/keys endpoint for listing variable keys with optional prefix filter."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (endpoint("/variables/keys", ["GET"]).didnt_exist,)

airflow-core/src/airflow/dag_processing/processor.py

@@ -50,6 +50,7 @@
     GetTaskStates,
     GetTICount,
     GetVariable,
+    GetVariableKeys,
     GetXCom,
     GetXComCount,
     GetXComSequenceItem,
@@ -61,6 +62,7 @@
     PrevSuccessfulDagRunResult,
     PutVariable,
     TaskStatesResult,
+    VariableKeysResult,
     VariableResult,
     XComCountResponse,
     XComResult,
@@ -128,6 +130,7 @@ class DagFileParsingResult(BaseModel):
     DagFileParsingResult
     | GetConnection
     | GetVariable
+    | GetVariableKeys
     | PutVariable
     | GetTaskStates
     | GetTICount
@@ -147,6 +150,7 @@ class DagFileParsingResult(BaseModel):
     DagFileParseRequest
     | ConnectionResult
     | VariableResult
+    | VariableKeysResult
     | TaskStatesResult
     | PreviousDagRunResult
     | PreviousTIResult
@@ -628,6 +632,10 @@ def _handle_request(self, msg: ToManager, log: FilteringBoundLogger, req_id: int
                 dump_opts = {"exclude_unset": True}
             else:
                 resp = var
+        elif isinstance(msg, GetVariableKeys):
+            from airflow.sdk.execution_time.request_handlers import handle_get_variable_keys
+
+            resp, dump_opts = handle_get_variable_keys(self.client, msg)
         elif isinstance(msg, PutVariable):
             self.client.variables.set(msg.key, msg.value, msg.description)
         elif isinstance(msg, DeleteVariable):

airflow-core/src/airflow/executors/base_executor.py

@@ -297,15 +297,15 @@ def _get_workloads_to_schedule(self, open_slots: int) -> list[tuple[WorkloadKey,
 
         return workloads_to_schedule
 
-    def _process_workloads(self, workloads: Sequence[ExecutorWorkload]) -> None:
+    def _process_workloads(self, workload_items: Sequence[ExecutorWorkload]) -> None:
         """
         Process the given workloads.
 
         This method must be implemented by subclasses to define how they handle
         the execution of workloads (e.g., queuing them to workers, submitting to
         external systems, etc.).
 
-        :param workloads: List of workloads to process
+        :param workload_items: List of workloads to process
         """
         raise NotImplementedError(f"{type(self).__name__} must implement _process_workloads()")
 

airflow-core/src/airflow/jobs/triggerer_job_runner.py

@@ -74,6 +74,7 @@
     GetTaskStates,
     GetTICount,
     GetVariable,
+    GetVariableKeys,
     GetXCom,
     MaskSecret,
     OKResponse,
@@ -82,6 +83,7 @@
     TaskStatesResult,
     TICount,
     UpdateHITLDetail,
+    VariableKeysResult,
     VariableResult,
     XComResult,
     _new_encoder,
@@ -90,6 +92,7 @@
 from airflow.sdk.execution_time.request_handlers import (
     handle_get_connection,
     handle_get_variable,
+    handle_get_variable_keys,
     handle_mask_secret,
 )
 from airflow.sdk.execution_time.supervisor import WatchedSubprocess, make_buffered_socket_reader
@@ -302,6 +305,7 @@ def from_api_response(cls, response: HITLDetailResponse) -> HITLDetailResponseRe
     | messages.TriggerStateSync
     | ConnectionResult
     | VariableResult
+    | VariableKeysResult
     | XComResult
     | DagRunStateResult
     | DRCount
@@ -323,6 +327,7 @@ def from_api_response(cls, response: HITLDetailResponse) -> HITLDetailResponseRe
     | GetConnection
     | DeleteVariable
     | GetVariable
+    | GetVariableKeys
     | PutVariable
     | DeleteXCom
     | GetXCom
@@ -534,6 +539,8 @@ def _handle_request(self, msg: ToTriggerSupervisor, log: FilteringBoundLogger, r
             resp = self.client.variables.delete(msg.key)
         elif isinstance(msg, GetVariable):
             resp, dump_opts = handle_get_variable(self.client, msg)
+        elif isinstance(msg, GetVariableKeys):
+            resp, dump_opts = handle_get_variable_keys(self.client, msg)
         elif isinstance(msg, PutVariable):
             self.client.variables.set(msg.key, msg.value, msg.description)
         elif isinstance(msg, DeleteXCom):

airflow-core/src/airflow/models/backfill.py

@@ -295,6 +295,7 @@ def _do_dry_run(
     reverse: bool,
     reprocess_behavior: ReprocessBehavior,
     session: Session,
+    dag_run_conf: dict | None = None,
 ) -> Iterable[DagRunInfo]:
     from airflow.models.serialized_dag import SerializedDagModel
 
@@ -310,7 +311,7 @@ def _do_dry_run(
     if dag.allowed_run_types is not None and DagRunType.BACKFILL_JOB not in dag.allowed_run_types:
         raise DagRunTypeNotAllowed(f"Dag with dag_id: '{dag_id}' does not allow backfill runs")
 
-    _validate_backfill_params(dag, reverse, from_date, to_date, reprocess_behavior)
+    _validate_backfill_params(dag, reverse, from_date, to_date, reprocess_behavior, dag_run_conf)
 
     dagrun_info_list = _get_info_list(
         dag=dag,