josh-fell commented on code in PR #39154:
URL: https://github.com/apache/airflow/pull/39154#discussion_r1590188472
##########
airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py:
##########
@@ -434,6 +434,82 @@ def execute(self, context: Context) -> None:
hook.delete_transfer_job(job_name=self.job_name,
project_id=self.project_id)
+class CloudDataTransferServiceRunJobOperator(GoogleCloudBaseOperator):
+ """
+ Runs a transfer job.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the
guide:
+ :ref:`howto/operator:CloudDataTransferServiceRunJobOperator`
+
+ :param job_name: (Required) Name of the job to be run
+ :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]
+ operator_extra_links = (CloudStorageTransferJobLink(),)
+
+ def __init__(
+ self,
+ *,
+ job_name: str,
+ gcp_conn_id: str = "google_cloud_default",
+ api_version: str = "v1",
+ project_id: str = PROVIDE_PROJECT_ID,
+ google_impersonation_chain: str | Sequence[str] | None = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ 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()
+
+ def _validate_inputs(self) -> None:
+ if not self.job_name:
+ raise AirflowException("The required parameter 'job_name' is empty
or None")
Review Comment:
It might be worth moving this check to the `execute()` method because
`job_name` is a template field. If a Jinja expression is passed, this
validation will always be True because it will be a string (or even an
`XComArg` object); however, the rendered value could result to a `None` value.
--
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-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
