SameerMesiah97 commented on code in PR #77069:
URL: https://github.com/apache/airflow/pull/77069#discussion_r4569077789
##########
providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py:
##########
@@ -460,6 +462,21 @@ def fetch_job_id(self) -> str | None:
self.log.info("Google Cloud DataFlow job not available yet..")
return self._job_id
+ def fetch_latest_job_by_name(self, job_name: str) -> dict | None:
+ """
+ Return the latest created job with the given name, in any state.
+
+ Dataflow API does not support getting a job by name, so all jobs in
the location are listed
+ and filtered here.
+
+ :param job_name: Name of the job. It is converted to lowercase because
Dataflow job names
+ are always lowercase.
+ :return: The job, or None if no job has this name.
+ """
+ job_name = job_name.lower()
+ jobs = [job for job in self._fetch_all_jobs() if job.get("name") ==
job_name]
+ return max(jobs, key=lambda job: job.get("createTime", ""),
default=None)
Review Comment:
You could avoid the intermediate list by doing something like this:
```
return max(
(job for job in self._fetch_all_jobs() if job.get("name") == job_name),
key=lambda job: job.get("createTime", ""),
default=None,
)
```
I see that you have used this pattern multiple times in this PR so this
feedback applies there too.
Also, could we parse `createTime` before comparing jobs? RFC 3339 strings
with different fractional precision do not always sort chronologically. For
example, ...00Z sorts after ...00.100Z, although it is earlier. Please add a
regression case covering this too.
##########
providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py:
##########
@@ -1540,6 +1582,31 @@ async def list_jobs(
page_result: ListJobsAsyncPager = await
client.list_jobs(request=request)
return page_result
+ async def get_latest_job_by_name(
+ self,
+ job_name: str,
+ project_id: str | None = PROVIDE_PROJECT_ID,
+ location: str | None = DEFAULT_DATAFLOW_LOCATION,
+ ) -> Job | None:
+ """
+ Get the latest created job with the given name, in any state.
+
+ Job names are not unique in Dataflow, so if more than one job has this
name, the newest one
+ is returned.
+
+ :param job_name: Name of the job. It is converted to lowercase because
Dataflow job names
+ are always lowercase.
+ :param project_id: Optional. The Google Cloud project ID in which the
job was started.
+ If set to None or missing, the default project_id from the Google
Cloud connection is used.
+ :param location: Optional. The location of the Dataflow job (for
example europe-west1).
+ :return: The job, or None if no job has this name.
+ """
+ job_name = job_name.lower()
+ jobs = await self.list_jobs(project_id=project_id, location=location)
+ matching_jobs = [job async for job in jobs if job.name == job_name]
+ # create_time is typed as protobuf Timestamp, but proto-plus returns
it as a datetime.
+ return max(matching_jobs, key=lambda job: cast("datetime",
job.create_time), default=None)
Review Comment:
Please see my comment above.
##########
providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py:
##########
@@ -1540,6 +1582,31 @@ async def list_jobs(
page_result: ListJobsAsyncPager = await
client.list_jobs(request=request)
return page_result
+ async def get_latest_job_by_name(
+ self,
+ job_name: str,
+ project_id: str | None = PROVIDE_PROJECT_ID,
+ location: str | None = DEFAULT_DATAFLOW_LOCATION,
+ ) -> Job | None:
+ """
+ Get the latest created job with the given name, in any state.
+
+ Job names are not unique in Dataflow, so if more than one job has this
name, the newest one
+ is returned.
+
+ :param job_name: Name of the job. It is converted to lowercase because
Dataflow job names
+ are always lowercase.
+ :param project_id: Optional. The Google Cloud project ID in which the
job was started.
+ If set to None or missing, the default project_id from the Google
Cloud connection is used.
+ :param location: Optional. The location of the Dataflow job (for
example europe-west1).
+ :return: The job, or None if no job has this name.
+ """
+ job_name = job_name.lower()
+ jobs = await self.list_jobs(project_id=project_id, location=location)
Review Comment:
Above comment applies here too. Is it possible to filter via the API?
##########
providers/google/docs/operators/cloud/dataflow.rst:
##########
@@ -410,6 +410,12 @@ This operator can be run in deferrable mode by passing
``deferrable=True`` as a
:start-after: [START howto_sensor_wait_for_job_status_deferrable]
:end-before: [END howto_sensor_wait_for_job_status_deferrable]
+If the job is started outside Airflow and its ID is not known, you can pass
``job_name`` instead of
+``job_id``. Job names are not unique in Dataflow, so the sensor checks the
latest created job with that
+name. If no job with that name exists yet, the sensor keeps waiting. If the
same job name is used for every
+run, add something unique like the date to the name, for example ``my-job-{{
ds_nodash }}``. Otherwise the
+sensor can pick up the previous run before the new run starts.
Review Comment:
Pass ``job_name`` instead of ``job_id`` to monitor the latest created job
with that name. The sensor waits if no match exists. Use a name unique to the
intended run, such as ``my-job-{{ ds_nodash }}``, to avoid selecting a previous
run.
##########
providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py:
##########
@@ -460,6 +462,21 @@ def fetch_job_id(self) -> str | None:
self.log.info("Google Cloud DataFlow job not available yet..")
return self._job_id
+ def fetch_latest_job_by_name(self, job_name: str) -> dict | None:
+ """
+ Return the latest created job with the given name, in any state.
+
+ Dataflow API does not support getting a job by name, so all jobs in
the location are listed
+ and filtered here.
+
+ :param job_name: Name of the job. It is converted to lowercase because
Dataflow job names
+ are always lowercase.
+ :return: The job, or None if no job has this name.
+ """
+ job_name = job_name.lower()
+ jobs = [job for job in self._fetch_all_jobs() if job.get("name") ==
job_name]
Review Comment:
Could we pass name to the list request instead of fetching all jobs and
filtering locally?
--
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]
