Implement GCSObjectsWithPrefixExistenceSensorAsync and GCSUploadSessionCompleteSensor #90

rajaths010494 · 2022-03-03T09:32:31Z

Issue: #69

Implement the sensor GCSObjectsWithPrefixExistenceSensorAsync
Implement triggers for GCSObjectsWithPrefixExistenceSensorAsync
Add example DAG for GCSObjectsWithPrefixExistenceSensorAsync
Add test for GCSObjectsWithPrefixExistenceSensorAsync sensor, trigger and hooks

Implement the sensor GCSUploadSessionCompleteSensor
Implement triggers for GCSUploadSessionCompleteSensor
Add example DAG for GCSUploadSessionCompleteSensor
Add test for GCSUploadSessionCompleteSensor sensor, trigger and hooks

codecov · 2022-03-03T11:14:55Z

Codecov Report

Merging #90 (8382289) into main (c033f6c) will decrease coverage by 0.16%.
The diff coverage is 99.10%.

❗ Current head 8382289 differs from pull request most recent head 932659f. Consider uploading reports for the commit 932659f to get more accurate results

@@            Coverage Diff             @@
##             main      #90      +/-   ##
==========================================
- Coverage   89.16%   88.99%   -0.17%     
==========================================
  Files          35       35              
  Lines        1689     1636      -53     
==========================================
- Hits         1506     1456      -50     
+ Misses        183      180       -3

Impacted Files	Coverage Δ
astronomer/providers/google/cloud/sensors/gcs.py	`97.82% <96.42%> (-2.18%)`	⬇️
astronomer/providers/google/cloud/triggers/gcs.py	`100.00% <100.00%> (+7.69%)`	⬆️
astronomer/providers/amazon/aws/triggers/s3.py	`80.39% <0.00%> (-17.26%)`	⬇️
...r/providers/amazon/aws/sensors/redshift_cluster.py	`84.00% <0.00%> (-2.96%)`	⬇️
...nomer/providers/amazon/aws/hooks/base_aws_async.py	`87.09% <0.00%> (-0.41%)`	⬇️
...omer/providers/amazon/aws/triggers/redshift_sql.py	`100.00% <0.00%> (ø)`
.../providers/amazon/aws/triggers/redshift_cluster.py	`79.62% <0.00%> (+0.78%)`	⬆️
astronomer/providers/amazon/aws/sensors/s3.py	`100.00% <0.00%> (+1.56%)`	⬆️
...onomer/providers/amazon/aws/hooks/redshift_data.py	`98.11% <0.00%> (+3.37%)`	⬆️
...mer/providers/amazon/aws/operators/redshift_sql.py	`96.00% <0.00%> (+3.69%)`	⬆️
... and 2 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update c033f6c...932659f. Read the comment docs.

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

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

sunank200

Overall, implementation LGTM.

I had a question - Is there a specific reason to keep certain methods as private methods in triggers than to keep in hooks?

Also, there are few comments I have added.

astronomer/providers/google/cloud/example_dags/example_gcs.py

sunank200 · 2022-03-07T08:02:45Z

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

@@ -55,7 +66,7 @@ def __init__(
        self.delegate_to = delegate_to
        self.impersonation_chain = impersonation_chain

-    def execute(self, context):
+    def execute(self, context: "Context"):


Return type is missing for execute().

This is already existing will be fixed as part of mypy-strict branch

sunank200 · 2022-03-07T08:03:25Z

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

+
+
+class GCSObjectsWithPrefixExistenceSensorAsync(GCSObjectsWithPrefixExistenceSensor):
+    def __init__(


Docstring and comment missing explaining the sensor.

it is inherited from sync version added doc string for only async inherited from base operator

It still needs to be added as we have one more kwarg: polling_interval that's not in the inherited operator. And this will be need for our Sphinx docs

sunank200 · 2022-03-07T08:04:25Z

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

+
+
+class GCSUploadSessionCompleteSensorAsync(GCSUploadSessionCompleteSensor):
+    def __init__(


Docstring and comment missing explaining the sensor.

it is inherited from sync version added doc string for only async inherited from base operator

sunank200 · 2022-03-07T08:05:44Z

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

+
+    def execute_complete(
+        self, context: "Context", event: Optional[Dict[Any, Any]] = None
+    ) -> None:  # pylint: disable=unused-argument


return type is Optional[str] as per the implementation rather than None.

updated the return type

sunank200 · 2022-03-07T08:06:36Z

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

+
+    def serialize(self) -> Tuple[str, Dict[str, Any]]:
+        """
+        Serializes GCSBlobTrigger arguments and classpath.


It should be GCSPrefixBlobTrigger instead of GCSBlobTrigger

sunank200 · 2022-03-07T08:13:23Z

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

+        :param prefix: The prefix of the blob_names to match in the Google cloud
+            storage bucket.
+        """
+        async with Session() as s:


minor comment: s can be named better.

sunank200 · 2022-03-07T08:20:39Z

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

+        """
+        return datetime.now()
+
+    def _is_bucket_updated(self, current_objects: Set[str]) -> dict[str, str]:


Any reason to keep it in triggers than in hooks?

Its used by GCSUploadSessionTrigger only if its was reused in other also would have moved to hooks

pankajastro · 2022-03-08T13:07:57Z

astronomer/providers/google/cloud/example_dags/example_gcs.py


-START_DATE = datetime(2022, 1, 1)
+START_DATE = datetime(2021, 1, 1)


nit: since START_DATE is getting used once only we should directly use datetime(2021, 1, 1) in line 51

pankajastro · 2022-03-08T13:10:41Z

astronomer/providers/google/cloud/example_dags/example_gcs.py


+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "astronomer-airflow-providers")
+BUCKET_1 = "test-gcs-example-bucket"


Can we add this also env variable?

added bucket in env

kaxil · 2022-03-08T20:47:59Z

astronomer/providers/google/cloud/example_dags/example_gcs.py

+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.


Suggested change

#

# Licensed to the Apache Software Foundation (ASF) under one

# or more contributor license agreements. See the NOTICE file

# distributed with this work for additional information

# regarding copyright ownership. The ASF licenses this file

# to you under the Apache License, Version 2.0 (the

# "License"); you may not use this file except in compliance

# with the License. You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing,

# software distributed under the License is distributed on an

# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

# KIND, either express or implied. See the License for the

# specific language governing permissions and limitations

# under the License.

This project is not maintained by ASF

kaxil · 2022-03-08T20:52:48Z

Need to link PR to issue -> https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword

You need to add the following in the PR description
closes #69

kaxil · 2022-03-08T21:07:07Z

astronomer/providers/google/cloud/example_dags/example_gcs.py

    catchup=False,
-    schedule_interval="@once",
-    tags=["example"],
+    schedule_interval='@once',


unrelated change

- Implement the GCSPrefixBlobTrigger( - Implement the sensor GCSObjectWithPrefixExistenceSensor

- add example dag for GCSObjectsWithPrefixExistenceSensorAsync - modify trigger super.__init__

- Implement the GCSUploadSessionTrigger - Implement the sensor GCSUploadSessionCompleteSensorAsync

…rAsync

We don't use Pylint.

Addressed changes

kaxil · 2022-03-08T22:08:11Z

Well done, I made some changes @rajaths010494 , please check the last few commits and my comments -- do ask if you need any clarification

rajaths010494 requested review from sunank200, bharanidharan14, pankajastro, kaxil and phanikumv March 3, 2022 09:32

bharanidharan14 reviewed Mar 3, 2022

View reviewed changes

astronomer/providers/google/cloud/triggers/gcs.py Outdated Show resolved Hide resolved

astronomer/providers/google/cloud/triggers/gcs.py Show resolved Hide resolved

sunank200 previously requested changes Mar 7, 2022

View reviewed changes

rajaths010494 changed the title ~~Implement GCSObjectsWithPrefixExistenceSensorAsync~~ Implement GCSObjectsWithPrefixExistenceSensorAsync and GCSUploadSessionCompleteSensor Mar 8, 2022

rajaths010494 mentioned this pull request Mar 8, 2022

Implement async versions of the remaining GCS sensors #69

Closed

11 tasks

pankajastro reviewed Mar 8, 2022

View reviewed changes

kaxil force-pushed the gcs_sensors branch from 60158ec to 8382289 Compare March 8, 2022 20:47

kaxil reviewed Mar 8, 2022

View reviewed changes

rajaths010494 added 9 commits March 8, 2022 21:14

Implement GCSObjectWithPrefixExistenceSensor

d36de15

- Implement the GCSPrefixBlobTrigger( - Implement the sensor GCSObjectWithPrefixExistenceSensor

Fix static check in triggers

904f038

add example dag for GCSObjectsWithPrefixExistenceSensorAsync

8fb6867

- add example dag for GCSObjectsWithPrefixExistenceSensorAsync - modify trigger super.__init__

Add tests for GCSObjectsWithPrefixExistenceSensorAsync

b2c9817

Fix mypy issues in gcs sensor

4f4e994

Update doc_string in GCSPrefixBlobTrigger

8075b3b

Implement GCSUploadSessionCompleteSensorAsync

6a2442d

- Implement the GCSUploadSessionTrigger - Implement the sensor GCSUploadSessionCompleteSensorAsync

Add example dag and unit test cases for GCSUploadSessionCompleteSenso…

ffd0171

…rAsync

Fix mypy changes and add env in example dag

ca61c99

kaxil force-pushed the gcs_sensors branch from 8382289 to 6fe0863 Compare March 8, 2022 21:27

kaxil added 3 commits March 8, 2022 21:29

Disable string normalization in Black

47b3601

Remove pylint disables

4750c72

We don't use Pylint.

Fix pending issues

f8ec4d4

kaxil force-pushed the gcs_sensors branch from 6fe0863 to f8ec4d4 Compare March 8, 2022 21:29

kaxil marked this pull request as ready for review March 8, 2022 21:29

kaxil requested a review from sunank200 March 8, 2022 21:29

Fix test

932659f

kaxil approved these changes Mar 8, 2022

View reviewed changes

kaxil merged commit 6a1c9d6 into main Mar 8, 2022

kaxil deleted the gcs_sensors branch March 8, 2022 22:06



		class GCSObjectsWithPrefixExistenceSensorAsync(GCSObjectsWithPrefixExistenceSensor):
		def __init__(



		class GCSUploadSessionCompleteSensorAsync(GCSUploadSessionCompleteSensor):
		def __init__(


		START_DATE = datetime(2022, 1, 1)
		START_DATE = datetime(2021, 1, 1)


		PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "astronomer-airflow-providers")
		BUCKET_1 = "test-gcs-example-bucket"

Implement GCSObjectsWithPrefixExistenceSensorAsync and GCSUploadSessionCompleteSensor #90

Implement GCSObjectsWithPrefixExistenceSensorAsync and GCSUploadSessionCompleteSensor #90

Uh oh!

Conversation

rajaths010494 commented Mar 3, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov bot commented Mar 3, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sunank200 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rajaths010494 Mar 8, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kaxil commented Mar 8, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kaxil commented Mar 8, 2022

Uh oh!

Uh oh!

rajaths010494 commented Mar 3, 2022 •

edited

Loading

codecov bot commented Mar 3, 2022 •

edited

Loading

rajaths010494 Mar 8, 2022 •

edited

Loading

kaxil commented Mar 8, 2022 •

edited

Loading