Batch Create

Handler for creating and preparing AWS Batch job definitions.

---

AWS Batch job definition creation handlers.

Provides Lambda handlers for creating and registering AWS Batch job definitions.

CreateDefinitionAndPrepareArgsHandler dataclass

CreateDefinitionAndPrepareArgsHandler()

Bases: LambdaHandler[CreateDefinitionAndPrepareArgsRequest, CreateDefinitionAndPrepareArgsResponse]

Handler for creating AWS Batch job definitions.

Registers a job definition with AWS Batch and prepares the arguments needed for job submission.

handle

handle(request)

Create a job definition and prepare submission arguments.

Parameters:

Name	Type	Description	Default
request	CreateDefinitionAndPrepareArgsRequest	Request containing job definition configuration.	required

Returns:

Type	Description
CreateDefinitionAndPrepareArgsResponse	Response containing the job definition ARN and submission args.

Raises:

Type	Description
ValueError	If job_queue_name is not provided.

Source code in src/aibs_informatics_aws_lambda/handlers/batch/create.py

def handle(
    self, request: CreateDefinitionAndPrepareArgsRequest
) -> CreateDefinitionAndPrepareArgsResponse:
    """Create a job definition and prepare submission arguments.

    Args:
        request (CreateDefinitionAndPrepareArgsRequest): Request containing
            job definition configuration.

    Returns:
        Response containing the job definition ARN and submission args.

    Raises:
        ValueError: If job_queue_name is not provided.
    """
    job_def_builder = BatchJobBuilder(
        image=request.image
        if DockerImageUri.is_valid(request.image)
        else resolve_image_uri(request.image),
        job_definition_name=request.job_definition_name,
        job_name=request.job_name or f"{request.job_definition_name}-{uuid_str()}",
        command=request.command,
        environment=request.environment,
        job_definition_tags=request.job_definition_tags,
        resource_requirements=request.resource_requirements,
        mount_points=request.mount_points,
        volumes=request.volumes,
        privileged=request.privileged,
        job_role_arn=request.job_role_arn,
    )

    response = register_job_definition(
        job_definition_name=job_def_builder.job_definition_name,
        container_properties=job_def_builder.container_properties,
        retry_strategy=request.retry_strategy or build_retry_strategy(),
        parameters=None,
        tags=job_def_builder.job_definition_tags,
    )
    job_definition_arn = response["jobDefinitionArn"]

    job_queue_name = request.job_queue_name
    if not job_queue_name:
        raise ValueError("job_queue_name must be provided")
    return CreateDefinitionAndPrepareArgsResponse(
        job_name=job_def_builder.job_name,
        job_definition_arn=job_definition_arn,
        job_queue_arn=job_queue_name,
        parameters={},
        container_overrides=job_def_builder.container_overrides__sfn,
    )

DockerImageUri

Bases: ValidatedStr

Validated string representing a Docker image URI.

Validates and parses Docker image URIs from various registries including Docker Hub, ECR, and GitHub Container Registry.

Note

This class is intended to be moved to aibs-informatics-core once it is considered stable; that refactor is tracked in the team's external issue tracker.

Attributes:

Name	Type	Description
regex_pattern	Pattern	Pattern for validating Docker image URIs.

registry property

registry

Get the registry portion of the image URI.

Returns:

Type	Description
str	The registry hostname.

repo_name property

repo_name

Get the repository name.

Returns:

Type	Description
str	The repository name.

sha256 property

sha256

Get the SHA256 digest if specified.

Returns:

Type	Description
Optional[str]	The SHA256 digest, or None if using a tag.

tag property

tag

Get the image tag if specified.

Returns:

Type	Description
Optional[str]	The tag, or None if using a SHA digest.