Google Cloud Scheduler Client Library

Warnings

• breaking The library explicitly requires Python 3.9 or newer. Running on older Python versions (3.8 or below) will result in installation failures or runtime errors.
  Fix: Upgrade your Python environment to 3.9 or higher. For example, using `pyenv` or updating your system's Python installation.
• gotcha Authentication is a common source of errors. The client library uses Application Default Credentials (ADC) by default, but it must be correctly configured in your environment (e.g., via `gcloud auth application-default login` locally, or by attaching a service account to your compute resource in GCP). Incorrect permissions for the executing service account (e.g., lacking `roles/cloudscheduler.admin` or permissions to invoke the target) will lead to 401/403 errors.
• gotcha Google Cloud client libraries often offer multiple API versions (e.g., `v1`, `v1beta1`). While `v1beta1` might contain newer features, it is a beta API and not guaranteed to be stable or backward compatible. Always prefer `v1` for production workloads to ensure stability and avoid unexpected breaking changes.
  Fix: Use `from google.cloud import scheduler_v1` for stable production code. Only use `scheduler_v1beta1` if you explicitly require a beta feature and understand the implications.
• gotcha Incorrectly configuring job targets (HTTP, Pub/Sub, App Engine) can lead to silent failures or unexpected behavior. Pay close attention to `uri`, `http_method`, `headers`, `schedule` (cron format), and `time_zone`. For HTTP targets, ensure the `Content-Type` header is explicitly set if your endpoint expects a specific type, as `application/octet-stream` is a common default.
• gotcha Cloud Scheduler jobs have a default timeout (e.g., 30 minutes for HTTP targets). If your scheduled task is long-running and exceeds this, Cloud Scheduler will report a failure even if the downstream service eventually succeeds. Also, retry configurations must be carefully tuned to avoid overwhelming the target or incurring excessive costs.

Install

• pip install google-cloud-scheduler Install stable version

Imports

• CloudSchedulerClient

  from google.cloud import scheduler_v1

  The client class is nested under the `scheduler_v1` module, representing the v1 API.
• CloudSchedulerClient (v1beta1)

  from google.cloud import scheduler_v1beta1

  Use this import for the beta version of the API. Generally, `v1` is preferred for production unless specific beta features are required.

Quickstart

This quickstart demonstrates how to create a simple Cloud Scheduler job that triggers an HTTP endpoint every hour using the `scheduler_v1` API. Ensure your `GCP_PROJECT_ID`, `GCP_LOCATION_ID`, `SCHEDULER_JOB_NAME`, and `SCHEDULER_TARGET_URL` environment variables are set or replaced in the code. Authentication is handled automatically via Application Default Credentials if running in a Google Cloud environment or with `gcloud auth application-default login` locally.

import os
from google.cloud import scheduler_v1

project_id = os.environ.get('GCP_PROJECT_ID', 'your-project-id')
location_id = os.environ.get('GCP_LOCATION_ID', 'us-central1') # e.g., 'us-central1'
job_name = os.environ.get('SCHEDULER_JOB_NAME', 'my-scheduled-job')
target_url = os.environ.get('SCHEDULER_TARGET_URL', 'https://your-http-endpoint.cloudfunctions.net/myFunction')

# Initialize a client
client = scheduler_v1.CloudSchedulerClient()

# Construct the full resource name of the parent of the job
parent = client.location_path(project_id, location_id)

# Construct the job body
job = scheduler_v1.Job()
job.name = client.job_path(project_id, location_id, job_name)
job.description = 'My first Cloud Scheduler job (Python client)'
job.schedule = '0 * * * *'  # Run every hour
job.time_zone = 'America/Los_Angeles'

# Configure an HTTP target
http_target = scheduler_v1.HttpTarget()
http_target.uri = target_url
http_target.http_method = scheduler_v1.HttpMethod.POST
job.http_target = http_target

# Create the job
try:
    response = client.create_job(parent=parent, job=job)
    print(f'Created job: {response.name}')
except Exception as e:
    print(f'Error creating job: {e}')

# To delete the job (uncomment to run)
# try:
#     client.delete_job(name=response.name)
#     print(f'Deleted job: {response.name}')
# except Exception as e:
#     print(f'Error deleting job: {e}')