dazza-codes commented on a change in pull request #6811: [AIRFLOW-6245] Add
custom waiters for AWS batch jobs
URL: https://github.com/apache/airflow/pull/6811#discussion_r363524955
##########
File path: airflow/providers/amazon/aws/operators/batch.py
##########
@@ -143,39 +107,65 @@ def __init__(
overrides,
array_properties=None,
parameters=None,
+ job_id=None,
+ waiters=None,
max_retries=None,
status_retries=None,
aws_conn_id=None,
region_name=None,
**kwargs,
): # pylint: disable=too-many-arguments
- super().__init__(**kwargs)
+ BaseOperator.__init__(self, **kwargs)
+ AwsBatchClient.__init__(
+ self,
+ max_retries=max_retries,
+ status_retries=status_retries,
+ aws_conn_id=aws_conn_id,
+ region_name=region_name,
+ )
+
+ self.job_id = job_id
self.job_name = job_name
- self.aws_conn_id = aws_conn_id
- self.region_name = region_name
self.job_definition = job_definition
self.job_queue = job_queue
self.overrides = overrides
self.array_properties = array_properties or {}
self.parameters = parameters
- self.max_retries = max_retries or self.MAX_RETRIES
- self.status_retries = status_retries or self.STATUS_RETRIES
+ self.waiters = waiters
- self.jobId = None # pylint: disable=invalid-name
- self.jobName = None # pylint: disable=invalid-name
+ def execute(self, context: Dict):
+ """
+ Submit and monitor an AWS Batch job
- self.hook = self.get_hook()
+ :raises: AirflowException
+ """
+ try:
+ self.submit_job(context)
+ self.monitor_job(context)
- def execute(self, context):
+ except Exception as e:
+ self.log.info("AWS Batch job (%s) failed", self.job_id)
+ raise AirflowException(e)
Review comment:
Thanks, good catch. Will revise in next push.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
