Lee-W commented on code in PR #36523:
URL: https://github.com/apache/airflow/pull/36523#discussion_r1443601217
##########
airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -230,36 +230,55 @@ def hook(self) -> BatchClientHook:
region_name=self.region_name,
)
- def execute(self, context: Context):
+ def execute(self, context: Context) -> str | None:
"""Submit and monitor an AWS Batch job.
:raises: AirflowException
"""
self.submit_job(context)
if self.deferrable:
- self.defer(
- timeout=self.execution_timeout,
- trigger=BatchJobTrigger(
- job_id=self.job_id,
- waiter_max_attempts=self.max_retries,
- aws_conn_id=self.aws_conn_id,
- region_name=self.region_name,
- waiter_delay=self.poll_interval,
- ),
- method_name="execute_complete",
- )
+ if not self.job_id:
+ raise AirflowException("AWS Batch job - job_id was not found")
+
+ job = self.hook.get_job_description(self.job_id)
+ job_status = job.get("status")
+ if job_status == self.hook.SUCCESS_STATE:
+ msg = f"{self.job_id} was completed successfully"
+ self.log.info(msg)
+ return self.job_id
+ elif job_status == self.hook.FAILURE_STATE:
+ raise AirflowException(f"{self.job_id} failed")
Review Comment:
@o-nikolas, I tried to refine it a bit. Please let me know if it looks good
to you. Thanks!
--
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
