[GitHub] [airflow] dstandish commented on a change in pull request #14492: Added new community provider Airbyte

[GitBox <gi...@apache.org>]

dstandish commented on a change in pull request #14492:
URL: https://github.com/apache/airflow/pull/14492#discussion_r584090240



##########
File path: airflow/providers/airbyte/hooks/airbyte.py
##########
@@ -0,0 +1,92 @@
+#
+# 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.
+import time
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.providers.http.hooks.http import HttpHook
+
+
+class AirbyteJobController:
+    """Airbyte job status"""
+
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    CANCELLED = "canceled"
+    PENDING = "pending"
+    FAILED = "failed"
+    ERROR = "error"
+
+
+class AirbyteHook(HttpHook, AirbyteJobController):

Review comment:
       this seems a little odd
   
   i think more standard approach would be one of the following:
   
   1. define the constants in the module directly
   2. define the constants directly on the AirbyteHook class
   3. define the constants as part of an enum class (you can make an enum class whose elements also behave like a string)  -- and just reference the values directly rather than through inheritance
   
   i would suggest using one of these 3 alternative approaches.  AirbyteJobController makes it sound much more elaborate than what it is, and there's really not a need for inheritance here.
   
   ...
   
   i think the naming and the inheritance is the main thing that's sticking out to me...
   
   even if you just called it `AirbyteJobStatus` and didn't use inheritance that would be fine

##########
File path: airflow/providers/airbyte/hooks/airbyte.py
##########
@@ -0,0 +1,92 @@
+#
+# 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.
+import time
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.providers.http.hooks.http import HttpHook
+
+
+class AirbyteJobController:
+    """Airbyte job status"""
+
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    CANCELLED = "canceled"
+    PENDING = "pending"
+    FAILED = "failed"
+    ERROR = "error"
+
+
+class AirbyteHook(HttpHook, AirbyteJobController):
+    """Hook for Airbyte API"""
+
+    def __init__(self, airbyte_conn_id: str) -> None:
+        super().__init__(http_conn_id=airbyte_conn_id)
+
+    def wait_for_job(self, job_id: str, wait_time: int = 3, timeout: Optional[int] = None) -> None:

Review comment:
       this is very close to a sensor... might consider making a sensor so you can free up worker slot while waiting for job to complete.  

##########
File path: airflow/providers/airbyte/hooks/airbyte.py
##########
@@ -0,0 +1,92 @@
+#
+# 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.
+import time
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.providers.http.hooks.http import HttpHook
+
+
+class AirbyteJobController:
+    """Airbyte job status"""
+
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    CANCELLED = "canceled"
+    PENDING = "pending"
+    FAILED = "failed"
+    ERROR = "error"
+
+
+class AirbyteHook(HttpHook, AirbyteJobController):
+    """Hook for Airbyte API"""
+
+    def __init__(self, airbyte_conn_id: str) -> None:
+        super().__init__(http_conn_id=airbyte_conn_id)
+
+    def wait_for_job(self, job_id: str, wait_time: int = 3, timeout: Optional[int] = None) -> None:

Review comment:
       maybe `wait_seconds` would be more explicit

##########
File path: airflow/providers/airbyte/hooks/airbyte.py
##########
@@ -0,0 +1,92 @@
+#
+# 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.
+import time
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.providers.http.hooks.http import HttpHook
+
+
+class AirbyteJobController:
+    """Airbyte job status"""
+
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    CANCELLED = "canceled"
+    PENDING = "pending"
+    FAILED = "failed"
+    ERROR = "error"
+
+
+class AirbyteHook(HttpHook, AirbyteJobController):
+    """Hook for Airbyte API"""
+
+    def __init__(self, airbyte_conn_id: str) -> None:
+        super().__init__(http_conn_id=airbyte_conn_id)
+
+    def wait_for_job(self, job_id: str, wait_time: int = 3, timeout: Optional[int] = None) -> None:
+        """
+        Helper method which polls a job to check if it finishes.
+
+        :param job_id: Id of the Airbyte job
+        :type job_id: str
+        :param wait_time: Number of seconds between checks
+        :type wait_time: int
+        :param timeout: How many seconds wait for job to be ready. Used only if ``asynchronous`` is False
+        :type timeout: int
+        """
+        state = None
+        start = time.monotonic()
+        while state not in (self.ERROR, self.SUCCEEDED, self.CANCELLED):
+            if timeout and start + timeout < time.monotonic():
+                raise AirflowException(f"Timeout: Airbyte job {job_id} is not ready after {timeout}s")
+            time.sleep(wait_time)
+            try:
+                job = self.get_job(job_id=job_id)
+                state = job.json().get("job").get("status")
+            except AirflowException as err:
+                self.log.info("Retrying. Airbyte API returned server error when waiting for job: %s", err)
+
+        if state == self.ERROR:
+            raise AirflowException(f"Job failed:\n{job}")
+        if state == self.CANCELLED:
+            raise AirflowException(f"Job was cancelled:\n{job}")
+
+    def submit_job(self, connection_id: str) -> dict:

Review comment:
       is `connections/sync` the only type of job?
   
   the name `submit_job` sounds way more generic than just syncing a defined connection.  e.g. with like EMR maybe `submit_job` lets you pass all kinds of parameters like jarfile and args etc.  But here you can really only do one thing.  So in that case better to name the method more clearly like `submit_sync_connection` perhaps.  presumably airbyte lets you submit more kinds of jobs than simply this one case?




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org