You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "phanikumv (via GitHub)" <gi...@apache.org> on 2023/04/13 09:06:34 UTC
[GitHub] [airflow] phanikumv opened a new pull request, #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
phanikumv opened a new pull request, #30618:
URL: https://github.com/apache/airflow/pull/30618
This PR donates the `GCSObjectsWithPrefixExistenceSensorAsync` from [astronomer-providers](https://github.com/astronomer/astronomer-providers/pulls) to OSS
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1175553269
##########
docs/apache-airflow-providers-google/operators/cloud/gcs.rst:
##########
@@ -199,8 +199,20 @@ Use the :class:`~airflow.providers.google.cloud.sensors.gcs.GCSObjectsWithPrefix
:start-after: [START howto_sensor_object_with_prefix_exists_task]
:end-before: [END howto_sensor_object_with_prefix_exists_task]
+You can set the ``deferrable`` param to True if you want this sensor to run asynchronously - leading to efficient
+utilization of resources in your Airflow deployment. However the triggerer component needs to be up and running
Review Comment:
done
##########
docs/apache-airflow-providers-google/operators/cloud/gcs.rst:
##########
@@ -199,8 +199,20 @@ Use the :class:`~airflow.providers.google.cloud.sensors.gcs.GCSObjectsWithPrefix
:start-after: [START howto_sensor_object_with_prefix_exists_task]
:end-before: [END howto_sensor_object_with_prefix_exists_task]
+You can set the ``deferrable`` param to True if you want this sensor to run asynchronously - leading to efficient
Review Comment:
done
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1179065833
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -200,3 +200,76 @@ async def _is_blob_updated_after(
if blob_updated_time > target_date:
return True, {"status": "success", "message": "success"}
return False, {"status": "pending", "message": "pending"}
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ Looks for objects in bucket matching a prefix.
+ If none found, sleep for interval and check again. Otherwise, return matches.
+
+ :param bucket: the bucket in the google cloud storage where the objects are residing.
+ :param prefix: The prefix of the blob_names to match in the Google cloud storage bucket
+ :param google_cloud_conn_id: reference to the Google Connection
+ :param poke_interval: polling period in seconds to check
+ """
+
+ def __init__(
+ self,
+ bucket: str,
+ prefix: str,
+ poke_interval: float,
+ google_cloud_conn_id: str,
+ hook_params: dict[str, Any],
+ ):
+ super().__init__(
+ bucket=bucket,
+ object_name=prefix,
+ poke_interval=poke_interval,
+ google_cloud_conn_id=google_cloud_conn_id,
+ hook_params=hook_params,
+ )
+ self.prefix = prefix
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serializes GCSPrefixBlobTrigger arguments and classpath."""
+ return (
+ "airflow.providers.google.cloud.triggers.gcs.GCSPrefixBlobTrigger",
+ {
+ "bucket": self.bucket,
+ "prefix": self.prefix,
+ "poke_interval": self.poke_interval,
+ "google_cloud_conn_id": self.google_cloud_conn_id,
+ "hook_params": self.hook_params,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Simple loop until the matches are found for the given prefix on the bucket."""
+ try:
+ hook = self._get_async_hook()
+ while True:
+ res = await self._list_blobs_with_prefix(
+ hook=hook, bucket_name=self.bucket, prefix=self.prefix
+ )
+ if len(res) > 0:
+ yield TriggerEvent(
+ {"status": "success", "message": "Successfully completed", "matches": res}
+ )
+ await asyncio.sleep(self.poke_interval)
+ except Exception as e:
+ yield TriggerEvent({"status": "error", "message": str(e)})
+ return
+
+ async def _list_blobs_with_prefix(self, hook: GCSAsyncHook, bucket_name: str, prefix: str) -> list[str]:
+ """
+ Returns names of blobs which match the given prefix for a given bucket.
+
+ :param bucket_name: The Google Cloud Storage bucket where the object is.
Review Comment:
added
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1175551118
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -97,3 +97,76 @@ async def _object_exists(self, hook: GCSAsyncHook, bucket_name: str, object_name
if object_response:
return "success"
return "pending"
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ A trigger that fires and it looks for all the objects in the given bucket
+ which matches the given prefix if not found sleep for certain interval and checks again.
Review Comment:
done
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1179047448
##########
airflow/providers/google/cloud/sensors/gcs.py:
##########
@@ -307,10 +314,35 @@ def poke(self, context: Context) -> bool:
self._matches = hook.list(self.bucket, prefix=self.prefix)
return bool(self._matches)
- def execute(self, context: Context) -> list[str]:
+ def execute(self, context: Context):
"""Overridden to allow matches to be passed"""
- super().execute(context)
- return self._matches
+ if not self.deferrable:
+ super().execute(context)
+ return self._matches
+ else:
+ self.defer(
+ timeout=timedelta(seconds=self.timeout),
+ trigger=GCSPrefixBlobTrigger(
+ bucket=self.bucket,
+ prefix=self.prefix,
+ poke_interval=self.poke_interval,
+ google_cloud_conn_id=self.google_cloud_conn_id,
+ hook_params={
+ "impersonation_chain": self.impersonation_chain,
+ },
+ ),
+ method_name="execute_complete",
+ )
+
+ def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]:
+ """
+ Callback for when the trigger fires; returns immediately.
+ Relies on trigger to throw a success event
+ """
+ self.log.info("Checking for existence of object: %s, %s", self.bucket, self.prefix)
Review Comment:
changed it
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1179059814
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -200,3 +200,76 @@ async def _is_blob_updated_after(
if blob_updated_time > target_date:
return True, {"status": "success", "message": "success"}
return False, {"status": "pending", "message": "pending"}
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ Looks for objects in bucket matching a prefix.
+ If none found, sleep for interval and check again. Otherwise, return matches.
+
+ :param bucket: the bucket in the google cloud storage where the objects are residing.
+ :param prefix: The prefix of the blob_names to match in the Google cloud storage bucket
+ :param google_cloud_conn_id: reference to the Google Connection
+ :param poke_interval: polling period in seconds to check
+ """
+
+ def __init__(
+ self,
+ bucket: str,
+ prefix: str,
+ poke_interval: float,
+ google_cloud_conn_id: str,
+ hook_params: dict[str, Any],
+ ):
+ super().__init__(
+ bucket=bucket,
+ object_name=prefix,
+ poke_interval=poke_interval,
+ google_cloud_conn_id=google_cloud_conn_id,
+ hook_params=hook_params,
+ )
+ self.prefix = prefix
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serializes GCSPrefixBlobTrigger arguments and classpath."""
+ return (
+ "airflow.providers.google.cloud.triggers.gcs.GCSPrefixBlobTrigger",
+ {
+ "bucket": self.bucket,
+ "prefix": self.prefix,
+ "poke_interval": self.poke_interval,
+ "google_cloud_conn_id": self.google_cloud_conn_id,
+ "hook_params": self.hook_params,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Simple loop until the matches are found for the given prefix on the bucket."""
+ try:
+ hook = self._get_async_hook()
+ while True:
+ res = await self._list_blobs_with_prefix(
Review Comment:
made the change
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on PR #30618:
URL: https://github.com/apache/airflow/pull/30618#issuecomment-1521707377
@potiuk @dstandish could you please merge it if it looks good.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish merged pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish merged PR #30618:
URL: https://github.com/apache/airflow/pull/30618
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1178249248
##########
airflow/providers/google/cloud/sensors/gcs.py:
##########
@@ -307,10 +314,35 @@ def poke(self, context: Context) -> bool:
self._matches = hook.list(self.bucket, prefix=self.prefix)
return bool(self._matches)
- def execute(self, context: Context) -> list[str]:
+ def execute(self, context: Context):
"""Overridden to allow matches to be passed"""
- super().execute(context)
- return self._matches
+ if not self.deferrable:
+ super().execute(context)
+ return self._matches
+ else:
+ self.defer(
+ timeout=timedelta(seconds=self.timeout),
+ trigger=GCSPrefixBlobTrigger(
+ bucket=self.bucket,
+ prefix=self.prefix,
+ poke_interval=self.poke_interval,
+ google_cloud_conn_id=self.google_cloud_conn_id,
+ hook_params={
+ "impersonation_chain": self.impersonation_chain,
+ },
+ ),
+ method_name="execute_complete",
+ )
+
+ def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]:
+ """
+ Callback for when the trigger fires; returns immediately.
+ Relies on trigger to throw a success event
+ """
+ self.log.info("Checking for existence of object: %s, %s", self.bucket, self.prefix)
Review Comment:
```suggestion
self.log.info("Checking for existence of object: %s, %s", self.bucket, self.prefix)
```
sorry to nit pick this but here you are not checking -- the checking has already happened (in the trigger) and you're resuming from trigger and checking status
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -200,3 +200,76 @@ async def _is_blob_updated_after(
if blob_updated_time > target_date:
return True, {"status": "success", "message": "success"}
return False, {"status": "pending", "message": "pending"}
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ Looks for objects in bucket matching a prefix.
+ If none found, sleep for interval and check again. Otherwise, return matches.
+
+ :param bucket: the bucket in the google cloud storage where the objects are residing.
+ :param prefix: The prefix of the blob_names to match in the Google cloud storage bucket
+ :param google_cloud_conn_id: reference to the Google Connection
+ :param poke_interval: polling period in seconds to check
+ """
+
+ def __init__(
+ self,
+ bucket: str,
+ prefix: str,
+ poke_interval: float,
+ google_cloud_conn_id: str,
+ hook_params: dict[str, Any],
+ ):
+ super().__init__(
+ bucket=bucket,
+ object_name=prefix,
+ poke_interval=poke_interval,
+ google_cloud_conn_id=google_cloud_conn_id,
+ hook_params=hook_params,
+ )
+ self.prefix = prefix
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serializes GCSPrefixBlobTrigger arguments and classpath."""
+ return (
+ "airflow.providers.google.cloud.triggers.gcs.GCSPrefixBlobTrigger",
+ {
+ "bucket": self.bucket,
+ "prefix": self.prefix,
+ "poke_interval": self.poke_interval,
+ "google_cloud_conn_id": self.google_cloud_conn_id,
+ "hook_params": self.hook_params,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Simple loop until the matches are found for the given prefix on the bucket."""
+ try:
+ hook = self._get_async_hook()
+ while True:
+ res = await self._list_blobs_with_prefix(
+ hook=hook, bucket_name=self.bucket, prefix=self.prefix
+ )
+ if len(res) > 0:
+ yield TriggerEvent(
+ {"status": "success", "message": "Successfully completed", "matches": res}
+ )
+ await asyncio.sleep(self.poke_interval)
+ except Exception as e:
+ yield TriggerEvent({"status": "error", "message": str(e)})
+ return
+
+ async def _list_blobs_with_prefix(self, hook: GCSAsyncHook, bucket_name: str, prefix: str) -> list[str]:
+ """
+ Returns names of blobs which match the given prefix for a given bucket.
+
+ :param bucket_name: The Google Cloud Storage bucket where the object is.
Review Comment:
docstring missing param `hook`
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -200,3 +200,76 @@ async def _is_blob_updated_after(
if blob_updated_time > target_date:
return True, {"status": "success", "message": "success"}
return False, {"status": "pending", "message": "pending"}
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ Looks for objects in bucket matching a prefix.
+ If none found, sleep for interval and check again. Otherwise, return matches.
+
+ :param bucket: the bucket in the google cloud storage where the objects are residing.
+ :param prefix: The prefix of the blob_names to match in the Google cloud storage bucket
+ :param google_cloud_conn_id: reference to the Google Connection
+ :param poke_interval: polling period in seconds to check
+ """
+
+ def __init__(
+ self,
+ bucket: str,
+ prefix: str,
+ poke_interval: float,
Review Comment:
missing from docstring
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -200,3 +200,76 @@ async def _is_blob_updated_after(
if blob_updated_time > target_date:
return True, {"status": "success", "message": "success"}
return False, {"status": "pending", "message": "pending"}
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ Looks for objects in bucket matching a prefix.
+ If none found, sleep for interval and check again. Otherwise, return matches.
+
+ :param bucket: the bucket in the google cloud storage where the objects are residing.
+ :param prefix: The prefix of the blob_names to match in the Google cloud storage bucket
+ :param google_cloud_conn_id: reference to the Google Connection
+ :param poke_interval: polling period in seconds to check
+ """
+
+ def __init__(
+ self,
+ bucket: str,
+ prefix: str,
+ poke_interval: float,
+ google_cloud_conn_id: str,
+ hook_params: dict[str, Any],
+ ):
+ super().__init__(
+ bucket=bucket,
+ object_name=prefix,
+ poke_interval=poke_interval,
+ google_cloud_conn_id=google_cloud_conn_id,
+ hook_params=hook_params,
+ )
+ self.prefix = prefix
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serializes GCSPrefixBlobTrigger arguments and classpath."""
+ return (
+ "airflow.providers.google.cloud.triggers.gcs.GCSPrefixBlobTrigger",
+ {
+ "bucket": self.bucket,
+ "prefix": self.prefix,
+ "poke_interval": self.poke_interval,
+ "google_cloud_conn_id": self.google_cloud_conn_id,
+ "hook_params": self.hook_params,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Simple loop until the matches are found for the given prefix on the bucket."""
+ try:
+ hook = self._get_async_hook()
+ while True:
+ res = await self._list_blobs_with_prefix(
Review Comment:
post-2.6 with logging in triggerer, we can now consider logging every now and again messages like "checking for files". nice to know from logs that the trigger is still doing something. not saying you must do in this PR just a reminder that it's something that makes more sense now with 2.6 than it did previously.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1175546663
##########
airflow/providers/google/cloud/sensors/gcs.py:
##########
@@ -307,10 +310,37 @@ def poke(self, context: Context) -> bool:
self._matches = hook.list(self.bucket, prefix=self.prefix)
return bool(self._matches)
- def execute(self, context: Context) -> list[str]:
+ def execute(self, context: Context):
"""Overridden to allow matches to be passed"""
- super().execute(context)
- return self._matches
+ if not self.deferrable:
+ super().execute(context)
+ return self._matches
+ else:
+ self.defer(
+ timeout=timedelta(seconds=self.timeout),
+ trigger=GCSPrefixBlobTrigger(
+ bucket=self.bucket,
+ prefix=self.prefix,
+ poke_interval=self.poke_interval,
+ google_cloud_conn_id=self.google_cloud_conn_id,
+ hook_params={
+ "delegate_to": self.delegate_to,
+ "impersonation_chain": self.impersonation_chain,
+ },
+ ),
+ method_name="execute_complete",
+ )
+
+ def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]:
+ """
+ Callback for when the trigger fires - returns immediately.
+ Relies on trigger to throw an exception, otherwise it assumes execution was
Review Comment:
fixed the docstring
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1175549494
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -97,3 +97,76 @@ async def _object_exists(self, hook: GCSAsyncHook, bucket_name: str, object_name
if object_response:
return "success"
return "pending"
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ A trigger that fires and it looks for all the objects in the given bucket
+ which matches the given prefix if not found sleep for certain interval and checks again.
+
+ :param bucket: the bucket in the google cloud storage where the objects are residing.
+ :param prefix: The prefix of the blob_names to match in the Google cloud storage bucket
+ :param google_cloud_conn_id: reference to the Google Connection
+ :param poke_interval: polling period in seconds to check
+ """
+
+ def __init__(
+ self,
+ bucket: str,
+ prefix: str,
+ poke_interval: float,
+ google_cloud_conn_id: str,
+ hook_params: dict[str, Any],
+ ):
+ super().__init__(
+ bucket=bucket,
+ object_name=prefix,
+ poke_interval=poke_interval,
+ google_cloud_conn_id=google_cloud_conn_id,
+ hook_params=hook_params,
+ )
+ self.prefix = prefix
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serializes GCSPrefixBlobTrigger arguments and classpath."""
+ return (
+ "airflow.providers.google.cloud.triggers.gcs.GCSPrefixBlobTrigger",
+ {
+ "bucket": self.bucket,
+ "prefix": self.prefix,
+ "poke_interval": self.poke_interval,
+ "google_cloud_conn_id": self.google_cloud_conn_id,
+ "hook_params": self.hook_params,
+ },
+ )
+
+ async def run(self) -> AsyncIterator["TriggerEvent"]:
+ """Simple loop until the matches are found for the given prefix on the bucket."""
+ try:
+ hook = self._get_async_hook()
+ while True:
+ res = await self._list_blobs_with_prefix(
+ hook=hook, bucket_name=self.bucket, prefix=self.prefix
+ )
+ if len(res) > 0:
+ yield TriggerEvent(
+ {"status": "success", "message": "Successfully completed", "matches": res}
+ )
+ await asyncio.sleep(self.poke_interval)
+ except Exception as e:
+ yield TriggerEvent({"status": "error", "message": str(e)})
+ return
+
+ async def _list_blobs_with_prefix(self, hook: GCSAsyncHook, bucket_name: str, prefix: str) -> list[str]:
+ """
+ Returns list of blobs which matches the given prefix for the given bucket.
Review Comment:
changed
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1175548233
##########
airflow/providers/google/cloud/sensors/gcs.py:
##########
@@ -307,10 +310,37 @@ def poke(self, context: Context) -> bool:
self._matches = hook.list(self.bucket, prefix=self.prefix)
return bool(self._matches)
- def execute(self, context: Context) -> list[str]:
+ def execute(self, context: Context):
"""Overridden to allow matches to be passed"""
- super().execute(context)
- return self._matches
+ if not self.deferrable:
+ super().execute(context)
+ return self._matches
+ else:
+ self.defer(
+ timeout=timedelta(seconds=self.timeout),
+ trigger=GCSPrefixBlobTrigger(
+ bucket=self.bucket,
+ prefix=self.prefix,
+ poke_interval=self.poke_interval,
+ google_cloud_conn_id=self.google_cloud_conn_id,
+ hook_params={
+ "delegate_to": self.delegate_to,
+ "impersonation_chain": self.impersonation_chain,
+ },
+ ),
+ method_name="execute_complete",
+ )
+
+ def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]:
+ """
+ Callback for when the trigger fires - returns immediately.
+ Relies on trigger to throw an exception, otherwise it assumes execution was
+ successful.
+ """
+ self.log.info("Sensor checks existence of objects: %s, %s", self.bucket, self.prefix)
Review Comment:
I've modified the log message as suggested above
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1175546168
##########
airflow/providers/google/cloud/sensors/gcs.py:
##########
@@ -307,10 +310,37 @@ def poke(self, context: Context) -> bool:
self._matches = hook.list(self.bucket, prefix=self.prefix)
return bool(self._matches)
- def execute(self, context: Context) -> list[str]:
+ def execute(self, context: Context):
"""Overridden to allow matches to be passed"""
- super().execute(context)
- return self._matches
+ if not self.deferrable:
+ super().execute(context)
+ return self._matches
+ else:
+ self.defer(
+ timeout=timedelta(seconds=self.timeout),
+ trigger=GCSPrefixBlobTrigger(
+ bucket=self.bucket,
+ prefix=self.prefix,
+ poke_interval=self.poke_interval,
+ google_cloud_conn_id=self.google_cloud_conn_id,
+ hook_params={
+ "delegate_to": self.delegate_to,
+ "impersonation_chain": self.impersonation_chain,
+ },
+ ),
+ method_name="execute_complete",
+ )
+
+ def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]:
+ """
+ Callback for when the trigger fires - returns immediately.
Review Comment:
done
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1169174297
##########
airflow/providers/google/cloud/sensors/gcs.py:
##########
@@ -307,10 +310,37 @@ def poke(self, context: Context) -> bool:
self._matches = hook.list(self.bucket, prefix=self.prefix)
return bool(self._matches)
- def execute(self, context: Context) -> list[str]:
+ def execute(self, context: Context):
"""Overridden to allow matches to be passed"""
- super().execute(context)
- return self._matches
+ if not self.deferrable:
+ super().execute(context)
+ return self._matches
+ else:
+ self.defer(
+ timeout=timedelta(seconds=self.timeout),
+ trigger=GCSPrefixBlobTrigger(
+ bucket=self.bucket,
+ prefix=self.prefix,
+ poke_interval=self.poke_interval,
+ google_cloud_conn_id=self.google_cloud_conn_id,
+ hook_params={
+ "delegate_to": self.delegate_to,
+ "impersonation_chain": self.impersonation_chain,
+ },
+ ),
+ method_name="execute_complete",
+ )
+
+ def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]:
+ """
+ Callback for when the trigger fires - returns immediately.
+ Relies on trigger to throw an exception, otherwise it assumes execution was
+ successful.
+ """
+ self.log.info("Sensor checks existence of objects: %s, %s", self.bucket, self.prefix)
Review Comment:
```suggestion
self.log.info("Checking for existence of object: %s, %s", self.bucket, self.prefix)
```
maybe tiny bit better?
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -97,3 +97,76 @@ async def _object_exists(self, hook: GCSAsyncHook, bucket_name: str, object_name
if object_response:
return "success"
return "pending"
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ A trigger that fires and it looks for all the objects in the given bucket
+ which matches the given prefix if not found sleep for certain interval and checks again.
+
+ :param bucket: the bucket in the google cloud storage where the objects are residing.
+ :param prefix: The prefix of the blob_names to match in the Google cloud storage bucket
+ :param google_cloud_conn_id: reference to the Google Connection
+ :param poke_interval: polling period in seconds to check
+ """
+
+ def __init__(
+ self,
+ bucket: str,
+ prefix: str,
+ poke_interval: float,
+ google_cloud_conn_id: str,
+ hook_params: dict[str, Any],
+ ):
+ super().__init__(
+ bucket=bucket,
+ object_name=prefix,
+ poke_interval=poke_interval,
+ google_cloud_conn_id=google_cloud_conn_id,
+ hook_params=hook_params,
+ )
+ self.prefix = prefix
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serializes GCSPrefixBlobTrigger arguments and classpath."""
+ return (
+ "airflow.providers.google.cloud.triggers.gcs.GCSPrefixBlobTrigger",
+ {
+ "bucket": self.bucket,
+ "prefix": self.prefix,
+ "poke_interval": self.poke_interval,
+ "google_cloud_conn_id": self.google_cloud_conn_id,
+ "hook_params": self.hook_params,
+ },
+ )
+
+ async def run(self) -> AsyncIterator["TriggerEvent"]:
+ """Simple loop until the matches are found for the given prefix on the bucket."""
+ try:
+ hook = self._get_async_hook()
+ while True:
+ res = await self._list_blobs_with_prefix(
+ hook=hook, bucket_name=self.bucket, prefix=self.prefix
+ )
+ if len(res) > 0:
+ yield TriggerEvent(
+ {"status": "success", "message": "Successfully completed", "matches": res}
+ )
+ await asyncio.sleep(self.poke_interval)
+ except Exception as e:
+ yield TriggerEvent({"status": "error", "message": str(e)})
+ return
+
+ async def _list_blobs_with_prefix(self, hook: GCSAsyncHook, bucket_name: str, prefix: str) -> list[str]:
+ """
+ Returns list of blobs which matches the given prefix for the given bucket.
Review Comment:
```suggestion
Returns names of blobs which matches the given prefix for the given bucket.
```
##########
airflow/providers/google/cloud/sensors/gcs.py:
##########
@@ -307,10 +310,37 @@ def poke(self, context: Context) -> bool:
self._matches = hook.list(self.bucket, prefix=self.prefix)
return bool(self._matches)
- def execute(self, context: Context) -> list[str]:
+ def execute(self, context: Context):
"""Overridden to allow matches to be passed"""
- super().execute(context)
- return self._matches
+ if not self.deferrable:
+ super().execute(context)
+ return self._matches
+ else:
+ self.defer(
+ timeout=timedelta(seconds=self.timeout),
+ trigger=GCSPrefixBlobTrigger(
+ bucket=self.bucket,
+ prefix=self.prefix,
+ poke_interval=self.poke_interval,
+ google_cloud_conn_id=self.google_cloud_conn_id,
+ hook_params={
+ "delegate_to": self.delegate_to,
+ "impersonation_chain": self.impersonation_chain,
+ },
+ ),
+ method_name="execute_complete",
+ )
+
+ def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]:
+ """
+ Callback for when the trigger fires - returns immediately.
Review Comment:
```suggestion
Callback for when the trigger fires; returns immediately.
```
saw same in https://github.com/apache/airflow/pull/30579
##########
airflow/providers/google/cloud/sensors/gcs.py:
##########
@@ -307,10 +310,37 @@ def poke(self, context: Context) -> bool:
self._matches = hook.list(self.bucket, prefix=self.prefix)
return bool(self._matches)
- def execute(self, context: Context) -> list[str]:
+ def execute(self, context: Context):
"""Overridden to allow matches to be passed"""
- super().execute(context)
- return self._matches
+ if not self.deferrable:
+ super().execute(context)
+ return self._matches
+ else:
+ self.defer(
+ timeout=timedelta(seconds=self.timeout),
+ trigger=GCSPrefixBlobTrigger(
+ bucket=self.bucket,
+ prefix=self.prefix,
+ poke_interval=self.poke_interval,
+ google_cloud_conn_id=self.google_cloud_conn_id,
+ hook_params={
+ "delegate_to": self.delegate_to,
+ "impersonation_chain": self.impersonation_chain,
+ },
+ ),
+ method_name="execute_complete",
+ )
+
+ def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]:
+ """
+ Callback for when the trigger fires - returns immediately.
+ Relies on trigger to throw an exception, otherwise it assumes execution was
Review Comment:
Is this true anymore? Looks like it checks now whether success. Saw same in https://github.com/apache/airflow/pull/30579
##########
docs/apache-airflow-providers-google/operators/cloud/gcs.rst:
##########
@@ -199,8 +199,20 @@ Use the :class:`~airflow.providers.google.cloud.sensors.gcs.GCSObjectsWithPrefix
:start-after: [START howto_sensor_object_with_prefix_exists_task]
:end-before: [END howto_sensor_object_with_prefix_exists_task]
+You can set the ``deferrable`` param to True if you want this sensor to run asynchronously - leading to efficient
+utilization of resources in your Airflow deployment. However the triggerer component needs to be up and running
Review Comment:
```suggestion
utilization of resources in your Airflow deployment. However the triggerer component needs to be enabled
```
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -97,3 +97,76 @@ async def _object_exists(self, hook: GCSAsyncHook, bucket_name: str, object_name
if object_response:
return "success"
return "pending"
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ A trigger that fires and it looks for all the objects in the given bucket
+ which matches the given prefix if not found sleep for certain interval and checks again.
Review Comment:
```suggestion
Looks for objects in bucket matching the prefix.
If none found, sleep for interval and check again. Otherwise, return matches.
```
##########
docs/apache-airflow-providers-google/operators/cloud/gcs.rst:
##########
@@ -199,8 +199,20 @@ Use the :class:`~airflow.providers.google.cloud.sensors.gcs.GCSObjectsWithPrefix
:start-after: [START howto_sensor_object_with_prefix_exists_task]
:end-before: [END howto_sensor_object_with_prefix_exists_task]
+You can set the ``deferrable`` param to True if you want this sensor to run asynchronously - leading to efficient
Review Comment:
```suggestion
You can set the ``deferrable`` param to True if you want this sensor to run asynchronously, leading to more efficient
```
##########
docs/apache-airflow-providers-google/operators/cloud/gcs.rst:
##########
@@ -199,8 +199,20 @@ Use the :class:`~airflow.providers.google.cloud.sensors.gcs.GCSObjectsWithPrefix
:start-after: [START howto_sensor_object_with_prefix_exists_task]
:end-before: [END howto_sensor_object_with_prefix_exists_task]
+You can set the ``deferrable`` param to True if you want this sensor to run asynchronously - leading to efficient
+utilization of resources in your Airflow deployment. However the triggerer component needs to be up and running
Review Comment:
above two comments also apply to https://github.com/apache/airflow/pull/30579
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] phanikumv commented on a diff in pull request #30618: Add deferrable mode to `GCSObjectsWithPrefixExistenceSensor`
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #30618:
URL: https://github.com/apache/airflow/pull/30618#discussion_r1179056405
##########
airflow/providers/google/cloud/triggers/gcs.py:
##########
@@ -200,3 +200,76 @@ async def _is_blob_updated_after(
if blob_updated_time > target_date:
return True, {"status": "success", "message": "success"}
return False, {"status": "pending", "message": "pending"}
+
+
+class GCSPrefixBlobTrigger(GCSBlobTrigger):
+ """
+ Looks for objects in bucket matching a prefix.
+ If none found, sleep for interval and check again. Otherwise, return matches.
+
+ :param bucket: the bucket in the google cloud storage where the objects are residing.
+ :param prefix: The prefix of the blob_names to match in the Google cloud storage bucket
+ :param google_cloud_conn_id: reference to the Google Connection
+ :param poke_interval: polling period in seconds to check
+ """
+
+ def __init__(
+ self,
+ bucket: str,
+ prefix: str,
+ poke_interval: float,
Review Comment:
added hook_params in docstring
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org