[GitHub] [airflow] abhinavraj23 commented on a diff in pull request #24563: feat: add an operator to run a transfer job

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

abhinavraj23 commented on code in PR #24563:
URL: https://github.com/apache/airflow/pull/24563#discussion_r928273721


##########
airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py:
##########
@@ -326,6 +326,67 @@ def execute(self, context: 'Context') -> dict:
         return hook.update_transfer_job(job_name=self.job_name, body=self.body)
 
 
+class CloudDataTransferServiceRunJobOperator(BaseOperator):
+    """
+    Runs a transfer job in Google Storage Transfer Service.
+
+    :param job_name: (Required) Name of the transfer job.
+    :param project_id: (Optional) the ID of the project that owns the Transfer
+        Job. If set to None or missing, the default project_id from the Google Cloud
+        connection is used.
+    :param gcp_conn_id: The connection ID used to connect to Google Cloud.
+    :param api_version: API version used (e.g. v1).
+    :param google_impersonation_chain: Optional Google 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).
+    """
+
+    # [START gcp_transfer_job_run_template_fields]
+    template_fields: Sequence[str] = (
+        'job_name',
+        'project_id',
+        'gcp_conn_id',
+        'api_version',
+        'google_impersonation_chain',
+    )
+    # [END gcp_transfer_job_run_template_fields]
+
+    def __init__(
+        self,
+        *,
+        job_name: str,
+        gcp_conn_id: str = "google_cloud_default",
+        api_version: str = "v1",
+        project_id: Optional[str] = None,
+        google_impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+        **kwargs,
+    ) -> None:
+        self.job_name = job_name
+        self.project_id = project_id
+        self.gcp_conn_id = gcp_conn_id
+        self.api_version = api_version
+        self.google_impersonation_chain = google_impersonation_chain
+        self._validate_inputs()
+        super().__init__(**kwargs)
+
+    def _validate_inputs(self) -> None:
+        if not self.job_name:
+            raise AirflowException("The required parameter 'job_name' is empty or None")
+
+    def execute(self, context: "Context") -> None:
+        hook = CloudDataTransferServiceHook(
+            api_version=self.api_version,
+            gcp_conn_id=self.gcp_conn_id,
+            impersonation_chain=self.google_impersonation_chain,
+        )
+        hook.run_transfer_job(job_name=self.job_name, project_id=self.project_id)

Review Comment:
   @turbaszek  It will create another transfer operation corresponding to this transfer job, here's the documentation of the same - https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferJobs



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