e-galan commented on code in PR #39018:
URL: https://github.com/apache/airflow/pull/39018#discussion_r1577409161
##########
airflow/providers/google/cloud/hooks/dataflow.py:
##########
@@ -736,6 +736,69 @@ def start_template_dataflow(
jobs_controller.wait_for_done()
return response["job"]
+ @_fallback_to_location_from_variables
+ @_fallback_to_project_id_from_variables
+ @GoogleBaseHook.fallback_to_default_project_id
+ def launch_job_with_template(
Review Comment:
@Lee-W I was thinking about it, but because of `name =
self.build_dataflow_job_name(job_name, append_job_name)` on line 684, I can't
do that without changing the original method's parameters somehow. Either that,
or `launch_job_with_template` method will need to return both the _job name_
and _job response_, and we only need the _job response_ for the deferrable
mode. So it won't be exactly clean.
However if the goal is to minimize code duplication, there is one block that
could be extracted into a separate method:
` service: Resource = self.get_conn()
request = (
service.projects()
.locations()
.templates()
.launch(
projectId=project_id,
location=location,
gcsPath=dataflow_template,
body={
"jobName": name,
"parameters": parameters,
"environment": environment,
},
)
)
response = request.execute(num_retries=self.num_retries)
job = response["job"]`
It can be used in `start_template_dataflow` and `launch_job_with_template`
methods.
Does it make sense to you?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
