Add partition_key and partition_key_pattern filters for asset events

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

@@ -247,6 +247,20 @@ Inlet asset events can be read with the ``inlet_events`` accessor in the executi
 
 Each value in the ``inlet_events`` mapping is a sequence-like object that orders past events of a given asset by ``timestamp``, earliest to latest. It supports most of Python's list interface, so you can use ``[-1]`` to access the last event, ``[-2:]`` for the last two, etc. The accessor is lazy and only hits the database when you access items inside it.
 
+The accessor also supports chaining methods to filter events before fetching them. For example, to retrieve only events matching a specific partition key pattern (using database-native regex):
+
+.. code-block:: python
+
+    @task(inlets=[regional_sales])
+    def process_us_sales(*, inlet_events):
+        us_events = inlet_events[regional_sales].partition_key_pattern(r"^us\|")
+        for event in us_events:
+            print(event.extra, event.partition_key)
+
+For an exact partition key match (no regex), use ``.partition_key(value)`` instead. The two are mutually exclusive; setting one clears the other.
+
+Other chaining methods include ``.after(timestamp)``, ``.before(timestamp)``, ``.ascending()``, and ``.limit(n)``.
+
 Dependency between ``@asset``, ``@task``, and classic operators
 ---------------------------------------------------------------
 
@@ -955,6 +969,46 @@ When a runtime run emits exactly one partition key, the producing
 these events the same way as timetable-produced partitions, through
 ``PartitionedAssetTimetable``.
 
+You can also query asset events filtered by partition key using the REST API.
+Two parameters are available:
+
+- ``partition_key`` for **exact match** — uses the B-tree index for fast lookups:
+
+.. code-block:: bash
+
+    curl -G "http://<airflow-host>/api/v2/assets/events" \
+      --data-urlencode "partition_key=us|2026-03-10"
+
+- ``partition_key_pattern`` for **regex filtering** — uses database-native regex
+  (PostgreSQL ``~`` operator, MySQL ``REGEXP``, SQLite ``re.match``):
+
+.. code-block:: bash
+
+    curl -G "http://<airflow-host>/api/v2/assets/events" \
+      --data-urlencode "partition_key_pattern=^us"
+
+These parameters are mutually exclusive; providing both returns a 400 error.
+
+.. note::
+
+    ``partition_key_pattern`` is evaluated by the database's own regular-expression engine, which
+    is a Regular expression Denial of Service (ReDoS) surface. For that reason it is **disabled by
+    default** and must be enabled with ``[api] enable_regexp_query_filters = True``. When enabled,
+    the query runs on PostgreSQL under a bounded ``statement_timeout`` controlled by
+    ``[api] regexp_query_timeout`` (seconds); MySQL bounds regex evaluation with its built-in
+    ``regexp_time_limit``. Prefer the exact-match ``partition_key`` (which uses the B-tree index and
+    is always enabled) whenever a full key is known.
+
+The same filters are available in the ``InletEventsAccessor``:
+
+.. code-block:: python
+
+    # Exact match
+    events = inlet_events[Asset("my_asset")].partition_key("us|2026-03-10").limit(1)
+
+    # Regex pattern
+    events = inlet_events[Asset("my_asset")].partition_key_pattern(r"^us\|2026-03-").limit(10)
+
 Fan-out mappers
 ~~~~~~~~~~~~~~~
 

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                                                  |
 +=========================+==================+===================+==============================================================+
-| ``d2f4e1b3c5a7`` (head) | ``9ff64e1c35d3`` | ``3.3.0``         | Add partition_date to asset_partition_dag_run.               |
+| ``7a98f1b7dbd3`` (head) | ``d2f4e1b3c5a7`` | ``3.4.0``         | Add index on asset_event (asset_id, partition_key).          |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``d2f4e1b3c5a7``        | ``9ff64e1c35d3`` | ``3.3.0``         | Add partition_date to asset_partition_dag_run.               |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``9ff64e1c35d3``        | ``dd5f3a8e2b91`` | ``3.3.0``         | Add indexes on dag_run.created_dag_version_id and            |
 |                         |                  |                   | task_instance.dag_version_id.                                |

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

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import re
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Iterable, Sequence
 from datetime import datetime
@@ -515,6 +516,71 @@ def depends_search(
     return depends_search
 
 
+class _RegexParam(BaseParam[str]):
+    """
+    Filter using database-level regex matching (regexp_match).
+
+    SQLAlchemy's ``regexp_match`` is supported on PostgreSQL (``~`` operator),
+    MySQL/MariaDB (``REGEXP``), and SQLite (via Python ``re.match``).
+    Note: SQLite uses ``re.match`` semantics (anchored at the start of the
+    string), while PostgreSQL uses ``re.search`` (matches anywhere).
+    """
+
+    def __init__(self, attribute: ColumnElement, skip_none: bool = True) -> None:
+        super().__init__(skip_none=skip_none)
+        self.attribute: ColumnElement = attribute
+
+    def to_orm(self, select: Select) -> Select:
+        if self.value is None and self.skip_none:
+            return select
+        return select.where(self.attribute.regexp_match(self.value))
+
+    @classmethod
+    def depends(cls, *args: Any, **kwargs: Any) -> Self:
+        raise NotImplementedError("Use regex_param_factory instead, depends is not implemented.")
+
+
+def _validate_regex_pattern(value: str | None) -> str | None:
+    """Validate that the regex pattern is enabled and syntactically correct."""
+    if value is None:
+        return value
+    if not conf.getboolean("api", "enable_regexp_query_filters"):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Regex query filters are disabled. Set [api] enable_regexp_query_filters = True to use them.",
+        )
+    try:
+        re.compile(value)
+    except re.error as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Invalid regular expression: {e}",
+        )
+    return value
+
+
+_DEFAULT_REGEX_DESCRIPTION = (
+    "Regex filter. Uses database-native regex "
+    "(PostgreSQL ~ operator, MySQL REGEXP, SQLite re.match). "
+    "Note: on SQLite, matching is anchored at the start of the string."
+)
+
+
+def regex_param_factory(
+    attribute: ColumnElement,
+    pattern_name: str,
+    skip_none: bool = True,
+    description: str = _DEFAULT_REGEX_DESCRIPTION,
+) -> Callable[[str | None], _RegexParam]:
+    def depends_regex(
+        value: str | None = Query(alias=pattern_name, default=None, description=description),
+    ) -> _RegexParam:
+        _validate_regex_pattern(value)
+        return _RegexParam(attribute, skip_none).set_value(value)
+
+    return depends_regex
+
+
 def prefix_search_param_factory(
     attribute: ColumnElement,
     prefix_pattern_name: str,
@@ -1584,6 +1650,13 @@ def _transform_ti_states(states: list[str] | None) -> list[TaskInstanceState | N
 QueryAssetAliasNamePrefixPatternSearch = Annotated[
     _PrefixSearchParam, Depends(prefix_search_param_factory(AssetAliasModel.name, "name_prefix_pattern"))
 ]
+QueryAssetEventPartitionKeyFilter = Annotated[
+    FilterParam[str | None],
+    Depends(filter_param_factory(AssetEvent.partition_key, str | None, filter_name="partition_key")),
+]
+QueryAssetEventPartitionKeyRegex = Annotated[
+    _RegexParam, Depends(regex_param_factory(AssetEvent.partition_key, "partition_key_pattern"))
+]
 QueryAssetDagIdPatternSearch = Annotated[
     _DagIdAssetReferenceFilter, Depends(_DagIdAssetReferenceFilter.depends)
 ]

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

@@ -433,6 +433,28 @@ paths:
           - type: integer
           - type: 'null'
           title: Source Map Index
+      - name: partition_key
+        in: query
+        required: false
+        schema:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Partition Key
+      - name: partition_key_pattern
+        in: query
+        required: false
+        schema:
+          anyOf:
+          - type: string
+          - type: 'null'
+          description: 'Regex filter. Uses database-native regex (PostgreSQL ~ operator,
+            MySQL REGEXP, SQLite re.match). Note: on SQLite, matching is anchored
+            at the start of the string.'
+          title: Partition Key Pattern
+        description: 'Regex filter. Uses database-native regex (PostgreSQL ~ operator,
+          MySQL REGEXP, SQLite re.match). Note: on SQLite, matching is anchored at
+          the start of the string.'
       - name: name_pattern
         in: query
         required: false

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

@@ -37,6 +37,8 @@
     QueryAssetAliasNamePatternSearch,
     QueryAssetAliasNamePrefixPatternSearch,
     QueryAssetDagIdPatternSearch,
+    QueryAssetEventPartitionKeyFilter,
+    QueryAssetEventPartitionKeyRegex,
     QueryAssetNamePatternSearch,
     QueryAssetNamePrefixPatternSearch,
     QueryLimit,
@@ -83,6 +85,7 @@
     TaskOutletAssetReference,
 )
 from airflow.typing_compat import Unpack
+from airflow.utils.sqlalchemy import apply_regex_query_timeout
 from airflow.utils.state import DagRunState
 from airflow.utils.types import DagRunTriggeredByType, DagRunType
 
@@ -319,12 +322,24 @@ def get_asset_events(
     source_map_index: Annotated[
         FilterParam[int | None], Depends(filter_param_factory(AssetEvent.source_map_index, int | None))
     ],
+    partition_key: QueryAssetEventPartitionKeyFilter,
+    partition_key_pattern: QueryAssetEventPartitionKeyRegex,
     name_pattern: QueryAssetNamePatternSearch,
     name_prefix_pattern: QueryAssetNamePrefixPatternSearch,
     timestamp_range: Annotated[RangeFilter, Depends(datetime_range_filter_factory("timestamp", AssetEvent))],
     session: SessionDep,
 ) -> AssetEventCollectionResponse:
     """Get asset events."""
+    if partition_key.value is not None and partition_key_pattern.value is not None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="partition_key and partition_key_pattern are mutually exclusive.",
+        )
+
+    if partition_key_pattern.value is not None:
+        # Bound the runtime of the user-supplied regex to guard against ReDoS (PostgreSQL only).
+        apply_regex_query_timeout(session)
+
     base_statement = select(AssetEvent)
     if name_pattern.value or name_prefix_pattern.value:
         base_statement = base_statement.join(AssetModel, AssetEvent.asset_id == AssetModel.id)
@@ -337,6 +352,8 @@ def get_asset_events(
             source_task_id,
             source_run_id,
             source_map_index,
+            partition_key,
+            partition_key_pattern,
             name_pattern,
             name_prefix_pattern,
             timestamp_range,

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

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import re
 from typing import Annotated
 
 from fastapi import APIRouter, HTTPException, Query, status
@@ -29,7 +30,9 @@
     AssetEventResponse,
     AssetEventsResponse,
 )
+from airflow.configuration import conf
 from airflow.models.asset import AssetAliasModel, AssetEvent, AssetModel
+from airflow.utils.sqlalchemy import apply_regex_query_timeout
 
 router = APIRouter(
     responses={
@@ -44,7 +47,7 @@ def _get_asset_events_through_sql_clauses(
 ) -> AssetEventsResponse:
     order_by_clause = AssetEvent.timestamp.asc() if ascending else AssetEvent.timestamp.desc()
     asset_events_query = select(AssetEvent).join(join_clause).where(where_clause).order_by(order_by_clause)
-    if limit:
+    if limit is not None:
         asset_events_query = asset_events_query.limit(limit)
     asset_events = session.scalars(asset_events_query)
     return AssetEventsResponse.model_validate(
@@ -73,6 +76,40 @@ def _get_asset_events_through_sql_clauses(
     )
 
 
+def _validate_partition_key_params(partition_key: str | None, partition_key_pattern: str | None) -> None:
+    """Validate partition_key and partition_key_pattern are not both set, and that the pattern is valid regex."""
+    if partition_key is not None and partition_key_pattern is not None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={
+                "reason": "Mutually exclusive parameters",
+                "message": "partition_key and partition_key_pattern cannot both be provided. "
+                "Use partition_key for exact match or partition_key_pattern for regex filtering.",
+            },
+        )
+    if partition_key_pattern is None:
+        return
+    if not conf.getboolean("api", "enable_regexp_query_filters"):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={
+                "reason": "Regex query filters disabled",
+                "message": "Regex query filters are disabled. "
+                "Set [api] enable_regexp_query_filters = True to use partition_key_pattern.",
+            },
+        )
+    try:
+        re.compile(partition_key_pattern)
+    except re.error as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={
+                "reason": "Invalid regex",
+                "message": f"The partition_key_pattern is not a valid regular expression: {e}",
+            },
+        )
+
+
 @router.get("/by-asset")
 def get_asset_event_by_asset_name_uri(
     name: Annotated[str | None, Query(description="The name of the Asset")],
@@ -82,6 +119,18 @@ def get_asset_event_by_asset_name_uri(
     before: Annotated[UtcDateTime | None, Query(description="The end of the time range")] = None,
     ascending: Annotated[bool, Query(description="Whether to sort results in ascending order")] = True,
     limit: Annotated[int | None, Query(description="The maximum number of results to return")] = None,
+    partition_key: Annotated[
+        str | None,
+        Query(description="Exact partition key filter."),
+    ] = None,
+    partition_key_pattern: Annotated[
+        str | None,
+        Query(
+            description="Partition key regex filter. Uses database-native regex "
+            "(PostgreSQL ~ operator, MySQL REGEXP, SQLite re.match). "
+            "Note: on SQLite, matching is anchored at the start of the string."
+        ),
+    ] = None,
 ) -> AssetEventsResponse:
     if name and uri:
         where_clause = and_(AssetModel.name == name, AssetModel.uri == uri)
@@ -103,6 +152,13 @@ def get_asset_event_by_asset_name_uri(
     if before:
         where_clause = and_(where_clause, AssetEvent.timestamp <= before)
 
+    _validate_partition_key_params(partition_key, partition_key_pattern)
+    if partition_key is not None:
+        where_clause = and_(where_clause, AssetEvent.partition_key == partition_key)
+    elif partition_key_pattern is not None:
+        apply_regex_query_timeout(session)
+        where_clause = and_(where_clause, AssetEvent.partition_key.regexp_match(partition_key_pattern))
+
     return _get_asset_events_through_sql_clauses(
         join_clause=AssetEvent.asset,
         where_clause=where_clause,
@@ -120,13 +176,32 @@ def get_asset_event_by_asset_alias(
     before: Annotated[UtcDateTime | None, Query(description="The end of the time range")] = None,
     ascending: Annotated[bool, Query(description="Whether to sort results in ascending order")] = True,
     limit: Annotated[int | None, Query(description="The maximum number of results to return")] = None,
+    partition_key: Annotated[
+        str | None,
+        Query(description="Exact partition key filter."),
+    ] = None,
+    partition_key_pattern: Annotated[
+        str | None,
+        Query(
+            description="Partition key regex filter. Uses database-native regex "
+            "(PostgreSQL ~ operator, MySQL REGEXP, SQLite re.match). "
+            "Note: on SQLite, matching is anchored at the start of the string."
+        ),
+    ] = None,
 ) -> AssetEventsResponse:
     where_clause = AssetAliasModel.name == name
     if after:
         where_clause = and_(where_clause, AssetEvent.timestamp >= after)
     if before:
         where_clause = and_(where_clause, AssetEvent.timestamp <= before)
 
+    _validate_partition_key_params(partition_key, partition_key_pattern)
+    if partition_key is not None:
+        where_clause = and_(where_clause, AssetEvent.partition_key == partition_key)
+    elif partition_key_pattern is not None:
+        apply_regex_query_timeout(session)
+        where_clause = and_(where_clause, AssetEvent.partition_key.regexp_match(partition_key_pattern))
+
     return _get_asset_events_through_sql_clauses(
         join_clause=AssetEvent.source_aliases,
         where_clause=where_clause,