Fix pending issues

Author: kaxil

astronomer/providers/google/cloud/sensors/gcs.py

@@ -92,6 +92,30 @@ def execute_complete(self, context: "Context", event=None) -> None:
 
 
 class GCSObjectsWithPrefixExistenceSensorAsync(GCSObjectsWithPrefixExistenceSensor):
+    """
+    Async Operator that Checks for the existence of GCS objects at a given prefix, passing matches via XCom.
+
+    When files matching the given prefix are found, the poke method's criteria will be
+    fulfilled and the matching objects will be returned from the operator and passed
+    through XCom for downstream tasks.
+
+    :param bucket: The Google Cloud Storage bucket where the object is.
+    :param prefix: The name of the prefix to check in the Google cloud storage bucket.
+    :param google_cloud_conn_id: The connection ID to use when connecting to Google Cloud Storage.
+    :param delegate_to: The account to impersonate using domain-wide delegation of authority,
+        if any. For this to work, the service account making the request must have
+        domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    :param polling_interval: The interval in seconds to wait between checks for matching objects.
+    """
+
     def __init__(
         self,
         polling_interval: float = 5.0,
@@ -126,6 +150,41 @@ def execute_complete(self, context: "Context", event: Optional[Dict[Any, Any]] =
 
 
 class GCSUploadSessionCompleteSensorAsync(GCSUploadSessionCompleteSensor):
+    """
+    Checks for changes in the number of objects at prefix in Google Cloud Storage
+    bucket and returns True if the inactivity period has passed with no
+    increase in the number of objects. Note, this sensor will no behave correctly
+    in reschedule mode, as the state of the listed objects in the GCS bucket will
+    be lost between rescheduled invocations.
+
+    :param bucket: The Google Cloud Storage bucket where the objects are expected.
+    :param prefix: The name of the prefix to check in the Google cloud storage bucket.
+    :param inactivity_period: The total seconds of inactivity to designate
+        an upload session is over. Note, this mechanism is not real time and
+        this operator may not return until a poke_interval after this period
+        has passed with no additional objects sensed.
+    :param min_objects: The minimum number of objects needed for upload session
+        to be considered valid.
+    :param previous_objects: The set of object ids found during the last poke.
+    :param allow_delete: Should this sensor consider objects being deleted
+        between pokes valid behavior. If true a warning message will be logged
+        when this happens. If false an error will be raised.
+    :param google_cloud_conn_id: The connection ID to use when connecting
+        to Google Cloud Storage.
+    :param delegate_to: The account to impersonate using domain-wide delegation of authority,
+        if any. For this to work, the service account making the request must have
+        domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    :param polling_interval: The interval in seconds to wait between checks for matching objects.
+    """
+
     def __init__(
         self,
         polling_interval: float = 5.0,

astronomer/providers/google/cloud/triggers/gcs.py

@@ -317,15 +317,16 @@ def _is_bucket_updated(self, current_objects: Set[str]) -> Dict[str, str]:
 
             if current_num_objects >= self.min_objects:
                 success_message = (
-                    "SUCCESS: Sensor found %s objects at %s. Waited at least %s seconds, with no new objects dropped.",
-                    current_num_objects,
-                    path,
-                    self.inactivity_period,
+                    "SUCCESS: Sensor found %s objects at %s. Waited at least %s "
+                    "seconds, with no new objects dropped."
                 )
-                self.log.info(success_message)
-                return {"status": "success", "message": success_message}
-
-            error_message = "FAILURE: Inactivity Period passed, not enough objects found in %s", path
-            self.log.error(error_message)
-            return {"status": "error", "message": error_message}
+                self.log.info(success_message, current_num_objects, path, self.inactivity_seconds)
+                return {
+                    "status": "success",
+                    "message": success_message % (current_num_objects, path, self.inactivity_seconds),
+                }
+
+            error_message = "FAILURE: Inactivity Period passed, not enough objects found in %s"
+            self.log.error(error_message, path)
+            return {"status": "error", "message": error_message % path}
         return {"status": "pending"}