Copilot commented on code in PR #64610:
URL: https://github.com/apache/airflow/pull/64610#discussion_r3070030871
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -611,6 +615,18 @@ def limit(self, limit: int) -> Self:
self._reset_cache()
return self
+ def partition_key(self, key: str) -> Self:
+ """Filter by exact partition key match."""
+ self._partition_key = key
+ self._reset_cache()
+ return self
+
+ def partition_key_pattern(self, pattern: str) -> Self:
+ """Filter by partition key regex pattern."""
+ self._partition_key_pattern = pattern
Review Comment:
`partition_key` and `partition_key_pattern` are mutually exclusive at the
API level, but the accessor allows both to be set (e.g. chaining
`.partition_key(...).partition_key_pattern(...)`), which will always result in
a runtime 400 when the request is made. Consider clearing the other field when
one is set, or raising a ValueError when attempting to set both, so callers
fail fast and the fluent API is harder to misuse.
```suggestion
self._partition_key = key
self._partition_key_pattern = None
self._reset_cache()
return self
def partition_key_pattern(self, pattern: str) -> Self:
"""Filter by partition key regex pattern."""
self._partition_key_pattern = pattern
self._partition_key = None
```
##########
task-sdk/tests/task_sdk/execution_time/test_supervisor.py:
##########
@@ -2055,6 +2071,88 @@ class RequestTestCase:
),
test_id="get_asset_events_by_asset_alias_with_filters",
),
+ RequestTestCase(
+ message=GetAssetEventByAsset(
+ uri="s3://bucket/obj",
+ name="test",
+ partition_key="^us\\|2024-",
+ ),
+ expected_body={
Review Comment:
These new test cases pass regex-like values (e.g. starting with `^`) via the
`partition_key` field. `partition_key` is documented/implemented as an
exact-match filter, while regex should go through `partition_key_pattern`. This
makes the test exercise the wrong query param and could mask real regressions.
Use `partition_key_pattern` for regex patterns here (or change the value to an
actual exact partition key if that’s what you intend to test).
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -683,8 +685,12 @@ def get(
if before:
common_params["before"] = before.isoformat()
common_params["ascending"] = ascending
- if limit:
+ if limit is not None:
common_params["limit"] = limit
+ if partition_key is not None:
+ common_params["partition_key"] = partition_key
+ if partition_key_pattern is not None:
+ common_params["partition_key_pattern"] = partition_key_pattern
Review Comment:
The client now accepts both `partition_key` and `partition_key_pattern` but
doesn’t enforce mutual exclusivity. Since the server returns a 400 when both
are set, consider validating in the client (e.g. raise ValueError) to fail fast
and avoid an unnecessary network call.
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_asset_events.py:
##########
@@ -521,3 +521,335 @@ def test_get_by_asset(self, client):
},
]
}
+
+
+class TestGetAssetEventByAssetPartitionKey:
+ """Tests for partition_key_pattern regex filter on execution API.
+
+ All patterns use ^-anchored regex so they work consistently across
+ PostgreSQL (~), MySQL (REGEXP), and SQLite (re.match).
Review Comment:
The class docstring claims “All patterns use ^-anchored regex”, but the
tests include patterns like `.*\|2024-01-01$` that are not `^`-anchored. Update
the docstring to reflect the actual patterns (e.g. explain that patterns are
written to be SQLite `re.match`-compatible, not necessarily `^`-anchored).
```suggestion
Patterns are written to work consistently across PostgreSQL (~),
MySQL (REGEXP), and SQLite (re.match). They are not necessarily
^-anchored; some cases use explicit prefixes such as ``.*`` to
achieve SQLite ``re.match``-compatible behavior.
```
##########
task-sdk/tests/task_sdk/api/test_client.py:
##########
@@ -1164,6 +1164,97 @@ def handle_request(request: httpx.Request) ->
httpx.Response:
assert result.asset_events[0].asset.name == "this_asset"
assert result.asset_events[0].asset.uri == "s3://bucket/key"
+ def test_partition_key_exact_match_param_passed(self):
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ params = request.url.params
+ assert params.get("partition_key") == "2024-01-01"
+ assert "partition_key_pattern" not in params
+ return httpx.Response(
+ status_code=200,
+ json={
+ "asset_events": [
+ {
+ "id": 1,
+ "asset": {
+ "name": "this_asset",
+ "uri": "s3://bucket/key",
+ "group": "asset",
+ },
+ "created_dagruns": [],
+ "timestamp": "2023-01-01T00:00:00Z",
+ "partition_key": "2024-01-01",
+ }
+ ]
+ },
+ )
+
+ client = make_client(httpx.MockTransport(handle_request))
+ result = client.asset_events.get(name="this_asset",
partition_key="2024-01-01")
+
+ assert isinstance(result, AssetEventsResponse)
+ assert len(result.asset_events) == 1
+
+ def test_partition_key_pattern_param_passed(self):
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ params = request.url.params
+ assert params.get("partition_key_pattern") == "^2024-01-"
+ assert "partition_key" not in params
+ return httpx.Response(
+ status_code=200,
+ json={"asset_events": []},
+ )
+
+ client = make_client(httpx.MockTransport(handle_request))
+ result = client.asset_events.get(name="this_asset",
partition_key_pattern="^2024-01-")
+
+ assert isinstance(result, AssetEventsResponse)
+ assert len(result.asset_events) == 0
+
+ def test_partition_key_pattern_with_other_params(self):
+ from datetime import datetime, timezone
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
Review Comment:
New inline import inside the test function (`from datetime import datetime,
timezone`). Project guidelines prefer imports at module top-level (unless
there’s a specific circular/lazy-load reason). Please move this import to the
top of the file.
##########
airflow-core/docs/authoring-and-scheduling/assets.rst:
##########
@@ -245,6 +245,18 @@ 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(r"^us\|")
Review Comment:
This example describes regex filtering but calls `.partition_key(...)`,
which is the exact-match filter. Switch the example to
`.partition_key_pattern(...)` (or adjust the surrounding text) so the docs
match the actual API.
```suggestion
us_events =
inlet_events[regional_sales].partition_key_pattern(r"^us\|")
```
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py:
##########
@@ -1042,6 +1042,154 @@ def test_should_mask_sensitive_extra(self, test_client,
session):
}
+class TestGetAssetEventsPartitionKeyRegex(TestAssets):
+ """Tests for partition_key_pattern regex filter on GET /assets/events.
+
+ All patterns use ^-anchored regex so they work consistently across
+ PostgreSQL (~), MySQL (REGEXP), and SQLite (re.match).
Review Comment:
This docstring claims “All patterns use ^-anchored regex”, but the
parametrized cases include patterns like `.*\|2024-01-01$` that are not
`^`-anchored. Please adjust the docstring to match the test data (or update the
patterns).
```suggestion
The parametrized cases use regex patterns chosen to behave consistently
across PostgreSQL (~), MySQL (REGEXP), and SQLite (re.match), including
both anchored and unanchored expressions where appropriate.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
