This is an automated email from the ASF dual-hosted git repository.
uranusjr pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 819902bb943 Accept datetime in InletEventsAccessor.after() and
before() (#75773)
819902bb943 is described below
commit 819902bb9437a1c9f3a91466770d65544b61c1f4
Author: Tzu-ping Chung <[email protected]>
AuthorDate: Wed Oct 28 15:27:30 2026 +0800
Accept datetime in InletEventsAccessor.after() and before() (#75773)
---
.../docs/authoring-and-scheduling/assets.rst | 7 +++-
task-sdk/src/airflow/sdk/execution_time/context.py | 15 ++++----
.../tests/task_sdk/execution_time/test_context.py | 41 ++++++++++++++++++++++
3 files changed, 56 insertions(+), 7 deletions(-)
diff --git a/airflow-core/docs/authoring-and-scheduling/assets.rst
b/airflow-core/docs/authoring-and-scheduling/assets.rst
index 45d1a5753a5..4082e41ad70 100644
--- a/airflow-core/docs/authoring-and-scheduling/assets.rst
+++ b/airflow-core/docs/authoring-and-scheduling/assets.rst
@@ -269,7 +269,12 @@ You can also filter events by their ``extra`` key-value
pairs:
for event in events:
print(event.extra)
-The ``.extra(key, value)`` method can be chained multiple times; all
conditions are combined with AND logic. Other chaining methods include
``.after(timestamp)``, ``.before(timestamp)``, ``.ascending()``, and
``.limit(n)``.
+The ``.extra(key, value)`` method can be chained multiple times; all conditions
+are combined with AND logic. Other chaining methods include
+``.after(timestamp)``, ``.before(timestamp)``, ``.ascending()``, and
+``.limit(n)``. The timestamp passed to ``.after()`` and ``.before()`` can be an
+ISO 8601 string, or a :class:`~datetime.datetime` object. It must include
+timezone information in either case.
The REST API also supports filtering asset events by extra key-value pairs
using the ``extra`` query parameter (repeat for multiple conditions):
diff --git a/task-sdk/src/airflow/sdk/execution_time/context.py
b/task-sdk/src/airflow/sdk/execution_time/context.py
index 5e5fcc93ee9..5313cbaf382 100644
--- a/task-sdk/src/airflow/sdk/execution_time/context.py
+++ b/task-sdk/src/airflow/sdk/execution_time/context.py
@@ -31,6 +31,7 @@ from uuid import UUID
import attrs
import structlog
+from pydantic import AwareDatetime, TypeAdapter
from airflow.sdk._shared.observability.metrics import stats
from airflow.sdk._shared.state import AssetScope
@@ -157,6 +158,8 @@ NEVER_EXPIRE: timedelta = timedelta.max
_EXTERNAL_STATE_REF_KEY = "__airflow_state_ref__"
+_AWARE_DATETIME_ADAPTER = TypeAdapter(AwareDatetime)
+
def _wrap_external_ref(ref: str) -> dict[str, JsonValue]:
return {_EXTERNAL_STATE_REF_KEY: ref}
@@ -1385,8 +1388,8 @@ class OutletEventAccessors(
@attrs.define(init=False)
class InletEventsAccessor(Sequence["AssetEventResult"]):
- _after: str | datetime | None
- _before: str | datetime | None
+ _after: datetime | None
+ _before: datetime | None
_ascending: bool
_limit: int | None
_partition_key: str | None
@@ -1410,13 +1413,13 @@ class InletEventsAccessor(Sequence["AssetEventResult"]):
self._partition_key_regexp_pattern = None
self._extra: dict[str, str] = {}
- def after(self, after: str) -> Self:
- self._after = after
+ def after(self, after: datetime | str) -> Self:
+ self._after = _AWARE_DATETIME_ADAPTER.validate_python(after)
self._reset_cache()
return self
- def before(self, before: str) -> Self:
- self._before = before
+ def before(self, before: datetime | str) -> Self:
+ self._before = _AWARE_DATETIME_ADAPTER.validate_python(before)
self._reset_cache()
return self
diff --git a/task-sdk/tests/task_sdk/execution_time/test_context.py
b/task-sdk/tests/task_sdk/execution_time/test_context.py
index b48f1eb4d79..2a539aef507 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_context.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_context.py
@@ -1056,6 +1056,47 @@ class TestInletEventAccessor:
name="test_uri", uri="test://test/", after=None, before=None,
limit=10, ascending=False
)
+ @pytest.mark.parametrize(
+ "bound",
+ [
+ pytest.param("2024-01-01T00:00:00Z", id="str"),
+ pytest.param(datetime(2024, 1, 1, tzinfo=UTC), id="datetime"),
+ ],
+ )
+ def test__get_item__with_bounds(self, sample_inlet_evnets_accessor,
mock_supervisor_comms, bound):
+ mock_supervisor_comms.send.side_effect =
[AssetEventsResult(asset_events=[])]
+ expected = datetime(2024, 1, 1, tzinfo=UTC)
+
+
list(sample_inlet_evnets_accessor[TEST_ASSET].after(bound).before(bound))
+
+ mock_supervisor_comms.send.assert_called_once_with(
+ GetAssetEventByAsset(
+ name="test_uri",
+ uri="test://test/",
+ after=expected,
+ before=expected,
+ limit=None,
+ ascending=True,
+ )
+ )
+
+ @pytest.mark.parametrize("method", ["after", "before"])
+ @pytest.mark.parametrize(
+ "bound",
+ [
+ pytest.param(datetime(2024, 1, 1), id="naive-datetime"),
+ pytest.param("2024-01-01T00:00:00", id="naive-str"),
+ pytest.param("not a timestamp", id="malformed-str"),
+ ],
+ )
+ def test_invalid_bound_raises_on_call(
+ self, sample_inlet_evnets_accessor, mock_supervisor_comms, method,
bound
+ ):
+ accessor = sample_inlet_evnets_accessor[TEST_ASSET]
+ with pytest.raises(ValidationError):
+ getattr(accessor, method)(bound)
+ mock_supervisor_comms.send.assert_not_called()
+
def test__get_item__with_extra_filters(self, sample_inlet_evnets_accessor,
mock_supervisor_comms):
asset_event_resp = AssetEventResult(
id=1,
