Merge branch 'main' into shared-serialization

Author: amoghrajesh

.github/CODEOWNERS

@@ -99,7 +99,9 @@ airflow-core/src/airflow/ui/public/i18n/locales/zh-TW/ @Lee-W @jason810496 @guan
 /providers/hashicorp/ @hussein-awala
 /providers/informatica/ @RNHTTR # + @cetingokhan @sertaykabuk @umutozel
 /providers/keycloak/ @vincbeck @bugraoz93
-/providers/microsoft/azure/ @dabla
+/providers/microsoft/azure/docs/**/msgraph.rst @dabla
+/providers/microsoft/azure/src/**/msgraph.py @dabla
+/providers/microsoft/azure/tests/**/*msgraph.py @dabla
 /providers/openlineage/ @mobuchowski
 /providers/smtp/ @hussein-awala
 /providers/snowflake/ @potiuk

airflow-core/docs/authoring-and-scheduling/assets.rst

@@ -401,3 +401,151 @@ As mentioned in :ref:`Fetching information from previously emitted asset events<
         def consume_asset_alias_events(*, inlet_events):
             events = inlet_events[AssetAlias("example-alias")]
             last_row_count = events[-1].extra["row_count"]
+
+Asset partitions
+----------------
+
+.. versionadded:: 3.2.0
+
+Asset events can include a ``partition_key`` to make it _partitioned__. This lets you model
+the same asset at partition granularity (for example, ``2026-03-10T09:00:00`` for an
+hourly partition).
+
+To produce partitioned events on a schedule, use
+``CronPartitionTimetable`` in the producer Dag (or ``@asset``). This timetable
+creates asset events with a partition key on each run.
+
+.. code-block:: python
+
+    from airflow.sdk import CronPartitionTimetable, asset
+
+
+    @asset(
+        uri="file://incoming/player-stats/team_b.csv",
+        schedule=CronPartitionTimetable("15 * * * *", timezone="UTC"),
+    )
+    def team_b_player_stats():
+        pass
+
+Partitioned events are intended for partition-aware downstream scheduling, and
+do not trigger non-partition-aware Dags.
+
+For downstream partition-aware scheduling, use ``PartitionedAssetTimetable``:
+
+.. code-block:: python
+
+    from airflow.sdk import DAG, HourlyMapper, PartitionedAssetTimetable
+
+    with DAG(
+        dag_id="clean_and_combine_player_stats",
+        schedule=PartitionedAssetTimetable(
+            assets=team_a_player_stats & team_b_player_stats & team_c_player_stats,
+            default_partition_mapper=HourlyMapper(),
+        ),
+        catchup=False,
+    ):
+        ...
+
+``PartitionedAssetTimetable`` requires partitioned asset events. If an asset
+event does not contain a ``partition_key``, it will not trigger a downstream
+Dag that uses ``PartitionedAssetTimetable``.
+
+``default_partition_mapper`` is used for every upstream asset unless you
+override it via ``partition_mapper_config``. The default mapper is
+``IdentityMapper`` (no key transformation).
+
+Partition mappers define how upstream partition keys are transformed to the
+downstream Dag partition key:
+
+* ``IdentityMapper`` keeps keys unchanged.
+* Temporal mappers such as ``HourlyMapper``, ``DailyMapper``, and
+  ``YearlyMapper`` normalize time keys to a chosen grain. For input key
+  ``2026-03-10T09:37:51``, the default outputs are:
+
+  * ``HourlyMapper`` -> ``2026-03-10T09``
+  * ``DailyMapper`` -> ``2026-03-10``
+  * ``YearlyMapper`` -> ``2026``
+* ``ProductMapper`` maps composite keys segment-by-segment.
+  It applies one mapper per segment and then rejoins the mapped segments.
+  For example, with key ``us|2026-03-10T09:00:00``,
+  ``ProductMapper(IdentityMapper(), DailyMapper())`` produces
+  ``us|2026-03-10``.
+* ``AllowedKeyMapper`` validates that keys are in a fixed allow-list and
+  passes the key through unchanged if valid.
+  For example, ``AllowedKeyMapper(["us", "eu", "apac"])`` accepts only those
+  region keys and rejects all others.
+
+Example of per-asset mapper configuration and composite-key mapping:
+
+.. code-block:: python
+
+    from airflow.sdk import (
+        Asset,
+        DailyMapper,
+        IdentityMapper,
+        PartitionedAssetTimetable,
+        ProductMapper,
+    )
+
+    regional_sales = Asset(uri="file://incoming/sales/regional.csv", name="regional_sales")
+
+    with DAG(
+        dag_id="aggregate_regional_sales",
+        schedule=PartitionedAssetTimetable(
+            assets=regional_sales,
+            default_partition_mapper=ProductMapper(IdentityMapper(), DailyMapper()),
+        ),
+    ):
+        ...
+
+You can also override mappers for specific upstream assets with
+``partition_mapper_config``:
+
+.. code-block:: python
+
+    from airflow.sdk import Asset, DAG, DailyMapper, IdentityMapper, PartitionedAssetTimetable
+
+    hourly_sales = Asset(uri="file://incoming/sales/hourly.csv", name="hourly_sales")
+    daily_targets = Asset(uri="file://incoming/sales/targets.csv", name="daily_targets")
+
+    with DAG(
+        dag_id="join_sales_and_targets",
+        schedule=PartitionedAssetTimetable(
+            assets=hourly_sales & daily_targets,
+            # Default behavior: map timestamp-like keys to daily keys.
+            default_partition_mapper=DailyMapper(),
+            # Override for assets that already emit daily partition keys.
+            partition_mapper_config={
+                daily_targets: IdentityMapper(),
+            },
+        ),
+    ):
+        ...
+
+If transformed partition keys from all required upstream assets do not align,
+the downstream Dag will not be triggered for that partition.
+
+The same applies when a mapper cannot transform a key. For example, if an
+upstream event has ``partition_key="random-text"`` and the downstream mapping
+uses ``DailyMapper`` (which expects a timestamp-like key), no downstream
+partition match can be produced, so the downstream Dag is not triggered for
+that key.
+
+Inside partitioned Dag runs, access the resolved partition through
+``dag_run.partition_key``.
+
+You can also trigger a DagRun manually with a partition key (for example,
+through the Trigger Dag window in the UI, or through the REST API by
+including ``partition_key`` in the request body):
+
+.. code-block:: bash
+
+    curl -X POST "http://<airflow-host>/api/v2/dags/aggregate_regional_sales/dagRuns" \
+      -H "Content-Type: application/json" \
+      -d '{
+        "logical_date": "2026-03-10T00:00:00Z",
+        "partition_key": "us|2026-03-10T09:00:00"
+      }'
+
+For complete runnable examples, see
+``airflow-core/src/airflow/example_dags/example_asset_partition.py``.

airflow-core/pyproject.toml

@@ -195,12 +195,7 @@ dependencies = [
     "memray>=1.19.0",
 ]
 "gunicorn" = [
-    # Temporary upper bound due to Gunicorn 25.1.0 control-socket deadlock
-    # seen in the Airflow API server path (preload + fork).
-    # Upstream tracking:
-    # - https://github.com/benoitc/gunicorn/issues/3529
-    # - https://github.com/benoitc/gunicorn/pull/3520
-    "gunicorn>=23.0.0,<25.1.0",
+    "gunicorn>=23.0.0,!=25.1.0",
 ]
 "otel" = [
     "opentelemetry-exporter-prometheus>=0.47b0",

airflow-core/src/airflow/api_fastapi/common/db/common.py

@@ -112,11 +112,6 @@ async def paginated_select_async(
     if return_total_entries:
         total_entries = await get_query_count_async(statement, session=session)
 
-    # TODO: Re-enable when permissions are handled. Readable / writable entities,
-    # for instance:
-    # readable_dags = get_auth_manager().get_authorized_dag_ids(user=g.user)
-    # dags_select = dags_select.where(DagModel.dag_id.in_(readable_dags))
-
     statement = apply_filters_to_select(
         statement=statement,
         filters=[order_by, offset, limit],
@@ -171,11 +166,6 @@ def paginated_select(
     if return_total_entries:
         total_entries = get_query_count(statement, session=session)
 
-    # TODO: Re-enable when permissions are handled. Readable / writable entities,
-    # for instance:
-    # readable_dags = get_auth_manager().get_authorized_dag_ids(user=g.user)
-    # dags_select = dags_select.where(DagModel.dag_id.in_(readable_dags))
-
     statement = apply_filters_to_select(statement=statement, filters=[order_by, offset, limit])
 
     return statement, total_entries

airflow-core/src/airflow/api_fastapi/core_api/services/ui/connections.py

@@ -49,10 +49,11 @@ def __init__(
             pass
 
         def __call__(self, form, field):
-            pass
+            """No-op call to satisfy WTForms validator protocol."""
+            return None
 
     class MockEnum:
-        """Mock for wtforms.validators.Optional."""
+        """Mock for wtforms.validators.AnyOf."""
 
         def __init__(self, allowed_values):
             self.allowed_values = allowed_values
@@ -156,23 +157,40 @@ def mock_any_of(allowed_values: list) -> HookMetaService.MockEnum:
                     raise ModuleNotFoundError
             except ModuleNotFoundError:
                 sys.modules[mod_name] = MagicMock()
-        with (
-            mock.patch("wtforms.StringField", HookMetaService.MockStringField),
-            mock.patch("wtforms.fields.StringField", HookMetaService.MockStringField),
-            mock.patch("wtforms.fields.simple.StringField", HookMetaService.MockStringField),
-            mock.patch("wtforms.IntegerField", HookMetaService.MockIntegerField),
-            mock.patch("wtforms.fields.IntegerField", HookMetaService.MockIntegerField),
-            mock.patch("wtforms.PasswordField", HookMetaService.MockPasswordField),
-            mock.patch("wtforms.BooleanField", HookMetaService.MockBooleanField),
-            mock.patch("wtforms.fields.BooleanField", HookMetaService.MockBooleanField),
-            mock.patch("wtforms.fields.simple.BooleanField", HookMetaService.MockBooleanField),
-            mock.patch("flask_babel.lazy_gettext", mock_lazy_gettext),
-            mock.patch("flask_appbuilder.fieldwidgets.BS3TextFieldWidget", HookMetaService.MockAnyWidget),
-            mock.patch("flask_appbuilder.fieldwidgets.BS3TextAreaFieldWidget", HookMetaService.MockAnyWidget),
-            mock.patch("flask_appbuilder.fieldwidgets.BS3PasswordFieldWidget", HookMetaService.MockAnyWidget),
-            mock.patch("wtforms.validators.Optional", HookMetaService.MockOptional),
-            mock.patch("wtforms.validators.any_of", mock_any_of),
-        ):
+
+        # We conditionally inject mock classes for missing dependencies
+        # to ensure `ProvidersManager` can initialize hook connection widgets
+        # without crashing when FAB/WTForms are not installed.
+        if "wtforms.StringField" not in sys.modules:
+            # Only apply mocks if the actual module wasn't loaded beforehand.
+            # This avoids thread-safety issues caused by `unittest.mock.patch` mutating global states.
+            with (
+                mock.patch("wtforms.StringField", HookMetaService.MockStringField),
+                mock.patch("wtforms.fields.StringField", HookMetaService.MockStringField),
+                mock.patch("wtforms.fields.simple.StringField", HookMetaService.MockStringField),
+                mock.patch("wtforms.IntegerField", HookMetaService.MockIntegerField),
+                mock.patch("wtforms.fields.IntegerField", HookMetaService.MockIntegerField),
+                mock.patch("wtforms.PasswordField", HookMetaService.MockPasswordField),
+                mock.patch("wtforms.BooleanField", HookMetaService.MockBooleanField),
+                mock.patch("wtforms.fields.BooleanField", HookMetaService.MockBooleanField),
+                mock.patch("wtforms.fields.simple.BooleanField", HookMetaService.MockBooleanField),
+                mock.patch("flask_babel.lazy_gettext", mock_lazy_gettext),
+                mock.patch("flask_appbuilder.fieldwidgets.BS3TextFieldWidget", HookMetaService.MockAnyWidget),
+                mock.patch(
+                    "flask_appbuilder.fieldwidgets.BS3TextAreaFieldWidget", HookMetaService.MockAnyWidget
+                ),
+                mock.patch(
+                    "flask_appbuilder.fieldwidgets.BS3PasswordFieldWidget", HookMetaService.MockAnyWidget
+                ),
+                mock.patch("wtforms.validators.Optional", HookMetaService.MockOptional),
+                mock.patch("wtforms.validators.any_of", mock_any_of),
+                # Prevent poisoning the global ProvidersManager singleton with mocks
+                mock.patch("airflow.providers_manager.ProvidersManager._instance", None),
+                mock.patch("airflow.providers_manager.ProvidersManager.initialized", return_value=False),
+            ):
+                pm = ProvidersManager()
+                return pm.hooks, pm.connection_form_widgets, pm.field_behaviours  # Will init providers hooks
+        else:
             pm = ProvidersManager()
             return pm.hooks, pm.connection_form_widgets, pm.field_behaviours  # Will init providers hooks
 
@@ -215,6 +233,45 @@ def _convert_extra_fields(form_widgets: dict[str, ConnectionFormWidgetInfo]) ->
             elif isinstance(form_widget.field, HookMetaService.MockBaseField):
                 # legacy path, form widgets created using mocked WTForms fields, need to convert to SerializedParam.dump()
                 hook_widgets[form_widget.field_name] = form_widget.field.param.dump()
+            elif type(form_widget.field).__name__ == "UnboundField":
+                # handle real WTForms fields gracefully without needing mock patches
+                field_class_name = getattr(form_widget.field.field_class, "__name__", "")
+                param_type = "string"
+                param_format = None
+                if field_class_name == "BooleanField":
+                    param_type = "boolean"
+                elif field_class_name == "IntegerField":
+                    param_type = "integer"
+                elif field_class_name == "PasswordField":
+                    param_format = "password"
+
+                label = (
+                    form_widget.field.args[0]
+                    if len(form_widget.field.args) > 0
+                    else form_widget.field.kwargs.get("label")
+                )
+                validators = form_widget.field.kwargs.get("validators", [])
+                description = form_widget.field.kwargs.get("description", "")
+                default = form_widget.field.kwargs.get("default", None)
+
+                enum = {}
+                for v in validators:
+                    if type(v).__name__ == "AnyOf":
+                        enum["enum"] = getattr(v, "values", [])
+
+                types = [param_type, "null"]
+                format_dict = {"format": param_format} if param_format else {}
+
+                param = SerializedParam(
+                    default=default,
+                    title=str(label) if label is not None else None,
+                    description=str(description) if description else None,
+                    source=None,
+                    type=types,
+                    **format_dict,
+                    **enum,
+                ).dump()
+                hook_widgets[form_widget.field_name] = param
             else:
                 log.error("Unknown form widget in %s: %s", hook_key, form_widget)
                 continue

airflow-core/src/airflow/assets/manager.py

@@ -357,7 +357,7 @@ def _queue_dagruns(
         )
 
         non_partitioned_dags = dags_to_queue.difference(partition_dags)  # don't double process
-        if not non_partitioned_dags:
+        if not non_partitioned_dags or partition_key is not None:
             return None
 
         # Possible race condition: if multiple dags or multiple (usually

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

@@ -136,7 +136,6 @@ def _get_latest_runs_stmt(dag_id: str) -> Select:
 
 def _get_latest_runs_stmt_partitioned(dag_id: str) -> Select:
     """Build a select statement to retrieve the last partitioned run for each Dag."""
-    # todo: AIP-76 we should add a partition date field
     latest_run_id = (
         select(DagRun.id)
         .where(
@@ -149,7 +148,11 @@ def _get_latest_runs_stmt_partitioned(dag_id: str) -> Select:
             ),
             DagRun.partition_key.is_not(None),
         )
-        .order_by(DagRun.id.desc())  # todo: AIP-76 add partition date and sort by it here
+        .order_by(
+            DagRun.partition_date.is_(None),
+            DagRun.partition_date.desc(),
+            DagRun.run_after.desc(),
+        )
         .limit(1)
         .scalar_subquery()
     )