Google Cloud Batch

import os
from google.cloud import batch_v1
from google.cloud.batch_v1 import types

def create_simple_container_job(
    project_id: str,
    region: str,
    job_name: str,
) -> types.Job:
    """Creates and runs a simple container job in Google Cloud Batch."""
    client = batch_v1.BatchServiceClient()

    # Define what will be done as part of the job.
    runnable = types.Runnable()
    runnable.container = types.Runnable.Container(
        image_uri="gcr.io/google-containers/busybox",
        entrypoint="/bin/sh",
        commands=[
            "-c",
            "echo Hello world! This is task ${BATCH_TASK_INDEX}. This job has a total of ${BATCH_TASK_COUNT} tasks.",
        ],
    )

    # Jobs can be divided into tasks. In this case, we have one task group with one task.
    task_spec = types.TaskSpec(runnables=[runnable])
    task_group = types.TaskGroup(
        task_spec=task_spec,
        task_count=1,
        parallelism=1,
    )

    # Policies for VM allocation.
    # Using a general purpose machine type like 'e2-standard-4'.
    # Ensure the specified region supports the machine type.
    allocation_policy = types.AllocationPolicy(
        instances=[
            types.AllocationPolicy.InstancePolicyOrTemplate(
                policy=types.AllocationPolicy.InstancePolicy(machine_type="e2-standard-4")
            ),
        ],
        location=types.AllocationPolicy.LocationPolicy(
            allowed_locations=[f"regions/{region}"]
        )
    )

    # Define the job itself.
    job = types.Job(
        name=job_name, # Name needs to be unique per project and region
        task_groups=[task_group],
        allocation_policy=allocation_policy,
        labels={
            "environment": "dev",
            "framework": "batch-quickstart",
        },
        logs_policy=types.LogsPolicy(destination=types.LogsPolicy.Destination.CLOUD_LOGGING),
    )

    request = types.CreateJobRequest(
        parent=f"projects/{project_id}/locations/{region}",
        job_id=job_name,
        job=job,
    )

    response = client.create_job(request=request)
    print(f"Job created: {response.name}")
    return response

if __name__ == "__main__":
    project_id = os.environ.get("GOOGLE_CLOUD_PROJECT", "your-gcp-project-id")
    region = os.environ.get("GOOGLE_CLOUD_REGION", "us-central1") # Choose an available region
    job_id = os.environ.get("BATCH_JOB_ID", "my-sample-batch-job-1") # Unique ID for the job

    if project_id == "your-gcp-project-id":
        print("Please set the GOOGLE_CLOUD_PROJECT environment variable or replace 'your-gcp-project-id'.")
    elif region == "us-central1":
        print("Consider setting the GOOGLE_CLOUD_REGION environment variable or choose a different region.")
    else:
        try:
            created_job = create_simple_container_job(project_id, region, job_id)
            print(f"Monitor job in console: https://console.cloud.google.com/batch/jobs/{region}/{job_id}?project={project_id}")
        except Exception as e:
            print(f"Error creating job: {e}")
            print("Ensure the Batch API is enabled and your service account has 'Batch Job Editor' (roles/batch.jobs.editor) or equivalent permissions.")