ramitkataria commented on code in PR #74293:
URL: https://github.com/apache/airflow/pull/74293#discussion_r4354400551
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/s3.py:
##########
@@ -266,3 +273,138 @@ 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.
+
+ :param bucket_name: Name of the S3 bucket.
+ :param bucket_key: Key of the object to watch.
+ :param aws_conn_id: Optional. Reference to the S3 connection. Default:
``"aws_default"``.
+ :param poke_interval: Optional. Seconds between polls. Default: ``60``.
+ :param region_name: Optional. AWS region for the hook. Default: ``None``.
+ :param verify: Optional. Whether to verify SSL certificates for the S3
connection.
+ Default: ``None``.
+ :param botocore_config: Optional. Configuration dictionary for the
underlying botocore
+ client. Default: ``None``.
+ :param last_seen_etag: Optional. ETag already reported. Leave unset to
treat the current
+ object as the first event. Default: ``None``.
+ :param hook_params: Optional. Additional parameters passed to the hook.
Default: ``None``.
+ """
+
+ 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,
+ )
+
+ @cached_property
+ def _watermark_key(self) -> str:
+ identity = json.dumps(
+ {
+ "bucket_name": self.bucket_name,
+ "bucket_key": self.bucket_key,
+ "aws_conn_id": self.aws_conn_id,
+ "region_name": self.region_name,
+ },
+ sort_keys=True,
+ separators=(",", ":"),
+ ).encode()
+ return
f"{_WATERMARK_KEY_PREFIX}:{hashlib.sha256(identity).hexdigest()}"
+
+ 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, self._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_object returns None while the key does not exist yet;
keep polling until
Review Comment:
As far I know, without the `s3:ListBucket` permission, this would return 403
and the error would propogate instead of returning none. I think we should
document that behaviour
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/s3.py:
##########
@@ -266,3 +273,138 @@ 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.
+
+ :param bucket_name: Name of the S3 bucket.
+ :param bucket_key: Key of the object to watch.
+ :param aws_conn_id: Optional. Reference to the S3 connection. Default:
``"aws_default"``.
+ :param poke_interval: Optional. Seconds between polls. Default: ``60``.
+ :param region_name: Optional. AWS region for the hook. Default: ``None``.
+ :param verify: Optional. Whether to verify SSL certificates for the S3
connection.
+ Default: ``None``.
+ :param botocore_config: Optional. Configuration dictionary for the
underlying botocore
+ client. Default: ``None``.
+ :param last_seen_etag: Optional. ETag already reported. Leave unset to
treat the current
+ object as the first event. Default: ``None``.
+ :param hook_params: Optional. Additional parameters passed to the hook.
Default: ``None``.
+ """
+
+ 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,
+ )
+
+ @cached_property
+ def _watermark_key(self) -> str:
+ identity = json.dumps(
+ {
+ "bucket_name": self.bucket_name,
+ "bucket_key": self.bucket_key,
+ "aws_conn_id": self.aws_conn_id,
+ "region_name": self.region_name,
+ },
+ sort_keys=True,
+ separators=(",", ":"),
+ ).encode()
+ return
f"{_WATERMARK_KEY_PREFIX}:{hashlib.sha256(identity).hexdigest()}"
+
+ 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:
Review Comment:
What would the behaviour be if it's an old version of airflow where the
store doesn't exist?
--
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]
