`airflow.providers.amazon.aws.hooks.batch_waiters`¶

AWS Batch service waiters.

Module Contents¶

Classes¶

BatchWaitersHook

A utility to manage waiters for AWS Batch services.

class airflow.providers.amazon.aws.hooks.batch_waiters.BatchWaitersHook(*args, waiter_config=None, **kwargs)[source]¶

Bases: airflow.providers.amazon.aws.hooks.batch_client.BatchClientHook

A utility to manage waiters for AWS Batch services.

import random
from airflow.providers.amazon.aws.operators.batch_waiters import BatchWaiters

# to inspect default waiters
waiters = BatchWaiters()
config = waiters.default_config  # type: Dict
waiter_names = waiters.list_waiters()  # -> ["JobComplete", "JobExists", "JobRunning"]

# The default_config is a useful stepping stone to creating custom waiters, e.g.
custom_config = waiters.default_config  # this is a deepcopy
# modify custom_config['waiters'] as necessary and get a new instance:
waiters = BatchWaiters(waiter_config=custom_config)
waiters.waiter_config  # check the custom configuration (this is a deepcopy)
waiters.list_waiters()  # names of custom waiters

# During the init for BatchWaiters, the waiter_config is used to build a waiter_model;
# and note that this only occurs during the class init, to avoid any accidental mutations
# of waiter_config leaking into the waiter_model.
waiters.waiter_model  # -> botocore.waiter.WaiterModel object

# The waiter_model is combined with the waiters.client to get a specific waiter
# and the details of the config on that waiter can be further modified without any
# accidental impact on the generation of new waiters from the defined waiter_model, e.g.
waiters.get_waiter("JobExists").config.delay  # -> 5
waiter = waiters.get_waiter("JobExists")  # -> botocore.waiter.Batch.Waiter.JobExists object
waiter.config.delay = 10
waiters.get_waiter("JobExists").config.delay  # -> 5 as defined by waiter_model

# To use a specific waiter, update the config and call the `wait()` method for jobId, e.g.
waiter = waiters.get_waiter("JobExists")  # -> botocore.waiter.Batch.Waiter.JobExists object
waiter.config.delay = random.uniform(1, 10)  # seconds
waiter.config.max_attempts = 10
waiter.wait(jobs=[jobId])

See also

Parameters

waiter_config (dict | None) – a custom waiter configuration for AWS Batch services
aws_conn_id – connection id of AWS credentials / region name. If None, credential boto3 strategy will be used (https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html).
region_name – region name to use in AWS client. Override the AWS region in connection (if provided)

property default_config: dict[source]¶

An immutable default waiter configuration.

Returns: a waiter configuration for AWS Batch services
Return type: dict

property waiter_config: dict[source]¶

An immutable waiter configuration for this instance; a deepcopy is returned by this property.

During the init for BatchWaiters, the waiter_config is used to build a waiter_model and this only occurs during the class init, to avoid any accidental mutations of waiter_config leaking into the waiter_model.

Returns: a waiter configuration for AWS Batch services
Return type: dict

property waiter_model: botocore.waiter.WaiterModel[source]¶

A configured waiter model used to generate waiters on AWS Batch services.

Returns: a waiter model for AWS Batch services
Return type: botocore.waiter.WaiterModel

get_waiter(waiter_name, _=None, deferrable=False, client=None)[source]¶

Get an AWS Batch service waiter, using the configured .waiter_model.

The .waiter_model is combined with the .client to get a specific waiter and the properties of that waiter can be modified without any accidental impact on the generation of new waiters from the .waiter_model, e.g.

waiters.get_waiter("JobExists").config.delay  # -> 5
waiter = waiters.get_waiter("JobExists")  # a new waiter object
waiter.config.delay = 10
waiters.get_waiter("JobExists").config.delay  # -> 5 as defined by waiter_model

To use a specific waiter, update the config and call the wait() method for jobId, e.g.

import random

waiter = waiters.get_waiter("JobExists")  # a new waiter object
waiter.config.delay = random.uniform(1, 10)  # seconds
waiter.config.max_attempts = 10
waiter.wait(jobs=[jobId])

Parameters

waiter_name (str) – The name of the waiter. The name should match the name (including the casing) of the key name in the waiter model file (typically this is CamelCasing); see .list_waiters.
_ (dict[str, str] | None) – unused, just here to match the method signature in base_aws

Returns

a waiter object for the named AWS Batch service

Return type

botocore.waiter.Waiter

list_waiters()[source]¶

List the waiters in a waiter configuration for AWS Batch services.

Returns: waiter names for AWS Batch services
Return type: list[str]

wait_for_job(job_id, delay=None, get_batch_log_fetcher=None)[source]¶

Wait for Batch job to complete.

This assumes that the .waiter_model is configured using some variation of the .default_config so that it can generate waiters with the following names: “JobExists”, “JobRunning” and “JobComplete”.

Parameters

job_id (str) – a Batch job ID
delay (int | float | None) – A delay before polling for job status
get_batch_log_fetcher (Callable[[str], airflow.providers.amazon.aws.utils.task_log_fetcher.AwsTaskLogFetcher | None] | None) – A method that returns batch_log_fetcher of type AwsTaskLogFetcher or None when the CloudWatch log stream hasn’t been created yet.

Raises

AirflowException

Note

This method adds a small random jitter to the delay (+/- 2 sec, >= 1 sec). Using a random interval helps to avoid AWS API throttle limits when many concurrent tasks request job-descriptions.

It also modifies the max_attempts to use the sys.maxsize, which allows Airflow to manage the timeout on waiting.

airflow.providers.amazon.aws.hooks.batch_waiters¶

Module Contents¶

Classes¶

`airflow.providers.amazon.aws.hooks.batch_waiters`¶