SameerMesiah97 commented on code in PR #74293:
URL: https://github.com/apache/airflow/pull/74293#discussion_r4188085425
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/s3.py:
##########
@@ -266,3 +270,141 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
await asyncio.sleep(self.polling_period_seconds)
except Exception as e:
yield TriggerEvent({"status": "error", "message": str(e)})
+
+
+class S3KeyUpdateTrigger(BaseEventTrigger):
+ """
+ Fire an event whenever a single S3 object is updated.
+
+ Polls ``head_object`` for ``bucket_key`` and emits an event when the
object's ``ETag``
+ differs from the one the trigger last reported, which makes an upload to
that key usable
+ as a scheduling signal::
+
+ from airflow.sdk import Asset, AssetWatcher
+
+ report = Asset(
+ "daily_report",
+ watchers=[
+ AssetWatcher(
+ name="daily_report_updates",
+ trigger=S3KeyUpdateTrigger(bucket_name="my-bucket",
bucket_key="reports/daily.csv"),
+ )
+ ],
+ )
+
+
+ @dag(schedule=[report])
+ def downstream(): ...
+
+ The last-reported ``ETag`` is persisted in the asset state store, so a
triggerer restart
+ does not re-emit an unchanged object. On the first poll — and after a
restart when no
+ watermark was kept — the current object is itself the first event, so
``previous_etag`` is
+ ``None``. A task that must not run twice for one upload keys on ``etag``.
+
+ The event carries ``bucket_name``, ``bucket_key``, ``etag``,
``previous_etag``,
+ ``last_modified`` (ISO-8601) and ``size``. A missing key is not an error:
the trigger stays
+ silent and keeps polling until the object appears.
+
+ Updates are detected by ``ETag`` (the object's content fingerprint), so
re-uploading
+ byte-identical content does not fire. For a versioned bucket where every
upload must fire,
+ enable bucket versioning and the ``ETag`` still changes per upload.
Review Comment:
Are you sure about that last sentence? Bucket versioning generates a new
`VersionId` per upload, but the ETag can remain unchanged when identical
content is uploaded again. Since this trigger only compares ETags, enabling
versioning would not make every upload fire. I would double-check this just to
be sure.
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/s3.py:
##########
@@ -266,3 +270,141 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
await asyncio.sleep(self.polling_period_seconds)
except Exception as e:
yield TriggerEvent({"status": "error", "message": str(e)})
+
+
+class S3KeyUpdateTrigger(BaseEventTrigger):
+ """
+ Fire an event whenever a single S3 object is updated.
+
+ Polls ``head_object`` for ``bucket_key`` and emits an event when the
object's ``ETag``
+ differs from the one the trigger last reported, which makes an upload to
that key usable
+ as a scheduling signal::
+
+ from airflow.sdk import Asset, AssetWatcher
+
+ report = Asset(
+ "daily_report",
+ watchers=[
+ AssetWatcher(
+ name="daily_report_updates",
+ trigger=S3KeyUpdateTrigger(bucket_name="my-bucket",
bucket_key="reports/daily.csv"),
+ )
+ ],
+ )
+
+
+ @dag(schedule=[report])
+ def downstream(): ...
+
+ The last-reported ``ETag`` is persisted in the asset state store, so a
triggerer restart
+ does not re-emit an unchanged object. On the first poll — and after a
restart when no
+ watermark was kept — the current object is itself the first event, so
``previous_etag`` is
+ ``None``. A task that must not run twice for one upload keys on ``etag``.
+
+ The event carries ``bucket_name``, ``bucket_key``, ``etag``,
``previous_etag``,
+ ``last_modified`` (ISO-8601) and ``size``. A missing key is not an error:
the trigger stays
+ silent and keeps polling until the object appears.
+
+ Updates are detected by ``ETag`` (the object's content fingerprint), so
re-uploading
+ byte-identical content does not fire. For a versioned bucket where every
upload must fire,
+ enable bucket versioning and the ``ETag`` still changes per upload.
+
+ :param bucket_name: Name of the S3 bucket.
+ :param bucket_key: Key of the object to watch.
+ :param aws_conn_id: Reference to the S3 connection.
+ :param poke_interval: Seconds between polls.
+ :param region_name: AWS region for the hook.
+ :param verify: Whether to verify SSL certificates for the S3 connection.
+ :param botocore_config: Configuration dictionary for the underlying
botocore client.
+ :param last_seen_etag: ETag already reported. Leave unset to treat the
current object as
+ the first event.
+ :param hook_params: Additional parameters passed to the hook.
Review Comment:
I would add the default values for each of the optional parameters in this
docstring. Also, I would indicate wether the parameter is 'Optional'.
##########
providers/amazon/tests/unit/amazon/aws/triggers/test_s3.py:
##########
@@ -256,3 +260,201 @@ async def test_run_pending(self, mock_is_keys_unchanged,
mock_client):
# TriggerEvent was not returned
assert task.done() is False
asyncio.get_event_loop().stop()
+
+
+def _head(etag, last_modified=None, size=None):
+ """Build a head_object response pointing at ``etag``. ``None`` means the
key is absent."""
+ if etag is None:
+ return None
+ return {"ETag": etag, "LastModified": last_modified, "ContentLength": size}
+
+
+async def _collect(trigger, count, timeout=10.0):
+ """Pull up to ``count`` payloads off the trigger, giving up after
``timeout``."""
+ from contextlib import aclosing, suppress
+
+ payloads = []
+ generator = trigger.run()
+ async with aclosing(generator):
+
+ async def pump():
+ async for event in generator:
+ payloads.append(event.payload)
+ if len(payloads) >= count:
+ return
+
+ with suppress(asyncio.TimeoutError):
+ await asyncio.wait_for(pump(), timeout=timeout)
+ return payloads
+
+
+@async_mock.patch("airflow.providers.amazon.aws.triggers.s3.S3Hook.get_async_conn")
+class TestS3KeyUpdateTrigger:
+ @staticmethod
+ def _conn(mock_get_async_conn):
+ mock_get_async_conn.return_value.__aenter__.return_value =
async_mock.AsyncMock()
+
+ def test_serialize_round_trip(self, mock_get_async_conn):
+ trigger = S3KeyUpdateTrigger(
+ bucket_name="my-bucket",
+ bucket_key="reports/daily.csv",
+ aws_conn_id="my_conn",
+ poke_interval=5,
+ region_name="eu-west-1",
+ )
+ classpath, kwargs = trigger.serialize()
+
+ assert classpath ==
"airflow.providers.amazon.aws.triggers.s3.S3KeyUpdateTrigger"
+ assert kwargs == {
+ "bucket_name": "my-bucket",
+ "bucket_key": "reports/daily.csv",
+ "aws_conn_id": "my_conn",
+ "poke_interval": 5,
+ "region_name": "eu-west-1",
+ "verify": None,
+ "botocore_config": None,
+ "last_seen_etag": None,
+ "hook_params": {},
+ }
Review Comment:
Nit: I would use non-default params for verify, botocore_config,
last_seen_etag and hook_params,
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/s3.py:
##########
@@ -266,3 +270,141 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
await asyncio.sleep(self.polling_period_seconds)
except Exception as e:
yield TriggerEvent({"status": "error", "message": str(e)})
+
+
+class S3KeyUpdateTrigger(BaseEventTrigger):
+ """
+ Fire an event whenever a single S3 object is updated.
+
+ Polls ``head_object`` for ``bucket_key`` and emits an event when the
object's ``ETag``
+ differs from the one the trigger last reported, which makes an upload to
that key usable
+ as a scheduling signal::
+
+ from airflow.sdk import Asset, AssetWatcher
+
+ report = Asset(
+ "daily_report",
+ watchers=[
+ AssetWatcher(
+ name="daily_report_updates",
+ trigger=S3KeyUpdateTrigger(bucket_name="my-bucket",
bucket_key="reports/daily.csv"),
+ )
+ ],
+ )
+
+
+ @dag(schedule=[report])
+ def downstream(): ...
+
Review Comment:
Not sure if this belongs here rather than the provider documentation. In
general, I feel that this docstring is too long and detailed. Perhaps you could
just keep these 4 lines:
```
Fire an event whenever a single S3 object is updated.
Polls ``head_object`` for ``bucket_key`` and emits an event when the
object's ``ETag``
differs from the one the trigger last reported, which makes an upload to
that key usable
as a scheduling signal
```
And sprinkle the remaining context as comments throughout the file.
##########
providers/amazon/tests/unit/amazon/aws/triggers/test_s3.py:
##########
@@ -256,3 +260,201 @@ async def test_run_pending(self, mock_is_keys_unchanged,
mock_client):
# TriggerEvent was not returned
assert task.done() is False
asyncio.get_event_loop().stop()
+
+
+def _head(etag, last_modified=None, size=None):
+ """Build a head_object response pointing at ``etag``. ``None`` means the
key is absent."""
+ if etag is None:
+ return None
+ return {"ETag": etag, "LastModified": last_modified, "ContentLength": size}
+
+
+async def _collect(trigger, count, timeout=10.0):
+ """Pull up to ``count`` payloads off the trigger, giving up after
``timeout``."""
+ from contextlib import aclosing, suppress
+
+ payloads = []
+ generator = trigger.run()
+ async with aclosing(generator):
+
+ async def pump():
+ async for event in generator:
+ payloads.append(event.payload)
+ if len(payloads) >= count:
+ return
+
+ with suppress(asyncio.TimeoutError):
+ await asyncio.wait_for(pump(), timeout=timeout)
+ return payloads
+
+
+@async_mock.patch("airflow.providers.amazon.aws.triggers.s3.S3Hook.get_async_conn")
+class TestS3KeyUpdateTrigger:
+ @staticmethod
+ def _conn(mock_get_async_conn):
+ mock_get_async_conn.return_value.__aenter__.return_value =
async_mock.AsyncMock()
+
+ def test_serialize_round_trip(self, mock_get_async_conn):
+ trigger = S3KeyUpdateTrigger(
+ bucket_name="my-bucket",
+ bucket_key="reports/daily.csv",
+ aws_conn_id="my_conn",
+ poke_interval=5,
+ region_name="eu-west-1",
+ )
+ classpath, kwargs = trigger.serialize()
+
+ assert classpath ==
"airflow.providers.amazon.aws.triggers.s3.S3KeyUpdateTrigger"
+ assert kwargs == {
+ "bucket_name": "my-bucket",
+ "bucket_key": "reports/daily.csv",
+ "aws_conn_id": "my_conn",
+ "poke_interval": 5,
+ "region_name": "eu-west-1",
+ "verify": None,
+ "botocore_config": None,
+ "last_seen_etag": None,
+ "hook_params": {},
+ }
+ assert S3KeyUpdateTrigger(**kwargs).serialize() == (classpath, kwargs)
+
+ @pytest.mark.asyncio
+ @async_mock.patch(
+
"airflow.providers.amazon.aws.triggers.s3.S3Hook.get_head_object_async",
+ new_callable=async_mock.AsyncMock,
+ )
+ async def test_emits_current_object_on_first_poll(self, mock_head,
mock_get_async_conn):
+ """With no watermark the current object is itself the first event."""
+ self._conn(mock_get_async_conn)
+ mock_head.side_effect = [_head('"aaa"')]
+ trigger = S3KeyUpdateTrigger(bucket_name="b", bucket_key="k",
poke_interval=0.01)
+
+ payloads = await _collect(trigger, 1)
+
+ assert [(p["etag"], p["previous_etag"]) for p in payloads] ==
[('"aaa"', None)]
+
+ @pytest.mark.asyncio
+ @async_mock.patch(
+
"airflow.providers.amazon.aws.triggers.s3.S3Hook.get_head_object_async",
+ new_callable=async_mock.AsyncMock,
+ )
+ async def test_silent_while_etag_unchanged(self, mock_head,
mock_get_async_conn):
+ """An object that has not changed must not schedule anything."""
+ self._conn(mock_get_async_conn)
+ mock_head.return_value = _head('"aaa"')
+ trigger = S3KeyUpdateTrigger(
+ bucket_name="b", bucket_key="k", poke_interval=0.01,
last_seen_etag='"aaa"'
+ )
+
+ payloads = await _collect(trigger, 1, timeout=0.2)
+
+ assert payloads == []
+
Review Comment:
I would add this assert to ensure that polling actually happened:
`mock_head.assert_awaited()`
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/s3.py:
##########
@@ -266,3 +270,141 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
await asyncio.sleep(self.polling_period_seconds)
except Exception as e:
yield TriggerEvent({"status": "error", "message": str(e)})
+
+
+class S3KeyUpdateTrigger(BaseEventTrigger):
+ """
+ Fire an event whenever a single S3 object is updated.
+
+ Polls ``head_object`` for ``bucket_key`` and emits an event when the
object's ``ETag``
+ differs from the one the trigger last reported, which makes an upload to
that key usable
+ as a scheduling signal::
+
+ from airflow.sdk import Asset, AssetWatcher
+
+ report = Asset(
+ "daily_report",
+ watchers=[
+ AssetWatcher(
+ name="daily_report_updates",
+ trigger=S3KeyUpdateTrigger(bucket_name="my-bucket",
bucket_key="reports/daily.csv"),
+ )
+ ],
+ )
+
+
+ @dag(schedule=[report])
+ def downstream(): ...
+
+ The last-reported ``ETag`` is persisted in the asset state store, so a
triggerer restart
+ does not re-emit an unchanged object. On the first poll — and after a
restart when no
+ watermark was kept — the current object is itself the first event, so
``previous_etag`` is
+ ``None``. A task that must not run twice for one upload keys on ``etag``.
+
+ The event carries ``bucket_name``, ``bucket_key``, ``etag``,
``previous_etag``,
+ ``last_modified`` (ISO-8601) and ``size``. A missing key is not an error:
the trigger stays
+ silent and keeps polling until the object appears.
+
+ Updates are detected by ``ETag`` (the object's content fingerprint), so
re-uploading
+ byte-identical content does not fire. For a versioned bucket where every
upload must fire,
+ enable bucket versioning and the ``ETag`` still changes per upload.
+
+ :param bucket_name: Name of the S3 bucket.
+ :param bucket_key: Key of the object to watch.
+ :param aws_conn_id: Reference to the S3 connection.
+ :param poke_interval: Seconds between polls.
+ :param region_name: AWS region for the hook.
+ :param verify: Whether to verify SSL certificates for the S3 connection.
+ :param botocore_config: Configuration dictionary for the underlying
botocore client.
+ :param last_seen_etag: ETag already reported. Leave unset to treat the
current object as
+ the first event.
+ :param hook_params: Additional parameters passed to the hook.
+ """
+
+ def __init__(
+ self,
+ *,
+ bucket_name: str,
+ bucket_key: str,
+ aws_conn_id: str | None = "aws_default",
+ poke_interval: float = 60,
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ last_seen_etag: str | None = None,
+ hook_params: dict | None = None,
+ ) -> None:
+ super().__init__()
+ self.bucket_name = bucket_name
+ self.bucket_key = bucket_key
+ self.aws_conn_id = aws_conn_id
+ self.poke_interval = poke_interval
+ self.region_name = region_name
+ self.verify = verify
+ self.botocore_config = botocore_config
+ self.last_seen_etag = last_seen_etag
+ self.hook_params = hook_params or {}
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serialize S3KeyUpdateTrigger arguments and classpath."""
+ return (
+ "airflow.providers.amazon.aws.triggers.s3.S3KeyUpdateTrigger",
+ {
+ "bucket_name": self.bucket_name,
+ "bucket_key": self.bucket_key,
+ "aws_conn_id": self.aws_conn_id,
+ "poke_interval": self.poke_interval,
+ "region_name": self.region_name,
+ "verify": self.verify,
+ "botocore_config": self.botocore_config,
+ "last_seen_etag": self.last_seen_etag,
+ "hook_params": self.hook_params,
+ },
+ )
+
+ @cached_property
+ def hook(self) -> S3Hook:
+ return S3Hook(
+ aws_conn_id=self.aws_conn_id,
+ region_name=self.region_name,
+ verify=self.verify,
+ config=self.botocore_config,
+ **self.hook_params,
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Poll the object and emit an event each time its ETag changes."""
+ # serialize() is captured once when the trigger row is created, so a
value mutated on
+ # self is lost when the triggerer restarts and the current object
would be re-emitted as
+ # an update. The watermark survives that; the kwarg only seeds the
first run.
+ store = getattr(self, "asset_state_store", None)
+ if store is not None:
+ stored = await asyncio.to_thread(store.get, WATERMARK_KEY)
+ if stored is not None:
+ self.last_seen_etag = stored
+
+ async with await self.hook.get_async_conn() as client:
+ while True:
+ head = await self.hook.get_head_object_async(
+ client=client, key=self.bucket_key,
bucket_name=self.bucket_name
+ )
+ if head is not None:
+ etag = head.get("ETag")
+ if etag is not None and etag != self.last_seen_etag:
+ previous, self.last_seen_etag = self.last_seen_etag,
etag
+ if store is not None:
+ await asyncio.to_thread(store.set, WATERMARK_KEY,
etag)
+ last_modified = head.get("LastModified")
+ yield TriggerEvent(
+ {
+ "bucket_name": self.bucket_name,
+ "bucket_key": self.bucket_key,
+ "etag": etag,
+ "previous_etag": previous,
+ "last_modified": last_modified.isoformat()
+ if last_modified is not None
+ else None,
+ "size": head.get("ContentLength"),
+ }
+ )
+ await asyncio.sleep(self.poke_interval)
Review Comment:
I think this could be cleaner. Please see the below:
```
etag = head.get("ETag") if head is not None else None
if etag is not None and etag != self.last_seen_etag:
previous_etag = self.last_seen_etag
if store is not None:
await asyncio.to_thread(store.set, WATERMARK_KEY, etag)
self.last_seen_etag = etag
last_modified = head.get("LastModified")
yield TriggerEvent(
{
"bucket_name": self.bucket_name,
"bucket_key": self.bucket_key,
"etag": etag,
"previous_etag": previous_etag,
"last_modified": (
last_modified.isoformat() if last_modified is not None else
None
),
"size": head.get("ContentLength"),
}
)
await asyncio.sleep(self.poke_interval)
```
Double-check the logic above to make sure it matches what you intend.
--
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]
