From e8eb84ad5bcf9b43966e1f9f5dee24f2716f70d0 Mon Sep 17 00:00:00 2001 From: Jennifer Davis Date: Fri, 6 Nov 2026 20:18:31 +0000 Subject: [PATCH 1/4] feat(cloud_scheduler): modernize samples and cover documented job tasks The Cloud Scheduler Python samples covered only creating and deleting a job with an App Engine target. They returned values instead of printing results, had no error handling, and pinned an old library version. The App Engine handler app used a retired runtime. This change adds a sample for each documented job task, using google-cloud-scheduler 2.21.0, with one task and one region tag per file: - cloudscheduler_create_job: now shows the recommended path, an HTTP target that calls a Cloud Run service with an OIDC token. - cloudscheduler_create_appengine_job: the App Engine target sample, moved from create_job.py. - New samples: - cloudscheduler_create_http_job_oauth - cloudscheduler_create_pubsub_job - cloudscheduler_create_job_with_retry - cloudscheduler_update_job - cloudscheduler_pause_job - cloudscheduler_resume_job - cloudscheduler_run_job - cloudscheduler_get_job - cloudscheduler_list_jobs - cloudscheduler_delete_job: modernized. It also adds cloudscheduler_receiver, a Cloud Run function that handles scheduled requests and skips duplicate deliveries. It replaces the App Engine handler app, retiring the cloudscheduler_app and cloudscheduler_python_yaml region tags. Tests live in tests/ subdirectories: - Unit tests for every sample and for the receiver, using a mocked client. These run without Google Cloud credentials. - Integration tests that run every sample live, including a job that calls the receiver end to end. They are skipped unless GOOGLE_CLOUD_PROJECT and SERVICE_URL are set. Unit and integration tests pass on Python 3.13. --- cloud_scheduler/receiver/.python-version | 1 + .../app.yaml => receiver/__init__.py} | 13 +- cloud_scheduler/receiver/main.py | 70 +++++ cloud_scheduler/receiver/noxfile_config.py | 42 +++ .../receiver/requirements-test.txt | 1 + cloud_scheduler/receiver/requirements.txt | 1 + cloud_scheduler/receiver/tests/__init__.py | 13 + cloud_scheduler/receiver/tests/main_test.py | 62 +++++ cloud_scheduler/snippets/__init__.py | 13 + .../snippets/create_appengine_job.py | 93 +++++++ .../snippets/create_http_job_oauth.py | 118 ++++++++ cloud_scheduler/snippets/create_job.py | 118 +++++--- .../snippets/create_job_with_retry.py | 120 +++++++++ cloud_scheduler/snippets/create_pubsub_job.py | 91 +++++++ cloud_scheduler/snippets/delete_job.py | 58 ++-- cloud_scheduler/snippets/get_job.py | 72 +++++ cloud_scheduler/snippets/job_test.py | 69 ----- cloud_scheduler/snippets/list_jobs.py | 67 +++++ cloud_scheduler/snippets/main.py | 44 --- cloud_scheduler/snippets/main_test.py | 42 --- cloud_scheduler/snippets/noxfile_config.py | 2 +- cloud_scheduler/snippets/pause_job.py | 74 +++++ cloud_scheduler/snippets/requirements.txt | 5 +- cloud_scheduler/snippets/resume_job.py | 74 +++++ cloud_scheduler/snippets/run_job.py | 72 +++++ cloud_scheduler/snippets/tests/__init__.py | 13 + .../tests/create_appengine_job_test.py | 91 +++++++ .../tests/create_http_job_oauth_test.py | 105 ++++++++ .../snippets/tests/create_job_test.py | 104 +++++++ .../tests/create_job_with_retry_test.py | 103 +++++++ .../snippets/tests/create_pubsub_job_test.py | 96 +++++++ .../snippets/tests/delete_job_test.py | 87 ++++++ .../snippets/tests/get_job_test.py | 88 ++++++ .../snippets/tests/integration_test.py | 255 ++++++++++++++++++ .../snippets/tests/list_jobs_test.py | 90 +++++++ .../snippets/tests/pause_job_test.py | 87 ++++++ .../snippets/tests/resume_job_test.py | 87 ++++++ .../snippets/tests/run_job_test.py | 87 ++++++ .../snippets/tests/update_job_test.py | 89 ++++++ cloud_scheduler/snippets/update_job.py | 80 ++++++ 40 files changed, 2580 insertions(+), 217 deletions(-) create mode 100644 cloud_scheduler/receiver/.python-version rename cloud_scheduler/{snippets/app.yaml => receiver/__init__.py} (54%) create mode 100644 cloud_scheduler/receiver/main.py create mode 100644 cloud_scheduler/receiver/noxfile_config.py create mode 100644 cloud_scheduler/receiver/requirements-test.txt create mode 100644 cloud_scheduler/receiver/requirements.txt create mode 100644 cloud_scheduler/receiver/tests/__init__.py create mode 100644 cloud_scheduler/receiver/tests/main_test.py create mode 100644 cloud_scheduler/snippets/__init__.py create mode 100644 cloud_scheduler/snippets/create_appengine_job.py create mode 100644 cloud_scheduler/snippets/create_http_job_oauth.py create mode 100644 cloud_scheduler/snippets/create_job_with_retry.py create mode 100644 cloud_scheduler/snippets/create_pubsub_job.py create mode 100644 cloud_scheduler/snippets/get_job.py delete mode 100644 cloud_scheduler/snippets/job_test.py create mode 100644 cloud_scheduler/snippets/list_jobs.py delete mode 100644 cloud_scheduler/snippets/main.py delete mode 100644 cloud_scheduler/snippets/main_test.py create mode 100644 cloud_scheduler/snippets/pause_job.py create mode 100644 cloud_scheduler/snippets/resume_job.py create mode 100644 cloud_scheduler/snippets/run_job.py create mode 100644 cloud_scheduler/snippets/tests/__init__.py create mode 100644 cloud_scheduler/snippets/tests/create_appengine_job_test.py create mode 100644 cloud_scheduler/snippets/tests/create_http_job_oauth_test.py create mode 100644 cloud_scheduler/snippets/tests/create_job_test.py create mode 100644 cloud_scheduler/snippets/tests/create_job_with_retry_test.py create mode 100644 cloud_scheduler/snippets/tests/create_pubsub_job_test.py create mode 100644 cloud_scheduler/snippets/tests/delete_job_test.py create mode 100644 cloud_scheduler/snippets/tests/get_job_test.py create mode 100644 cloud_scheduler/snippets/tests/integration_test.py create mode 100644 cloud_scheduler/snippets/tests/list_jobs_test.py create mode 100644 cloud_scheduler/snippets/tests/pause_job_test.py create mode 100644 cloud_scheduler/snippets/tests/resume_job_test.py create mode 100644 cloud_scheduler/snippets/tests/run_job_test.py create mode 100644 cloud_scheduler/snippets/tests/update_job_test.py create mode 100644 cloud_scheduler/snippets/update_job.py diff --git a/cloud_scheduler/receiver/.python-version b/cloud_scheduler/receiver/.python-version new file mode 100644 index 00000000000..24ee5b1be99 --- /dev/null +++ b/cloud_scheduler/receiver/.python-version @@ -0,0 +1 @@ +3.13 diff --git a/cloud_scheduler/snippets/app.yaml b/cloud_scheduler/receiver/__init__.py similarity index 54% rename from cloud_scheduler/snippets/app.yaml rename to cloud_scheduler/receiver/__init__.py index e0e08d4312e..58d482ea386 100644 --- a/cloud_scheduler/snippets/app.yaml +++ b/cloud_scheduler/receiver/__init__.py @@ -1,18 +1,13 @@ -# Copyright 2019 Google LLC +# Copyright 2026 Google LLC # -# Licensed under the Apache License, Version 2.0 (the 'License'); +# Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # -# http://www.apache.org/licenses/LICENSE-2.0 +# http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an 'AS IS' BASIS, +# distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. - -# [START cloudscheduler_python_yaml] -runtime: python37 -service: my-service -# [END cloudscheduler_python_yaml] diff --git a/cloud_scheduler/receiver/main.py b/cloud_scheduler/receiver/main.py new file mode 100644 index 00000000000..982fdccc823 --- /dev/null +++ b/cloud_scheduler/receiver/main.py @@ -0,0 +1,70 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_receiver] +import json + +import flask +import functions_framework + +# Cloud Scheduler can deliver the same scheduled run more than once, so the +# function records which runs it has finished. This in-memory set only lasts +# for the life of one instance; store run keys in a durable database such as +# Firestore so that every instance sees them. +processed_runs: set[str] = set() + + +def log(message: str, **fields: str) -> None: + """Writes a structured log entry that Cloud Logging parses from stdout.""" + print(json.dumps({"severity": "INFO", "message": message, **fields})) + + +@functions_framework.http +def handle_scheduled_run(request: flask.Request) -> tuple[str, int]: + """Handles a request that a Cloud Scheduler job sends to this function. + + Deploy the function without unauthenticated access. Cloud Run functions + then verifies the job's OIDC token before the request reaches this + function, and only service accounts with the Cloud Run Invoker role can + call it. + """ + job_name = request.headers.get("X-CloudScheduler-JobName") + schedule_time = request.headers.get("X-CloudScheduler-ScheduleTime") + if not job_name or not schedule_time: + return "error: missing Cloud Scheduler headers", 400 + + # The schedule time stays the same when Cloud Scheduler retries a run, + # so the job name and schedule time identify a single scheduled run. + run_key = f"{job_name}/{schedule_time}" + if run_key in processed_runs: + log("Skipped duplicate run", jobName=job_name, scheduleTime=schedule_time) + return "Run already processed", 200 + + # Finish the work within the job's attempt deadline. Cloud Scheduler + # treats a slow response as a failure even if the work later succeeds. + # If the work raises an exception, the function returns HTTP 500, which + # tells Cloud Scheduler to retry the run according to the job's retry + # settings. + payload = request.get_data(as_text=True) + log( + "Processed scheduled run", + jobName=job_name, + scheduleTime=schedule_time, + payload=payload, + ) + processed_runs.add(run_key) + return "Run processed", 200 + + +# [END cloudscheduler_receiver] diff --git a/cloud_scheduler/receiver/noxfile_config.py b/cloud_scheduler/receiver/noxfile_config.py new file mode 100644 index 00000000000..95c2d9ac875 --- /dev/null +++ b/cloud_scheduler/receiver/noxfile_config.py @@ -0,0 +1,42 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Default TEST_CONFIG_OVERRIDE for python repos. + +# You can copy this file into your directory, then it will be imported from +# the noxfile.py. + +# The source of truth: +# https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/noxfile_config.py + +TEST_CONFIG_OVERRIDE = { + # You can opt out from the test for specific Python versions. + "ignored_versions": ["2.7", "3.7", "3.8", "3.9", "3.10", "3.11", "3.12"], + # Old samples are opted out of enforcing Python type hints + # All new samples should feature them + "enforce_type_hints": True, + # An envvar key for determining the project id to use. Change it + # to 'BUILD_SPECIFIC_GCLOUD_PROJECT' if you want to opt in using a + # build specific Cloud project. You can also use your own string + # to use your own Cloud project. + "gcloud_project_env": "GOOGLE_CLOUD_PROJECT", + # 'gcloud_project_env': 'BUILD_SPECIFIC_GCLOUD_PROJECT', + # If you need to use a specific version of pip, + # change pip_version_override to the string representation + # of the version number, for example, "20.2.4" + "pip_version_override": None, + # A dictionary you want to inject into your test. Don't put any + # secrets here. These values will override predefined values. + "envs": {}, +} diff --git a/cloud_scheduler/receiver/requirements-test.txt b/cloud_scheduler/receiver/requirements-test.txt new file mode 100644 index 00000000000..b3b2b3ca4cc --- /dev/null +++ b/cloud_scheduler/receiver/requirements-test.txt @@ -0,0 +1 @@ +pytest==9.0.3 diff --git a/cloud_scheduler/receiver/requirements.txt b/cloud_scheduler/receiver/requirements.txt new file mode 100644 index 00000000000..ddcee07f14a --- /dev/null +++ b/cloud_scheduler/receiver/requirements.txt @@ -0,0 +1 @@ +functions-framework==3.10.2 diff --git a/cloud_scheduler/receiver/tests/__init__.py b/cloud_scheduler/receiver/tests/__init__.py new file mode 100644 index 00000000000..58d482ea386 --- /dev/null +++ b/cloud_scheduler/receiver/tests/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/cloud_scheduler/receiver/tests/main_test.py b/cloud_scheduler/receiver/tests/main_test.py new file mode 100644 index 00000000000..050319d8369 --- /dev/null +++ b/cloud_scheduler/receiver/tests/main_test.py @@ -0,0 +1,62 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import uuid + +import flask +import pytest + +from .. import main + +app = flask.Flask(__name__) + + +def call_function( + headers: dict[str, str] | None = None, data: str = "example-payload" +) -> tuple[str, int]: + with app.test_request_context("/", method="POST", headers=headers, data=data): + return main.handle_scheduled_run(flask.request) + + +def scheduler_headers() -> dict[str, str]: + return { + "X-CloudScheduler-JobName": f"test-job-{uuid.uuid4().hex[:8]}", + "X-CloudScheduler-ScheduleTime": "2026-11-06T09:00:00Z", + } + + +def test_first_delivery_is_processed(capsys: pytest.CaptureFixture[str]) -> None: + body, status = call_function(scheduler_headers()) + + assert status == 200 + assert body == "Run processed" + assert "example-payload" in capsys.readouterr().out + + +def test_duplicate_delivery_is_skipped(capsys: pytest.CaptureFixture[str]) -> None: + headers = scheduler_headers() + call_function(headers) + + body, status = call_function(headers) + + assert status == 200 + assert body == "Run already processed" + assert "Skipped duplicate run" in capsys.readouterr().out + + +def test_missing_scheduler_headers_are_rejected() -> None: + body, status = call_function() + + assert status == 400 + assert body.startswith("error:") diff --git a/cloud_scheduler/snippets/__init__.py b/cloud_scheduler/snippets/__init__.py new file mode 100644 index 00000000000..58d482ea386 --- /dev/null +++ b/cloud_scheduler/snippets/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/cloud_scheduler/snippets/create_appengine_job.py b/cloud_scheduler/snippets/create_appengine_job.py new file mode 100644 index 00000000000..22358c8cc2c --- /dev/null +++ b/cloud_scheduler/snippets/create_appengine_job.py @@ -0,0 +1,93 @@ +# Copyright 2019 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_create_appengine_job] +import datetime +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 + + +def create_appengine_job(project_id: str, location_id: str, job_id: str) -> None: + """Creates a job that sends requests to an App Engine service on a schedule. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region of the project's App Engine app (for example, + 'us-central1'). + job_id: The ID for the new job (for example, 'example-job-id'). + """ + client = scheduler_v1.CloudSchedulerClient() + # App Engine targets must be in the same project and region as the + # App Engine app. + parent = f"projects/{project_id}/locations/{location_id}" + + app_engine_routing = scheduler_v1.AppEngineRouting(service="default") + app_engine_target = scheduler_v1.AppEngineHttpTarget( + app_engine_routing=app_engine_routing, + relative_uri="/example-handler", + http_method=scheduler_v1.HttpMethod.POST, + body=b"example-payload", + ) + job = scheduler_v1.Job( + name=f"{parent}/jobs/{job_id}", + app_engine_http_target=app_engine_target, + # Runs every Monday at 09:00. UTC avoids skipped or repeated runs when + # daylight saving time starts or ends. + schedule="0 9 * * 1", + time_zone="UTC", + # Cloud Scheduler reports a failure if the handler doesn't respond + # within this deadline. + attempt_deadline=datetime.timedelta(minutes=5), + ) + request = scheduler_v1.CreateJobRequest(parent=parent, job=job) + + try: + response = client.create_job(request=request) + except exceptions.AlreadyExists as err: + print( + f"error: job '{job_id}' already exists in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to create job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + print(f"Schedule: {response.schedule}") + print(f"State: {response.state.name}") + print(f"Time Zone: {response.time_zone}") + + +# [END cloudscheduler_create_appengine_job] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + create_appengine_job( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + ) diff --git a/cloud_scheduler/snippets/create_http_job_oauth.py b/cloud_scheduler/snippets/create_http_job_oauth.py new file mode 100644 index 00000000000..d1c218d36c6 --- /dev/null +++ b/cloud_scheduler/snippets/create_http_job_oauth.py @@ -0,0 +1,118 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_create_http_job_oauth] +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 + + +def create_http_job_oauth( + project_id: str, + location_id: str, + job_id: str, + workflow_template_id: str, + service_account_email: str, +) -> None: + """Creates a job that starts a Managed Spark workflow template on a schedule. + + The job calls the Dataproc API directly, authenticating with an OAuth + access token for a service account. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region of the job and the workflow template (for + example, 'us-central1'). + job_id: The ID for the new job (for example, 'example-job-id'). + workflow_template_id: The ID of the workflow template to start (for + example, 'example-workflow-template'). + service_account_email: The service account that the job uses to get an + OAuth token. It needs permission to start the workflow template. + """ + client = scheduler_v1.CloudSchedulerClient() + parent = f"projects/{project_id}/locations/{location_id}" + template_name = ( + f"projects/{project_id}/regions/{location_id}" + f"/workflowTemplates/{workflow_template_id}" + ) + + # Use an OAuth token, not an OIDC token, when the target is a Google API + # hosted on *.googleapis.com. + oauth_token = scheduler_v1.OAuthToken( + service_account_email=service_account_email, + scope="https://www.googleapis.com/auth/cloud-platform", + ) + http_target = scheduler_v1.HttpTarget( + uri=f"https://dataproc.googleapis.com/v1/{template_name}:instantiate", + http_method=scheduler_v1.HttpMethod.POST, + headers={"Content-Type": "application/json"}, + body=b"{}", + oauth_token=oauth_token, + ) + job = scheduler_v1.Job( + name=f"{parent}/jobs/{job_id}", + http_target=http_target, + # Runs every Monday at 09:00. UTC avoids skipped or repeated runs when + # daylight saving time starts or ends. + schedule="0 9 * * 1", + time_zone="UTC", + ) + request = scheduler_v1.CreateJobRequest(parent=parent, job=job) + + try: + response = client.create_job(request=request) + except exceptions.AlreadyExists as err: + print( + f"error: job '{job_id}' already exists in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to create job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + print(f"Schedule: {response.schedule}") + print(f"State: {response.state.name}") + print(f"Target URI: {response.http_target.uri}") + + +# [END cloudscheduler_create_http_job_oauth] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + create_http_job_oauth( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + workflow_template_id=arg( + 4, "WORKFLOW_TEMPLATE_ID", "example-workflow-template" + ), + service_account_email=arg( + 5, + "SERVICE_ACCOUNT_EMAIL", + "example-invoker@example-project-id.iam.gserviceaccount.com", + ), + ) diff --git a/cloud_scheduler/snippets/create_job.py b/cloud_scheduler/snippets/create_job.py index 94e52c295d3..bfde35bbcb1 100644 --- a/cloud_scheduler/snippets/create_job.py +++ b/cloud_scheduler/snippets/create_job.py @@ -1,60 +1,112 @@ -# Copyright 2019 Google LLC +# Copyright 2026 Google LLC # -# Licensed under the Apache License, Version 2.0 (the 'License'); +# Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # -# http://www.apache.org/licenses/LICENSE-2.0 +# http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an 'AS IS' BASIS, +# distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # [START cloudscheduler_create_job] +import datetime +import sys +from google.api_core import exceptions from google.cloud import scheduler_v1 -def create_scheduler_job( - project_id: str, location_id: str, service_id: str -) -> scheduler_v1.Job: - """Create a job with an App Engine target via the Cloud Scheduler API. +def create_job( + project_id: str, + location_id: str, + job_id: str, + service_url: str, + service_account_email: str, +) -> None: + """Creates a job that calls a Cloud Run service on a schedule. - Args: - project_id: The Google Cloud project id. - location_id: The location for the job. - service_id: An unique service id for the job. + The job authenticates with an OIDC token for a service account, so the + Cloud Run service doesn't need to allow unauthenticated access. - Returns: - The created job. + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region for the job (for example, 'us-central1'). + job_id: The ID for the new job (for example, 'example-job-id'). + service_url: The HTTPS URL of the Cloud Run service to call. + service_account_email: The service account that the job uses to get an + OIDC token. It needs the Cloud Run Invoker role on the service. """ - - # Create a client. client = scheduler_v1.CloudSchedulerClient() + parent = f"projects/{project_id}/locations/{location_id}" - # Construct the job. + # Cloud Run accepts the token only if its audience matches the service URL. + oidc_token = scheduler_v1.OidcToken( + service_account_email=service_account_email, + audience=service_url, + ) + http_target = scheduler_v1.HttpTarget( + uri=service_url, + http_method=scheduler_v1.HttpMethod.POST, + body=b"example-payload", + oidc_token=oidc_token, + ) job = scheduler_v1.Job( - app_engine_http_target=scheduler_v1.AppEngineHttpTarget( - app_engine_routing=scheduler_v1.AppEngineRouting(service=service_id), - relative_uri="/log_payload", - http_method=scheduler_v1.HttpMethod.POST, - body=b"Hello World", - ), - schedule="* * * * *", - time_zone="America/Los_Angeles", + name=f"{parent}/jobs/{job_id}", + http_target=http_target, + # Runs every Monday at 09:00. UTC avoids skipped or repeated runs when + # daylight saving time starts or ends. + schedule="0 9 * * 1", + time_zone="UTC", + # Cloud Scheduler reports a failure if the service doesn't respond + # within this deadline, so match it to the service's request timeout. + attempt_deadline=datetime.timedelta(minutes=5), ) + request = scheduler_v1.CreateJobRequest(parent=parent, job=job) - # Use the client to send the job creation request. - response = client.create_job( - scheduler_v1.CreateJobRequest( - parent=client.common_location_path(project_id, location_id), - job=job, + try: + response = client.create_job(request=request) + except exceptions.AlreadyExists as err: + print( + f"error: job '{job_id}' already exists in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, ) - ) - - return response + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to create job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + print(f"Schedule: {response.schedule}") + print(f"State: {response.state.name}") + print(f"Time Zone: {response.time_zone}") # [END cloudscheduler_create_job] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + create_job( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + service_url=arg(4, "SERVICE_URL", "https://example-service.a.run.app/"), + service_account_email=arg( + 5, + "SERVICE_ACCOUNT_EMAIL", + "example-invoker@example-project-id.iam.gserviceaccount.com", + ), + ) diff --git a/cloud_scheduler/snippets/create_job_with_retry.py b/cloud_scheduler/snippets/create_job_with_retry.py new file mode 100644 index 00000000000..59c02007a8b --- /dev/null +++ b/cloud_scheduler/snippets/create_job_with_retry.py @@ -0,0 +1,120 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_create_job_with_retry] +import datetime +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 + + +def create_job_with_retry( + project_id: str, + location_id: str, + job_id: str, + service_url: str, + service_account_email: str, +) -> None: + """Creates a job that retries failed calls to a Cloud Run service. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region for the job (for example, 'us-central1'). + job_id: The ID for the new job (for example, 'example-job-id'). + service_url: The HTTPS URL of the Cloud Run service to call. + service_account_email: The service account that the job uses to get an + OIDC token. It needs the Cloud Run Invoker role on the service. + """ + client = scheduler_v1.CloudSchedulerClient() + parent = f"projects/{project_id}/locations/{location_id}" + + oidc_token = scheduler_v1.OidcToken( + service_account_email=service_account_email, + audience=service_url, + ) + http_target = scheduler_v1.HttpTarget( + uri=service_url, + http_method=scheduler_v1.HttpMethod.POST, + body=b"example-payload", + oidc_token=oidc_token, + ) + # Cloud Scheduler stops retrying a run when it reaches either limit: + # retry_count retries or max_retry_duration since the first attempt. + # The wait between retries starts at min_backoff_duration, doubles + # max_doublings times, and then grows linearly up to max_backoff_duration. + retry_config = scheduler_v1.RetryConfig( + retry_count=3, + max_retry_duration=datetime.timedelta(hours=1), + min_backoff_duration=datetime.timedelta(seconds=10), + max_backoff_duration=datetime.timedelta(minutes=5), + max_doublings=3, + ) + job = scheduler_v1.Job( + name=f"{parent}/jobs/{job_id}", + http_target=http_target, + retry_config=retry_config, + # Runs every Monday at 09:00. UTC avoids skipped or repeated runs when + # daylight saving time starts or ends. + schedule="0 9 * * 1", + time_zone="UTC", + # A response that misses this deadline counts as a failed attempt and + # is retried, so match it to the service's request timeout. + attempt_deadline=datetime.timedelta(minutes=5), + ) + request = scheduler_v1.CreateJobRequest(parent=parent, job=job) + + try: + response = client.create_job(request=request) + except exceptions.AlreadyExists as err: + print( + f"error: job '{job_id}' already exists in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to create job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + print(f"Max Retry Duration: {response.retry_config.max_retry_duration}") + print(f"Retry Count: {response.retry_config.retry_count}") + print(f"State: {response.state.name}") + + +# [END cloudscheduler_create_job_with_retry] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + create_job_with_retry( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + service_url=arg(4, "SERVICE_URL", "https://example-service.a.run.app/"), + service_account_email=arg( + 5, + "SERVICE_ACCOUNT_EMAIL", + "example-invoker@example-project-id.iam.gserviceaccount.com", + ), + ) diff --git a/cloud_scheduler/snippets/create_pubsub_job.py b/cloud_scheduler/snippets/create_pubsub_job.py new file mode 100644 index 00000000000..e1eb0191fb4 --- /dev/null +++ b/cloud_scheduler/snippets/create_pubsub_job.py @@ -0,0 +1,91 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_create_pubsub_job] +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 + + +def create_pubsub_job( + project_id: str, location_id: str, job_id: str, topic_id: str +) -> None: + """Creates a job that publishes a Pub/Sub message on a schedule. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region for the job (for example, 'us-central1'). + job_id: The ID for the new job (for example, 'example-job-id'). + topic_id: The ID of an existing Pub/Sub topic in the project (for + example, 'example-topic'). + """ + client = scheduler_v1.CloudSchedulerClient() + parent = f"projects/{project_id}/locations/{location_id}" + + # Cloud Scheduler publishes by using its service agent, which needs the + # Cloud Scheduler Service Agent role in the project. + pubsub_target = scheduler_v1.PubsubTarget( + topic_name=f"projects/{project_id}/topics/{topic_id}", + data=b"example-payload", + attributes={"example-key": "example-value"}, + ) + job = scheduler_v1.Job( + name=f"{parent}/jobs/{job_id}", + pubsub_target=pubsub_target, + # Runs every Monday at 09:00. UTC avoids skipped or repeated runs when + # daylight saving time starts or ends. + schedule="0 9 * * 1", + time_zone="UTC", + ) + request = scheduler_v1.CreateJobRequest(parent=parent, job=job) + + try: + response = client.create_job(request=request) + except exceptions.AlreadyExists as err: + print( + f"error: job '{job_id}' already exists in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to create job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + print(f"Schedule: {response.schedule}") + print(f"State: {response.state.name}") + print(f"Topic: {response.pubsub_target.topic_name}") + + +# [END cloudscheduler_create_pubsub_job] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + create_pubsub_job( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + topic_id=arg(4, "TOPIC_ID", "example-topic"), + ) diff --git a/cloud_scheduler/snippets/delete_job.py b/cloud_scheduler/snippets/delete_job.py index 3f456c5b83a..527e83fe923 100644 --- a/cloud_scheduler/snippets/delete_job.py +++ b/cloud_scheduler/snippets/delete_job.py @@ -1,40 +1,68 @@ # Copyright 2023 Google LLC # -# Licensed under the Apache License, Version 2.0 (the 'License'); +# Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # -# http://www.apache.org/licenses/LICENSE-2.0 +# http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an 'AS IS' BASIS, +# distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # [START cloudscheduler_delete_job] +import sys +from google.api_core import exceptions from google.cloud import scheduler_v1 -def delete_scheduler_job(project_id: str, location_id: str, job_id: str) -> None: - """Delete a job via the Cloud Scheduler API. +def delete_job(project_id: str, location_id: str, job_id: str) -> None: + """Deletes a job so that it no longer runs. Args: - project_id: The Google Cloud project id. - location_id: The location for the job to delete. - job_id: The id of the job to delete. + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region of the job (for example, 'us-central1'). + job_id: The ID of the job to delete (for example, 'example-job-id'). """ - - # Create a client. client = scheduler_v1.CloudSchedulerClient() + name = f"projects/{project_id}/locations/{location_id}/jobs/{job_id}" + request = scheduler_v1.DeleteJobRequest(name=name) - # Use the client to send the job deletion request. - client.delete_job( - scheduler_v1.DeleteJobRequest( - name=client.job_path(project_id, location_id, job_id) + try: + client.delete_job(request=request) + except exceptions.NotFound as err: + print( + f"error: job '{job_id}' not found in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, ) - ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to delete job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {name}") + print("State: DELETED") # [END cloudscheduler_delete_job] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + delete_job( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + ) diff --git a/cloud_scheduler/snippets/get_job.py b/cloud_scheduler/snippets/get_job.py new file mode 100644 index 00000000000..6ff04ea95d2 --- /dev/null +++ b/cloud_scheduler/snippets/get_job.py @@ -0,0 +1,72 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_get_job] +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 + + +def get_job(project_id: str, location_id: str, job_id: str) -> None: + """Gets a job, including its state and the result of its last attempt. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region of the job (for example, 'us-central1'). + job_id: The ID of the job (for example, 'example-job-id'). + """ + client = scheduler_v1.CloudSchedulerClient() + name = f"projects/{project_id}/locations/{location_id}/jobs/{job_id}" + request = scheduler_v1.GetJobRequest(name=name) + + try: + response = client.get_job(request=request) + except exceptions.NotFound as err: + print( + f"error: job '{job_id}' not found in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to get job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + # The last attempt time and status code are empty until the job runs. + # A status code of 0 means the last attempt succeeded. + print(f"Last Attempt Time: {response.last_attempt_time}") + print(f"State: {response.state.name}") + print(f"Status Code: {response.status.code}") + + +# [END cloudscheduler_get_job] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + get_job( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + ) diff --git a/cloud_scheduler/snippets/job_test.py b/cloud_scheduler/snippets/job_test.py deleted file mode 100644 index 92cf67ef75c..00000000000 --- a/cloud_scheduler/snippets/job_test.py +++ /dev/null @@ -1,69 +0,0 @@ -# Copyright 2019 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import os -import uuid - -from google.api_core import exceptions -from google.cloud import scheduler_v1 - -import pytest - -import create_job -import delete_job - -TEST_PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT") -TEST_LOCATION = os.getenv("LOCATION_ID", "us-central1") -TEST_JOB_ID = f"my-job-{uuid.uuid4().hex}" - - -def test_create_job() -> None: - client = scheduler_v1.CloudSchedulerClient() - - job = create_job.create_scheduler_job(TEST_PROJECT_ID, TEST_LOCATION, "my-service") - assert job.name.startswith( - client.common_location_path(TEST_PROJECT_ID, TEST_LOCATION) - ) - - client.delete_job(scheduler_v1.DeleteJobRequest(name=job.name)) - - -def test_delete_job() -> None: - client = scheduler_v1.CloudSchedulerClient() - - client.create_job( - scheduler_v1.CreateJobRequest( - parent=client.common_location_path(TEST_PROJECT_ID, TEST_LOCATION), - job=scheduler_v1.Job( - name=client.job_path(TEST_PROJECT_ID, TEST_LOCATION, TEST_JOB_ID), - app_engine_http_target=scheduler_v1.AppEngineHttpTarget( - app_engine_routing=scheduler_v1.AppEngineRouting( - service="service-id" - ), - relative_uri="/log_payload", - http_method=scheduler_v1.HttpMethod.POST, - body=b"Hello World", - ), - schedule="* * * * *", - time_zone="America/Los_Angeles", - ), - ) - ) - - # Deleting an existant job succeeds - delete_job.delete_scheduler_job(TEST_PROJECT_ID, TEST_LOCATION, TEST_JOB_ID) - - # Deleting a non-existant job fails with a not found error - with pytest.raises(exceptions.NotFound): - delete_job.delete_scheduler_job(TEST_PROJECT_ID, TEST_LOCATION, TEST_JOB_ID) diff --git a/cloud_scheduler/snippets/list_jobs.py b/cloud_scheduler/snippets/list_jobs.py new file mode 100644 index 00000000000..20c67bb46d7 --- /dev/null +++ b/cloud_scheduler/snippets/list_jobs.py @@ -0,0 +1,67 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_list_jobs] +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 + + +def list_jobs(project_id: str, location_id: str) -> None: + """Lists the jobs in a location, with each job's schedule and state. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region to list jobs in (for example, 'us-central1'). + """ + client = scheduler_v1.CloudSchedulerClient() + parent = f"projects/{project_id}/locations/{location_id}" + request = scheduler_v1.ListJobsRequest(parent=parent) + + try: + # The client fetches more pages as the loop needs them. + for job in client.list_jobs(request=request): + print(f"Resource ID: {job.name}") + print(f"Schedule: {job.schedule}") + print(f"State: {job.state.name}") + except exceptions.InvalidArgument as err: + print( + f"error: invalid location '{location_id}' for project " + f"'{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to list jobs in location '{location_id}' for project " + f"'{project_id}': {err.message}", + file=sys.stderr, + ) + + +# [END cloudscheduler_list_jobs] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + list_jobs( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + ) diff --git a/cloud_scheduler/snippets/main.py b/cloud_scheduler/snippets/main.py deleted file mode 100644 index c61abb28842..00000000000 --- a/cloud_scheduler/snippets/main.py +++ /dev/null @@ -1,44 +0,0 @@ -# Copyright 2019 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""App Engine app to serve as an endpoint for Cloud Scheduler samples.""" - -# [START cloudscheduler_app] -from flask import Flask, request - -app = Flask(__name__) - - -# Define relative URI for job endpoint -@app.route("/log_payload", methods=["POST"]) -def example_task_handler() -> str: - """Log the job payload.""" - payload = request.get_data(as_text=True) or "(empty payload)" - print(f"Received job with payload: {payload}") - return f"Printed job payload: {payload}" - - -# [END cloudscheduler_app] - - -@app.route("/") -def hello() -> str: - """Basic index to verify app is serving.""" - return "Hello World!" - - -if __name__ == "__main__": - # This is used when running locally. Gunicorn is used to run the - # application on Google App Engine. See entrypoint in app.yaml. - app.run(host="127.0.0.1", port=8080, debug=True) diff --git a/cloud_scheduler/snippets/main_test.py b/cloud_scheduler/snippets/main_test.py deleted file mode 100644 index 9210be1a3c2..00000000000 --- a/cloud_scheduler/snippets/main_test.py +++ /dev/null @@ -1,42 +0,0 @@ -# Copyright 2019 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from flask.testing import FlaskClient -import pytest - - -@pytest.fixture -def app() -> FlaskClient: - import main - - main.app.testing = True - return main.app.test_client() - - -def test_index(app: FlaskClient) -> None: - r = app.get("/") - assert r.status_code == 200 - - -def test_log_payload(app: FlaskClient) -> None: - payload = "test_payload" - response = app.post("/log_payload", data=payload) - assert response.status_code == 200 - assert payload in response.text - - -def test_empty_payload(app: FlaskClient) -> None: - response = app.post("/log_payload") - assert response.status_code == 200 - assert "empty payload" in response.text diff --git a/cloud_scheduler/snippets/noxfile_config.py b/cloud_scheduler/snippets/noxfile_config.py index 0973c8621c7..d63baa25bfa 100644 --- a/cloud_scheduler/snippets/noxfile_config.py +++ b/cloud_scheduler/snippets/noxfile_config.py @@ -22,7 +22,7 @@ TEST_CONFIG_OVERRIDE = { # You can opt out from the test for specific Python versions. - "ignored_versions": ["3.8", "3.9", "3.11", "3.12", "3.13"], + "ignored_versions": ["2.7", "3.7", "3.8", "3.9", "3.10", "3.11", "3.12"], # Old samples are opted out of enforcing Python type hints # All new samples should feature them "enforce_type_hints": True, diff --git a/cloud_scheduler/snippets/pause_job.py b/cloud_scheduler/snippets/pause_job.py new file mode 100644 index 00000000000..ac502411076 --- /dev/null +++ b/cloud_scheduler/snippets/pause_job.py @@ -0,0 +1,74 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_pause_job] +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 + + +def pause_job(project_id: str, location_id: str, job_id: str) -> None: + """Pauses a job so that it stops running on its schedule. + + A paused job keeps its configuration and doesn't run until you resume it. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region of the job (for example, 'us-central1'). + job_id: The ID of the job (for example, 'example-job-id'). + """ + client = scheduler_v1.CloudSchedulerClient() + name = f"projects/{project_id}/locations/{location_id}/jobs/{job_id}" + request = scheduler_v1.PauseJobRequest(name=name) + + # Cloud Scheduler applies one change to a job at a time. If another change + # is still in progress, this call fails with ABORTED; retry after a few + # seconds. + try: + response = client.pause_job(request=request) + except exceptions.NotFound as err: + print( + f"error: job '{job_id}' not found in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to pause job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + print(f"Schedule: {response.schedule}") + print(f"State: {response.state.name}") + + +# [END cloudscheduler_pause_job] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + pause_job( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + ) diff --git a/cloud_scheduler/snippets/requirements.txt b/cloud_scheduler/snippets/requirements.txt index 0d8b19c3f04..3a4ae1b162c 100644 --- a/cloud_scheduler/snippets/requirements.txt +++ b/cloud_scheduler/snippets/requirements.txt @@ -1,4 +1 @@ -Flask==3.1.3; python_version >= '3.9' -gunicorn==23.0.0 -google-cloud-scheduler==2.14.1 -Werkzeug==3.1.9; python_version >= '3.10' +google-cloud-scheduler==2.21.0 diff --git a/cloud_scheduler/snippets/resume_job.py b/cloud_scheduler/snippets/resume_job.py new file mode 100644 index 00000000000..3675839268e --- /dev/null +++ b/cloud_scheduler/snippets/resume_job.py @@ -0,0 +1,74 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_resume_job] +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 + + +def resume_job(project_id: str, location_id: str, job_id: str) -> None: + """Resumes a paused job so that it runs on its schedule again. + + After it resumes, the job runs again at its next scheduled time. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region of the job (for example, 'us-central1'). + job_id: The ID of the job (for example, 'example-job-id'). + """ + client = scheduler_v1.CloudSchedulerClient() + name = f"projects/{project_id}/locations/{location_id}/jobs/{job_id}" + request = scheduler_v1.ResumeJobRequest(name=name) + + # Cloud Scheduler applies one change to a job at a time. If another change + # is still in progress, this call fails with ABORTED; retry after a few + # seconds. + try: + response = client.resume_job(request=request) + except exceptions.NotFound as err: + print( + f"error: job '{job_id}' not found in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to resume job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + print(f"Schedule: {response.schedule}") + print(f"State: {response.state.name}") + + +# [END cloudscheduler_resume_job] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + resume_job( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + ) diff --git a/cloud_scheduler/snippets/run_job.py b/cloud_scheduler/snippets/run_job.py new file mode 100644 index 00000000000..5e54942941c --- /dev/null +++ b/cloud_scheduler/snippets/run_job.py @@ -0,0 +1,72 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_run_job] +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 + + +def run_job(project_id: str, location_id: str, job_id: str) -> None: + """Runs a job immediately, outside its schedule. + + Running a job doesn't change its schedule. Use it to test a job's + target and authentication without waiting for the next scheduled run. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region of the job (for example, 'us-central1'). + job_id: The ID of the job (for example, 'example-job-id'). + """ + client = scheduler_v1.CloudSchedulerClient() + name = f"projects/{project_id}/locations/{location_id}/jobs/{job_id}" + request = scheduler_v1.RunJobRequest(name=name) + + try: + response = client.run_job(request=request) + except exceptions.NotFound as err: + print( + f"error: job '{job_id}' not found in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to run job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + print(f"Schedule: {response.schedule}") + print(f"State: {response.state.name}") + + +# [END cloudscheduler_run_job] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + run_job( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + ) diff --git a/cloud_scheduler/snippets/tests/__init__.py b/cloud_scheduler/snippets/tests/__init__.py new file mode 100644 index 00000000000..58d482ea386 --- /dev/null +++ b/cloud_scheduler/snippets/tests/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/cloud_scheduler/snippets/tests/create_appengine_job_test.py b/cloud_scheduler/snippets/tests/create_appengine_job_test.py new file mode 100644 index 00000000000..af09c6747e9 --- /dev/null +++ b/cloud_scheduler/snippets/tests/create_appengine_job_test.py @@ -0,0 +1,91 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import create_appengine_job + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_create_appengine_job_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.return_value = _job() + + create_appengine_job.create_appengine_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + request = client.create_job.call_args.kwargs["request"] + assert request.parent == PARENT + target = request.job.app_engine_http_target + assert target.app_engine_routing.service == "default" + assert target.relative_uri == "/example-handler" + assert request.job.time_zone == "UTC" + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "State: ENABLED" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_create_appengine_job_already_exists( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.AlreadyExists("test error detail") + + create_appengine_job.create_appengine_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' already exists") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_create_appengine_job_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.PermissionDenied("denied") + + create_appengine_job.create_appengine_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to create job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/create_http_job_oauth_test.py b/cloud_scheduler/snippets/tests/create_http_job_oauth_test.py new file mode 100644 index 00000000000..fa980f82e48 --- /dev/null +++ b/cloud_scheduler/snippets/tests/create_http_job_oauth_test.py @@ -0,0 +1,105 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import create_http_job_oauth + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" +WORKFLOW_TEMPLATE_ID = "test-template" +SERVICE_ACCOUNT_EMAIL = "invoker@test-project.iam.gserviceaccount.com" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_create_http_job_oauth_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.return_value = _job( + http_target=scheduler_v1.HttpTarget(uri="https://dataproc.googleapis.com/v1/x") + ) + + create_http_job_oauth.create_http_job_oauth( + PROJECT_ID, LOCATION_ID, JOB_ID, WORKFLOW_TEMPLATE_ID, SERVICE_ACCOUNT_EMAIL + ) + + request = client.create_job.call_args.kwargs["request"] + target = request.job.http_target + assert target.uri == ( + "https://dataproc.googleapis.com/v1/projects/test-project/regions/" + "us-central1/workflowTemplates/test-template:instantiate" + ) + assert target.oauth_token.service_account_email == SERVICE_ACCOUNT_EMAIL + assert target.oauth_token.scope == "https://www.googleapis.com/auth/cloud-platform" + assert target.headers["Content-Type"] == "application/json" + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "State: ENABLED" in out + assert "Target URI: https://dataproc.googleapis.com" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_create_http_job_oauth_already_exists( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.AlreadyExists("test error detail") + + create_http_job_oauth.create_http_job_oauth( + PROJECT_ID, LOCATION_ID, JOB_ID, WORKFLOW_TEMPLATE_ID, SERVICE_ACCOUNT_EMAIL + ) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' already exists") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_create_http_job_oauth_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.PermissionDenied("denied") + + create_http_job_oauth.create_http_job_oauth( + PROJECT_ID, LOCATION_ID, JOB_ID, WORKFLOW_TEMPLATE_ID, SERVICE_ACCOUNT_EMAIL + ) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to create job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/create_job_test.py b/cloud_scheduler/snippets/tests/create_job_test.py new file mode 100644 index 00000000000..6d7e8226b2e --- /dev/null +++ b/cloud_scheduler/snippets/tests/create_job_test.py @@ -0,0 +1,104 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import create_job + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" +SERVICE_URL = "https://test-service.a.run.app/" +SERVICE_ACCOUNT_EMAIL = "invoker@test-project.iam.gserviceaccount.com" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_create_job_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.return_value = _job() + + create_job.create_job( + PROJECT_ID, LOCATION_ID, JOB_ID, SERVICE_URL, SERVICE_ACCOUNT_EMAIL + ) + + request = client.create_job.call_args.kwargs["request"] + assert request.parent == PARENT + assert request.job.name == JOB_NAME + assert request.job.http_target.uri == SERVICE_URL + assert request.job.http_target.http_method == scheduler_v1.HttpMethod.POST + oidc_token = request.job.http_target.oidc_token + assert oidc_token.service_account_email == SERVICE_ACCOUNT_EMAIL + assert oidc_token.audience == SERVICE_URL + assert request.job.time_zone == "UTC" + assert request.job.attempt_deadline.total_seconds() == 300 + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "State: ENABLED" in out + assert "Time Zone: UTC" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_create_job_already_exists( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.AlreadyExists("test error detail") + + create_job.create_job( + PROJECT_ID, LOCATION_ID, JOB_ID, SERVICE_URL, SERVICE_ACCOUNT_EMAIL + ) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' already exists") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_create_job_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.PermissionDenied("denied") + + create_job.create_job( + PROJECT_ID, LOCATION_ID, JOB_ID, SERVICE_URL, SERVICE_ACCOUNT_EMAIL + ) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to create job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/create_job_with_retry_test.py b/cloud_scheduler/snippets/tests/create_job_with_retry_test.py new file mode 100644 index 00000000000..300255e1306 --- /dev/null +++ b/cloud_scheduler/snippets/tests/create_job_with_retry_test.py @@ -0,0 +1,103 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import create_job_with_retry + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" +SERVICE_URL = "https://test-service.a.run.app/" +SERVICE_ACCOUNT_EMAIL = "invoker@test-project.iam.gserviceaccount.com" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_create_job_with_retry_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.return_value = _job( + retry_config=scheduler_v1.RetryConfig(retry_count=3) + ) + + create_job_with_retry.create_job_with_retry( + PROJECT_ID, LOCATION_ID, JOB_ID, SERVICE_URL, SERVICE_ACCOUNT_EMAIL + ) + + request = client.create_job.call_args.kwargs["request"] + retry_config = request.job.retry_config + assert retry_config.retry_count == 3 + assert retry_config.max_retry_duration.total_seconds() == 3600 + assert retry_config.min_backoff_duration.total_seconds() == 10 + assert retry_config.max_backoff_duration.total_seconds() == 300 + assert retry_config.max_doublings == 3 + assert request.job.http_target.oidc_token.audience == SERVICE_URL + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "Retry Count: 3" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_create_job_with_retry_already_exists( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.AlreadyExists("test error detail") + + create_job_with_retry.create_job_with_retry( + PROJECT_ID, LOCATION_ID, JOB_ID, SERVICE_URL, SERVICE_ACCOUNT_EMAIL + ) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' already exists") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_create_job_with_retry_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.PermissionDenied("denied") + + create_job_with_retry.create_job_with_retry( + PROJECT_ID, LOCATION_ID, JOB_ID, SERVICE_URL, SERVICE_ACCOUNT_EMAIL + ) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to create job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/create_pubsub_job_test.py b/cloud_scheduler/snippets/tests/create_pubsub_job_test.py new file mode 100644 index 00000000000..52f6d028342 --- /dev/null +++ b/cloud_scheduler/snippets/tests/create_pubsub_job_test.py @@ -0,0 +1,96 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import create_pubsub_job + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" +TOPIC_ID = "test-topic" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_create_pubsub_job_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.return_value = _job( + pubsub_target=scheduler_v1.PubsubTarget( + topic_name="projects/test-project/topics/test-topic" + ) + ) + + create_pubsub_job.create_pubsub_job(PROJECT_ID, LOCATION_ID, JOB_ID, TOPIC_ID) + + request = client.create_job.call_args.kwargs["request"] + target = request.job.pubsub_target + assert target.topic_name == "projects/test-project/topics/test-topic" + assert target.data == b"example-payload" + assert target.attributes["example-key"] == "example-value" + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "State: ENABLED" in out + assert "Topic: projects/test-project/topics/test-topic" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_create_pubsub_job_already_exists( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.AlreadyExists("test error detail") + + create_pubsub_job.create_pubsub_job(PROJECT_ID, LOCATION_ID, JOB_ID, TOPIC_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' already exists") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_create_pubsub_job_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.create_job.side_effect = exceptions.PermissionDenied("denied") + + create_pubsub_job.create_pubsub_job(PROJECT_ID, LOCATION_ID, JOB_ID, TOPIC_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to create job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/delete_job_test.py b/cloud_scheduler/snippets/tests/delete_job_test.py new file mode 100644 index 00000000000..e4ae4ba1ac6 --- /dev/null +++ b/cloud_scheduler/snippets/tests/delete_job_test.py @@ -0,0 +1,87 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import delete_job + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_delete_job_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.delete_job.return_value = None + + delete_job.delete_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + request = client.delete_job.call_args.kwargs["request"] + assert request.name == JOB_NAME + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "State: DELETED" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_delete_job_not_found( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.delete_job.side_effect = exceptions.NotFound("test error detail") + + delete_job.delete_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' not found") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_delete_job_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.delete_job.side_effect = exceptions.PermissionDenied("denied") + + delete_job.delete_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to delete job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/get_job_test.py b/cloud_scheduler/snippets/tests/get_job_test.py new file mode 100644 index 00000000000..a4bf4ce5e7e --- /dev/null +++ b/cloud_scheduler/snippets/tests/get_job_test.py @@ -0,0 +1,88 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import get_job + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_get_job_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.get_job.return_value = _job() + + get_job.get_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + request = client.get_job.call_args.kwargs["request"] + assert request.name == JOB_NAME + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "State: ENABLED" in out + assert "Status Code: 0" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_get_job_not_found( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.get_job.side_effect = exceptions.NotFound("test error detail") + + get_job.get_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' not found") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_get_job_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.get_job.side_effect = exceptions.PermissionDenied("denied") + + get_job.get_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to get job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/integration_test.py b/cloud_scheduler/snippets/tests/integration_test.py new file mode 100644 index 00000000000..6dfaf5fd875 --- /dev/null +++ b/cloud_scheduler/snippets/tests/integration_test.py @@ -0,0 +1,255 @@ +# Copyright 2019 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Integration tests that run the Cloud Scheduler job samples live. + +These tests are skipped unless SERVICE_URL is set. + +The Terraform fixture in .maintenance/fixtures/cloudscheduler_receiver +provisions the resources below; export its outputs as these variables. + +Required environment variables: + GOOGLE_CLOUD_PROJECT: Project that runs the tests. It must have an + App Engine app in GOOGLE_CLOUD_LOCATION's region. + SERVICE_URL: URL of the deployed cloud_scheduler/receiver Cloud Run function. + +Optional environment variables: + GOOGLE_CLOUD_LOCATION: Region for the jobs (default 'us-central1'). + SERVICE_ACCOUNT_EMAIL: Service account with the Cloud Run Invoker role on + the receiver (default 'scheduler-invoker@PROJECT.iam.gserviceaccount.com'). + TOPIC_ID: Existing Pub/Sub topic (default 'scheduler-sample-topic'). +""" + +import os +import time +from typing import Callable, Iterator +import uuid + +from google.api_core import exceptions +from google.api_core import retry +from google.cloud import scheduler_v1 +import pytest + +from .. import ( + create_appengine_job, + create_http_job_oauth, + create_job, + create_job_with_retry, + create_pubsub_job, + delete_job, + get_job, + list_jobs, + pause_job, + resume_job, + run_job, + update_job, +) + +PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "") +LOCATION_ID = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1") +SERVICE_URL = os.environ.get("SERVICE_URL", "") +SERVICE_ACCOUNT_EMAIL = os.environ.get( + "SERVICE_ACCOUNT_EMAIL", + f"scheduler-invoker@{PROJECT_ID}.iam.gserviceaccount.com", +) +TOPIC_ID = os.environ.get("TOPIC_ID", "scheduler-sample-topic") + +pytestmark = pytest.mark.skipif( + not (PROJECT_ID and SERVICE_URL), + reason="Set GOOGLE_CLOUD_PROJECT and SERVICE_URL to run integration tests.", +) + +# Cloud Scheduler rejects a change to a job while a previous change is still +# being applied, with ABORTED: "sync mutate calls cannot be queued". +retry_aborted = retry.Retry( + predicate=retry.if_exception_type(exceptions.Aborted), timeout=120 +) + + +def run_sample( + capsys: pytest.CaptureFixture[str], sample: Callable[..., None], *args: str +) -> str: + """Runs a sample, retrying while the job has a change in progress.""" + for _ in range(10): + sample(*args) + out, err = capsys.readouterr() + if "cannot be queued" not in err: + assert "error:" not in err, err + return out + time.sleep(5) + pytest.fail(f"{sample.__name__} kept failing with ABORTED") + + +@pytest.fixture(scope="module") +def client() -> scheduler_v1.CloudSchedulerClient: + return scheduler_v1.CloudSchedulerClient() + + +def job_name(job_id: str) -> str: + return f"projects/{PROJECT_ID}/locations/{LOCATION_ID}/jobs/{job_id}" + + +@pytest.fixture +def job_id(client: scheduler_v1.CloudSchedulerClient) -> Iterator[str]: + """Provides an unused job ID and deletes the job after the test.""" + job_id = f"test-job-{uuid.uuid4().hex[:12]}" + yield job_id + try: + client.delete_job(name=job_name(job_id), retry=retry_aborted) + except exceptions.NotFound: + pass + + +@pytest.fixture +def existing_job_id(client: scheduler_v1.CloudSchedulerClient, job_id: str) -> str: + """Creates an HTTP job that calls the receiver with an OIDC token.""" + assert SERVICE_URL, "Set SERVICE_URL to a deployed receiver service." + client.create_job( + parent=f"projects/{PROJECT_ID}/locations/{LOCATION_ID}", + job=scheduler_v1.Job( + name=job_name(job_id), + http_target=scheduler_v1.HttpTarget( + uri=SERVICE_URL, + http_method=scheduler_v1.HttpMethod.POST, + body=b"test-payload", + oidc_token=scheduler_v1.OidcToken( + service_account_email=SERVICE_ACCOUNT_EMAIL, + audience=SERVICE_URL, + ), + ), + schedule="0 9 * * 1", + time_zone="UTC", + ), + retry=retry_aborted, + ) + return job_id + + +def test_create_job(capsys: pytest.CaptureFixture[str], job_id: str) -> None: + assert SERVICE_URL, "Set SERVICE_URL to a deployed receiver service." + create_job.create_job( + PROJECT_ID, LOCATION_ID, job_id, SERVICE_URL, SERVICE_ACCOUNT_EMAIL + ) + + out, err = capsys.readouterr() + assert f"Resource ID: {job_name(job_id)}" in out + assert "State: ENABLED" in out + assert "Time Zone: UTC" in out + assert "error:" not in err + + +def test_create_job_already_exists( + capsys: pytest.CaptureFixture[str], existing_job_id: str +) -> None: + create_job.create_job( + PROJECT_ID, LOCATION_ID, existing_job_id, SERVICE_URL, SERVICE_ACCOUNT_EMAIL + ) + + out, err = capsys.readouterr() + assert out == "" + assert f"error: job '{existing_job_id}' already exists" in err + + +def test_create_appengine_job(capsys: pytest.CaptureFixture[str], job_id: str) -> None: + create_appengine_job.create_appengine_job(PROJECT_ID, LOCATION_ID, job_id) + + out, err = capsys.readouterr() + assert f"Resource ID: {job_name(job_id)}" in out + assert "State: ENABLED" in out + assert "error:" not in err + + +def test_create_http_job_oauth(capsys: pytest.CaptureFixture[str], job_id: str) -> None: + create_http_job_oauth.create_http_job_oauth( + PROJECT_ID, + LOCATION_ID, + job_id, + "example-workflow-template", + SERVICE_ACCOUNT_EMAIL, + ) + + out, err = capsys.readouterr() + assert f"Resource ID: {job_name(job_id)}" in out + assert "dataproc.googleapis.com" in out + assert "example-workflow-template:instantiate" in out + assert "error:" not in err + + +def test_create_pubsub_job(capsys: pytest.CaptureFixture[str], job_id: str) -> None: + create_pubsub_job.create_pubsub_job(PROJECT_ID, LOCATION_ID, job_id, TOPIC_ID) + + out, err = capsys.readouterr() + assert f"Resource ID: {job_name(job_id)}" in out + assert f"Topic: projects/{PROJECT_ID}/topics/{TOPIC_ID}" in out + assert "error:" not in err + + +def test_create_job_with_retry(capsys: pytest.CaptureFixture[str], job_id: str) -> None: + assert SERVICE_URL, "Set SERVICE_URL to a deployed receiver service." + create_job_with_retry.create_job_with_retry( + PROJECT_ID, LOCATION_ID, job_id, SERVICE_URL, SERVICE_ACCOUNT_EMAIL + ) + + out, err = capsys.readouterr() + assert f"Resource ID: {job_name(job_id)}" in out + assert "Retry Count: 3" in out + assert "Max Retry Duration: 1:00:00" in out + assert "error:" not in err + + +def test_job_lifecycle( + capsys: pytest.CaptureFixture[str], existing_job_id: str +) -> None: + args = (PROJECT_ID, LOCATION_ID, existing_job_id) + + assert "State: PAUSED" in run_sample(capsys, pause_job.pause_job, *args) + assert "State: PAUSED" in run_sample(capsys, get_job.get_job, *args) + assert "State: ENABLED" in run_sample(capsys, resume_job.resume_job, *args) + out = run_sample(capsys, update_job.update_job, *args) + assert "Schedule: 0 10 * * 1-5" in out + out = run_sample(capsys, list_jobs.list_jobs, PROJECT_ID, LOCATION_ID) + assert f"Resource ID: {job_name(existing_job_id)}" in out + assert "State: DELETED" in run_sample(capsys, delete_job.delete_job, *args) + + +def test_run_job_reaches_receiver( + capsys: pytest.CaptureFixture[str], + client: scheduler_v1.CloudSchedulerClient, + existing_job_id: str, +) -> None: + run_job.run_job(PROJECT_ID, LOCATION_ID, existing_job_id) + out, err = capsys.readouterr() + assert f"Resource ID: {job_name(existing_job_id)}" in out + assert "error:" not in err + + # The run is asynchronous. Wait for the receiver to accept the request, + # which confirms that the Cloud Run function accepted the job's OIDC token. + deadline = time.monotonic() + 300 + job = client.get_job(name=job_name(existing_job_id)) + while time.monotonic() < deadline and not job.last_attempt_time: + time.sleep(5) + job = client.get_job(name=job_name(existing_job_id)) + assert job.last_attempt_time, "The job didn't run within 5 minutes." + assert job.status.code == 0, job.status.message + + out = run_sample(capsys, get_job.get_job, PROJECT_ID, LOCATION_ID, existing_job_id) + assert "Status Code: 0" in out + + +def test_job_not_found(capsys: pytest.CaptureFixture[str], job_id: str) -> None: + for sample in (get_job.get_job, pause_job.pause_job, delete_job.delete_job): + sample(PROJECT_ID, LOCATION_ID, job_id) + out, err = capsys.readouterr() + assert out == "" + assert f"error: job '{job_id}' not found" in err diff --git a/cloud_scheduler/snippets/tests/list_jobs_test.py b/cloud_scheduler/snippets/tests/list_jobs_test.py new file mode 100644 index 00000000000..8619a93d0cd --- /dev/null +++ b/cloud_scheduler/snippets/tests/list_jobs_test.py @@ -0,0 +1,90 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import list_jobs + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_list_jobs_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.list_jobs.return_value = [ + _job(), + scheduler_v1.Job(name=f"{PARENT}/jobs/other-job"), + ] + + list_jobs.list_jobs(PROJECT_ID, LOCATION_ID) + + request = client.list_jobs.call_args.kwargs["request"] + assert request.parent == PARENT + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert f"Resource ID: {PARENT}/jobs/other-job" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_list_jobs_invalid_location( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.list_jobs.side_effect = exceptions.InvalidArgument("test error detail") + + list_jobs.list_jobs(PROJECT_ID, LOCATION_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: invalid location 'us-central1'") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_list_jobs_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.list_jobs.side_effect = exceptions.PermissionDenied("denied") + + list_jobs.list_jobs(PROJECT_ID, LOCATION_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to list jobs") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/pause_job_test.py b/cloud_scheduler/snippets/tests/pause_job_test.py new file mode 100644 index 00000000000..fbfad3f80e9 --- /dev/null +++ b/cloud_scheduler/snippets/tests/pause_job_test.py @@ -0,0 +1,87 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import pause_job + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_pause_job_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.pause_job.return_value = _job(state=scheduler_v1.Job.State.PAUSED) + + pause_job.pause_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + request = client.pause_job.call_args.kwargs["request"] + assert request.name == JOB_NAME + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "State: PAUSED" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_pause_job_not_found( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.pause_job.side_effect = exceptions.NotFound("test error detail") + + pause_job.pause_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' not found") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_pause_job_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.pause_job.side_effect = exceptions.PermissionDenied("denied") + + pause_job.pause_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to pause job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/resume_job_test.py b/cloud_scheduler/snippets/tests/resume_job_test.py new file mode 100644 index 00000000000..8330c2c1900 --- /dev/null +++ b/cloud_scheduler/snippets/tests/resume_job_test.py @@ -0,0 +1,87 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import resume_job + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_resume_job_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.resume_job.return_value = _job() + + resume_job.resume_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + request = client.resume_job.call_args.kwargs["request"] + assert request.name == JOB_NAME + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "State: ENABLED" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_resume_job_not_found( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.resume_job.side_effect = exceptions.NotFound("test error detail") + + resume_job.resume_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' not found") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_resume_job_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.resume_job.side_effect = exceptions.PermissionDenied("denied") + + resume_job.resume_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to resume job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/run_job_test.py b/cloud_scheduler/snippets/tests/run_job_test.py new file mode 100644 index 00000000000..ef7fada606d --- /dev/null +++ b/cloud_scheduler/snippets/tests/run_job_test.py @@ -0,0 +1,87 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import run_job + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_run_job_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.run_job.return_value = _job() + + run_job.run_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + request = client.run_job.call_args.kwargs["request"] + assert request.name == JOB_NAME + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "Schedule: 0 9 * * 1" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_run_job_not_found( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.run_job.side_effect = exceptions.NotFound("test error detail") + + run_job.run_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' not found") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_run_job_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.run_job.side_effect = exceptions.PermissionDenied("denied") + + run_job.run_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to run job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/tests/update_job_test.py b/cloud_scheduler/snippets/tests/update_job_test.py new file mode 100644 index 00000000000..30fec9a1be2 --- /dev/null +++ b/cloud_scheduler/snippets/tests/update_job_test.py @@ -0,0 +1,89 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +import pytest + +from .. import update_job + +PROJECT_ID = "test-project" +LOCATION_ID = "us-central1" +JOB_ID = "test-job" +PARENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}" +JOB_NAME = f"{PARENT}/jobs/{JOB_ID}" +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" + + +def _job(**fields: object) -> scheduler_v1.Job: + """Returns a job response, with defaults that fields can override.""" + job_fields: dict[str, object] = { + "name": JOB_NAME, + "schedule": "0 9 * * 1", + "time_zone": "UTC", + "state": scheduler_v1.Job.State.ENABLED, + } + job_fields.update(fields) + return scheduler_v1.Job(**job_fields) + + +@mock.patch(CLIENT) +def test_update_job_success( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.update_job.return_value = _job(schedule="0 10 * * 1-5") + + update_job.update_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + request = client.update_job.call_args.kwargs["request"] + assert request.job.name == JOB_NAME + assert request.job.schedule == "0 10 * * 1-5" + assert list(request.update_mask.paths) == ["schedule", "time_zone"] + out, err = capsys.readouterr() + assert f"Resource ID: {JOB_NAME}" in out + assert "Schedule: 0 10 * * 1-5" in out + assert err == "" + + +@mock.patch(CLIENT) +def test_update_job_not_found( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.update_job.side_effect = exceptions.NotFound("test error detail") + + update_job.update_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: job 'test-job' not found") + assert "test error detail" in err + + +@mock.patch(CLIENT) +def test_update_job_api_error( + mock_client_class: mock.MagicMock, capsys: pytest.CaptureFixture[str] +) -> None: + client = mock_client_class.return_value + client.update_job.side_effect = exceptions.PermissionDenied("denied") + + update_job.update_job(PROJECT_ID, LOCATION_ID, JOB_ID) + + out, err = capsys.readouterr() + assert out == "" + assert err.startswith("error: failed to update job") + assert "denied" in err diff --git a/cloud_scheduler/snippets/update_job.py b/cloud_scheduler/snippets/update_job.py new file mode 100644 index 00000000000..15ed9678c2b --- /dev/null +++ b/cloud_scheduler/snippets/update_job.py @@ -0,0 +1,80 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# [START cloudscheduler_update_job] +import sys + +from google.api_core import exceptions +from google.cloud import scheduler_v1 +from google.protobuf import field_mask_pb2 + + +def update_job(project_id: str, location_id: str, job_id: str) -> None: + """Changes the schedule of an existing job. + + Args: + project_id: The Google Cloud project ID (for example, 'example-project-id'). + location_id: The region of the job (for example, 'us-central1'). + job_id: The ID of the job to update (for example, 'example-job-id'). + """ + client = scheduler_v1.CloudSchedulerClient() + name = f"projects/{project_id}/locations/{location_id}/jobs/{job_id}" + + # Runs every weekday at 10:00 UTC. + job = scheduler_v1.Job(name=name, schedule="0 10 * * 1-5", time_zone="UTC") + # Only the fields in the update mask change; the job keeps its other + # settings, such as its target. + update_mask = field_mask_pb2.FieldMask(paths=["schedule", "time_zone"]) + request = scheduler_v1.UpdateJobRequest(job=job, update_mask=update_mask) + + # Cloud Scheduler applies one change to a job at a time. If another change + # is still in progress, this call fails with ABORTED; retry after a few + # seconds. + try: + response = client.update_job(request=request) + except exceptions.NotFound as err: + print( + f"error: job '{job_id}' not found in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + except exceptions.GoogleAPICallError as err: + print( + f"error: failed to update job '{job_id}' in location '{location_id}' " + f"for project '{project_id}': {err.message}", + file=sys.stderr, + ) + else: + print(f"Resource ID: {response.name}") + print(f"Schedule: {response.schedule}") + print(f"State: {response.state.name}") + print(f"Time Zone: {response.time_zone}") + + +# [END cloudscheduler_update_job] + + +if __name__ == "__main__": + import os + + def arg(index: int, env_var: str, default: str) -> str: + """Reads an environment variable, then a CLI argument, then a default.""" + cli_value = sys.argv[index] if len(sys.argv) > index else None + return os.environ.get(env_var) or cli_value or default + + update_job( + project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), + location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + ) From 9f3cb141ed6f59e56840a4ee7063a70175674fef Mon Sep 17 00:00:00 2001 From: Jennifer Davis Date: Fri, 6 Nov 2026 21:18:04 +0000 Subject: [PATCH 2/4] docs(cloud_scheduler): state authentication requirements positively Replace the double negatives about unauthenticated access in the create_job and receiver docstrings with positive phrasing: the Cloud Run service can require authentication, and the receiver function is deployed to require authentication. --- cloud_scheduler/receiver/main.py | 4 ++-- cloud_scheduler/snippets/create_job.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/cloud_scheduler/receiver/main.py b/cloud_scheduler/receiver/main.py index 982fdccc823..86f664989e2 100644 --- a/cloud_scheduler/receiver/main.py +++ b/cloud_scheduler/receiver/main.py @@ -34,8 +34,8 @@ def log(message: str, **fields: str) -> None: def handle_scheduled_run(request: flask.Request) -> tuple[str, int]: """Handles a request that a Cloud Scheduler job sends to this function. - Deploy the function without unauthenticated access. Cloud Run functions - then verifies the job's OIDC token before the request reaches this + Deploy the function to require authentication. Cloud Run functions then + verifies the job's OIDC token before the request reaches this function, and only service accounts with the Cloud Run Invoker role can call it. """ diff --git a/cloud_scheduler/snippets/create_job.py b/cloud_scheduler/snippets/create_job.py index bfde35bbcb1..aedb8668aa0 100644 --- a/cloud_scheduler/snippets/create_job.py +++ b/cloud_scheduler/snippets/create_job.py @@ -30,7 +30,7 @@ def create_job( """Creates a job that calls a Cloud Run service on a schedule. The job authenticates with an OIDC token for a service account, so the - Cloud Run service doesn't need to allow unauthenticated access. + Cloud Run service can require authentication. Args: project_id: The Google Cloud project ID (for example, 'example-project-id'). From 5c6d9318601ab4aa3912a9313ce1edef97c239bd Mon Sep 17 00:00:00 2001 From: Jennifer Davis Date: Fri, 6 Nov 2026 21:33:38 +0000 Subject: [PATCH 3/4] refactor(cloud_scheduler): use argparse for sample entrypoints Replace the custom argument helper in each sample's validation entrypoint with argparse named flags (for example, --project-id and --job-id). Each flag defaults to its environment variable, such as GOOGLE_CLOUD_PROJECT, and then to an example value, so the samples run without flags when the environment variables are set. Flags override environment variables, and --help lists every option with its environment variable. Add unit tests for the entrypoint that cover environment variable defaults, flag overrides, and example defaults. --- .../snippets/create_appengine_job.py | 31 ++++++-- .../snippets/create_http_job_oauth.py | 52 ++++++++++---- cloud_scheduler/snippets/create_job.py | 48 +++++++++---- .../snippets/create_job_with_retry.py | 50 +++++++++---- cloud_scheduler/snippets/create_pubsub_job.py | 38 +++++++--- cloud_scheduler/snippets/delete_job.py | 29 ++++++-- cloud_scheduler/snippets/get_job.py | 29 ++++++-- cloud_scheduler/snippets/list_jobs.py | 22 ++++-- cloud_scheduler/snippets/pause_job.py | 29 ++++++-- cloud_scheduler/snippets/resume_job.py | 29 ++++++-- cloud_scheduler/snippets/run_job.py | 29 ++++++-- .../snippets/tests/entrypoint_test.py | 72 +++++++++++++++++++ cloud_scheduler/snippets/update_job.py | 29 ++++++-- 13 files changed, 386 insertions(+), 101 deletions(-) create mode 100644 cloud_scheduler/snippets/tests/entrypoint_test.py diff --git a/cloud_scheduler/snippets/create_appengine_job.py b/cloud_scheduler/snippets/create_appengine_job.py index 22358c8cc2c..2040605313c 100644 --- a/cloud_scheduler/snippets/create_appengine_job.py +++ b/cloud_scheduler/snippets/create_appengine_job.py @@ -79,15 +79,32 @@ def create_appengine_job(project_id: str, location_id: str, job_id: str) -> None if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser( + description=create_appengine_job.__doc__.splitlines()[0] + ) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + args = parser.parse_args() create_appengine_job( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, ) diff --git a/cloud_scheduler/snippets/create_http_job_oauth.py b/cloud_scheduler/snippets/create_http_job_oauth.py index d1c218d36c6..7cef27d08f2 100644 --- a/cloud_scheduler/snippets/create_http_job_oauth.py +++ b/cloud_scheduler/snippets/create_http_job_oauth.py @@ -96,23 +96,47 @@ def create_http_job_oauth( if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default - - create_http_job_oauth( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), - workflow_template_id=arg( - 4, "WORKFLOW_TEMPLATE_ID", "example-workflow-template" - ), - service_account_email=arg( - 5, + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser( + description=create_http_job_oauth.__doc__.splitlines()[0] + ) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + parser.add_argument( + "--workflow-template-id", + default=os.environ.get("WORKFLOW_TEMPLATE_ID", "example-workflow-template"), + help="ID of the workflow template to start (default: $WORKFLOW_TEMPLATE_ID)", + ) + parser.add_argument( + "--service-account-email", + default=os.environ.get( "SERVICE_ACCOUNT_EMAIL", "example-invoker@example-project-id.iam.gserviceaccount.com", ), + help="Service account that the job uses to get tokens (default: $SERVICE_ACCOUNT_EMAIL)", + ) + args = parser.parse_args() + + create_http_job_oauth( + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, + workflow_template_id=args.workflow_template_id, + service_account_email=args.service_account_email, ) diff --git a/cloud_scheduler/snippets/create_job.py b/cloud_scheduler/snippets/create_job.py index aedb8668aa0..93b293dc6df 100644 --- a/cloud_scheduler/snippets/create_job.py +++ b/cloud_scheduler/snippets/create_job.py @@ -92,21 +92,45 @@ def create_job( if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default - - create_job( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), - service_url=arg(4, "SERVICE_URL", "https://example-service.a.run.app/"), - service_account_email=arg( - 5, + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser(description=create_job.__doc__.splitlines()[0]) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + parser.add_argument( + "--service-url", + default=os.environ.get("SERVICE_URL", "https://example-service.a.run.app/"), + help="HTTPS URL of the Cloud Run service to call (default: $SERVICE_URL)", + ) + parser.add_argument( + "--service-account-email", + default=os.environ.get( "SERVICE_ACCOUNT_EMAIL", "example-invoker@example-project-id.iam.gserviceaccount.com", ), + help="Service account that the job uses to get tokens (default: $SERVICE_ACCOUNT_EMAIL)", + ) + args = parser.parse_args() + + create_job( + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, + service_url=args.service_url, + service_account_email=args.service_account_email, ) diff --git a/cloud_scheduler/snippets/create_job_with_retry.py b/cloud_scheduler/snippets/create_job_with_retry.py index 59c02007a8b..3120cd07dae 100644 --- a/cloud_scheduler/snippets/create_job_with_retry.py +++ b/cloud_scheduler/snippets/create_job_with_retry.py @@ -100,21 +100,47 @@ def create_job_with_retry( if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default - - create_job_with_retry( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), - service_url=arg(4, "SERVICE_URL", "https://example-service.a.run.app/"), - service_account_email=arg( - 5, + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser( + description=create_job_with_retry.__doc__.splitlines()[0] + ) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + parser.add_argument( + "--service-url", + default=os.environ.get("SERVICE_URL", "https://example-service.a.run.app/"), + help="HTTPS URL of the Cloud Run service to call (default: $SERVICE_URL)", + ) + parser.add_argument( + "--service-account-email", + default=os.environ.get( "SERVICE_ACCOUNT_EMAIL", "example-invoker@example-project-id.iam.gserviceaccount.com", ), + help="Service account that the job uses to get tokens (default: $SERVICE_ACCOUNT_EMAIL)", + ) + args = parser.parse_args() + + create_job_with_retry( + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, + service_url=args.service_url, + service_account_email=args.service_account_email, ) diff --git a/cloud_scheduler/snippets/create_pubsub_job.py b/cloud_scheduler/snippets/create_pubsub_job.py index e1eb0191fb4..21a352f6901 100644 --- a/cloud_scheduler/snippets/create_pubsub_job.py +++ b/cloud_scheduler/snippets/create_pubsub_job.py @@ -76,16 +76,38 @@ def create_pubsub_job( if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser( + description=create_pubsub_job.__doc__.splitlines()[0] + ) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + parser.add_argument( + "--topic-id", + default=os.environ.get("TOPIC_ID", "example-topic"), + help="ID of an existing Pub/Sub topic (default: $TOPIC_ID)", + ) + args = parser.parse_args() create_pubsub_job( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), - topic_id=arg(4, "TOPIC_ID", "example-topic"), + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, + topic_id=args.topic_id, ) diff --git a/cloud_scheduler/snippets/delete_job.py b/cloud_scheduler/snippets/delete_job.py index 527e83fe923..5852679ff1f 100644 --- a/cloud_scheduler/snippets/delete_job.py +++ b/cloud_scheduler/snippets/delete_job.py @@ -54,15 +54,30 @@ def delete_job(project_id: str, location_id: str, job_id: str) -> None: if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser(description=delete_job.__doc__.splitlines()[0]) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + args = parser.parse_args() delete_job( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, ) diff --git a/cloud_scheduler/snippets/get_job.py b/cloud_scheduler/snippets/get_job.py index 6ff04ea95d2..cddbeead8b2 100644 --- a/cloud_scheduler/snippets/get_job.py +++ b/cloud_scheduler/snippets/get_job.py @@ -58,15 +58,30 @@ def get_job(project_id: str, location_id: str, job_id: str) -> None: if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser(description=get_job.__doc__.splitlines()[0]) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + args = parser.parse_args() get_job( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, ) diff --git a/cloud_scheduler/snippets/list_jobs.py b/cloud_scheduler/snippets/list_jobs.py index 20c67bb46d7..990bba591aa 100644 --- a/cloud_scheduler/snippets/list_jobs.py +++ b/cloud_scheduler/snippets/list_jobs.py @@ -54,14 +54,24 @@ def list_jobs(project_id: str, location_id: str) -> None: if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser(description=list_jobs.__doc__.splitlines()[0]) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + args = parser.parse_args() list_jobs( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), + project_id=args.project_id, + location_id=args.location_id, ) diff --git a/cloud_scheduler/snippets/pause_job.py b/cloud_scheduler/snippets/pause_job.py index ac502411076..0eb55c44e7d 100644 --- a/cloud_scheduler/snippets/pause_job.py +++ b/cloud_scheduler/snippets/pause_job.py @@ -60,15 +60,30 @@ def pause_job(project_id: str, location_id: str, job_id: str) -> None: if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser(description=pause_job.__doc__.splitlines()[0]) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + args = parser.parse_args() pause_job( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, ) diff --git a/cloud_scheduler/snippets/resume_job.py b/cloud_scheduler/snippets/resume_job.py index 3675839268e..8455192a81b 100644 --- a/cloud_scheduler/snippets/resume_job.py +++ b/cloud_scheduler/snippets/resume_job.py @@ -60,15 +60,30 @@ def resume_job(project_id: str, location_id: str, job_id: str) -> None: if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser(description=resume_job.__doc__.splitlines()[0]) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + args = parser.parse_args() resume_job( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, ) diff --git a/cloud_scheduler/snippets/run_job.py b/cloud_scheduler/snippets/run_job.py index 5e54942941c..bc1ad121316 100644 --- a/cloud_scheduler/snippets/run_job.py +++ b/cloud_scheduler/snippets/run_job.py @@ -58,15 +58,30 @@ def run_job(project_id: str, location_id: str, job_id: str) -> None: if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser(description=run_job.__doc__.splitlines()[0]) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + args = parser.parse_args() run_job( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, ) diff --git a/cloud_scheduler/snippets/tests/entrypoint_test.py b/cloud_scheduler/snippets/tests/entrypoint_test.py new file mode 100644 index 00000000000..8e3b9104551 --- /dev/null +++ b/cloud_scheduler/snippets/tests/entrypoint_test.py @@ -0,0 +1,72 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import pathlib +import runpy +from unittest import mock + +from google.cloud import scheduler_v1 +import pytest + +SAMPLE = str(pathlib.Path(__file__).parent.parent / "get_job.py") +CLIENT = "google.cloud.scheduler_v1.CloudSchedulerClient" + + +def run_entrypoint(monkeypatch: pytest.MonkeyPatch, *flags: str) -> str: + """Runs get_job.py as a script and returns the job name it requested.""" + monkeypatch.setattr("sys.argv", ["get_job.py", *flags]) + with mock.patch(CLIENT) as mock_client_class: + client = mock_client_class.return_value + client.get_job.return_value = scheduler_v1.Job(name="job") + runpy.run_path(SAMPLE, run_name="__main__") + return client.get_job.call_args.kwargs["request"].name + + +def test_entrypoint_reads_environment_variables( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "env-project") + monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "europe-west1") + monkeypatch.setenv("SCHEDULER_JOB_ID", "env-job") + + name = run_entrypoint(monkeypatch) + + assert name == "projects/env-project/locations/europe-west1/jobs/env-job" + + +def test_entrypoint_flags_override_environment_variables( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "env-project") + monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "europe-west1") + monkeypatch.setenv("SCHEDULER_JOB_ID", "env-job") + + name = run_entrypoint(monkeypatch, "--job-id", "flag-job") + + assert name == "projects/env-project/locations/europe-west1/jobs/flag-job" + + +def test_entrypoint_uses_example_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + for env_var in ( + "GOOGLE_CLOUD_PROJECT", + "GOOGLE_CLOUD_LOCATION", + "SCHEDULER_JOB_ID", + ): + monkeypatch.delenv(env_var, raising=False) + + name = run_entrypoint(monkeypatch) + + assert name == ( + "projects/example-project-id/locations/us-central1/jobs/example-job-id" + ) diff --git a/cloud_scheduler/snippets/update_job.py b/cloud_scheduler/snippets/update_job.py index 15ed9678c2b..d9632a900f5 100644 --- a/cloud_scheduler/snippets/update_job.py +++ b/cloud_scheduler/snippets/update_job.py @@ -66,15 +66,30 @@ def update_job(project_id: str, location_id: str, job_id: str) -> None: if __name__ == "__main__": + import argparse import os - def arg(index: int, env_var: str, default: str) -> str: - """Reads an environment variable, then a CLI argument, then a default.""" - cli_value = sys.argv[index] if len(sys.argv) > index else None - return os.environ.get(env_var) or cli_value or default + # Each flag defaults to its environment variable, then to an example value. + parser = argparse.ArgumentParser(description=update_job.__doc__.splitlines()[0]) + parser.add_argument( + "--project-id", + default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"), + help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)", + ) + parser.add_argument( + "--location-id", + default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + help="Region of the job (default: $GOOGLE_CLOUD_LOCATION)", + ) + parser.add_argument( + "--job-id", + default=os.environ.get("SCHEDULER_JOB_ID", "example-job-id"), + help="ID of the job (default: $SCHEDULER_JOB_ID)", + ) + args = parser.parse_args() update_job( - project_id=arg(1, "GOOGLE_CLOUD_PROJECT", "example-project-id"), - location_id=arg(2, "GOOGLE_CLOUD_LOCATION", "us-central1"), - job_id=arg(3, "SCHEDULER_JOB_ID", "example-job-id"), + project_id=args.project_id, + location_id=args.location_id, + job_id=args.job_id, ) From 4016ddb1596d8e5679c2de35684692689ce0b7d4 Mon Sep 17 00:00:00 2001 From: Jennifer Davis Date: Fri, 6 Nov 2026 21:59:30 +0000 Subject: [PATCH 4/4] fix(cloud_scheduler): claim receiver runs atomically and tighten tests Address PR review feedback: - The receiver now checks and claims each run in one step under a lock, before doing the work, so two deliveries of the same run that arrive together can't both do the work. If the work fails, the claim is released so that Cloud Scheduler's retry can process the run. - Receiver tests reset the claimed runs before each test and cover releasing the claim after a failure. - Tests compare the full Dataproc target URI instead of checking for a hostname substring. --- cloud_scheduler/receiver/main.py | 54 ++++++++++++------- cloud_scheduler/receiver/tests/main_test.py | 40 +++++++++----- .../tests/create_http_job_oauth_test.py | 2 +- .../snippets/tests/integration_test.py | 7 ++- 4 files changed, 68 insertions(+), 35 deletions(-) diff --git a/cloud_scheduler/receiver/main.py b/cloud_scheduler/receiver/main.py index 86f664989e2..12e2fbdaa6f 100644 --- a/cloud_scheduler/receiver/main.py +++ b/cloud_scheduler/receiver/main.py @@ -14,15 +14,18 @@ # [START cloudscheduler_receiver] import json +import threading import flask import functions_framework -# Cloud Scheduler can deliver the same scheduled run more than once, so the -# function records which runs it has finished. This in-memory set only lasts -# for the life of one instance; store run keys in a durable database such as -# Firestore so that every instance sees them. -processed_runs: set[str] = set() +# Cloud Scheduler can deliver the same scheduled run more than once, sometimes +# at nearly the same time, so the function claims each run before doing its +# work. These in-memory claims last only for the life of one instance. In +# production, claim each run with a single atomic write, such as creating a +# Firestore document that fails if the document already exists. +claimed_runs: set[str] = set() +claimed_runs_lock = threading.Lock() def log(message: str, **fields: str) -> None: @@ -30,6 +33,16 @@ def log(message: str, **fields: str) -> None: print(json.dumps({"severity": "INFO", "message": message, **fields})) +def process_run(job_name: str, schedule_time: str, payload: str) -> None: + """Does the scheduled work. Replace the body with your own task.""" + log( + "Processed scheduled run", + jobName=job_name, + scheduleTime=schedule_time, + payload=payload, + ) + + @functions_framework.http def handle_scheduled_run(request: flask.Request) -> tuple[str, int]: """Handles a request that a Cloud Scheduler job sends to this function. @@ -47,23 +60,26 @@ def handle_scheduled_run(request: flask.Request) -> tuple[str, int]: # The schedule time stays the same when Cloud Scheduler retries a run, # so the job name and schedule time identify a single scheduled run. run_key = f"{job_name}/{schedule_time}" - if run_key in processed_runs: - log("Skipped duplicate run", jobName=job_name, scheduleTime=schedule_time) - return "Run already processed", 200 + + # Check and claim the run in one step, so that two deliveries of the same + # run that arrive together can't both do the work. + with claimed_runs_lock: + if run_key in claimed_runs: + log("Skipped duplicate run", jobName=job_name, scheduleTime=schedule_time) + return "Duplicate run skipped", 200 + claimed_runs.add(run_key) # Finish the work within the job's attempt deadline. Cloud Scheduler # treats a slow response as a failure even if the work later succeeds. - # If the work raises an exception, the function returns HTTP 500, which - # tells Cloud Scheduler to retry the run according to the job's retry - # settings. - payload = request.get_data(as_text=True) - log( - "Processed scheduled run", - jobName=job_name, - scheduleTime=schedule_time, - payload=payload, - ) - processed_runs.add(run_key) + try: + process_run(job_name, schedule_time, request.get_data(as_text=True)) + except Exception: + # Release the claim so that a retry can process the run. Re-raising + # returns HTTP 500, which tells Cloud Scheduler to retry the run + # according to the job's retry settings. + with claimed_runs_lock: + claimed_runs.discard(run_key) + raise return "Run processed", 200 diff --git a/cloud_scheduler/receiver/tests/main_test.py b/cloud_scheduler/receiver/tests/main_test.py index 050319d8369..74e9552c195 100644 --- a/cloud_scheduler/receiver/tests/main_test.py +++ b/cloud_scheduler/receiver/tests/main_test.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -import uuid +from unittest import mock import flask import pytest @@ -21,6 +21,17 @@ app = flask.Flask(__name__) +HEADERS = { + "X-CloudScheduler-JobName": "test-job", + "X-CloudScheduler-ScheduleTime": "2026-11-06T09:00:00Z", +} + + +@pytest.fixture(autouse=True) +def reset_claimed_runs() -> None: + """Gives each test an empty set of claimed runs.""" + main.claimed_runs.clear() + def call_function( headers: dict[str, str] | None = None, data: str = "example-payload" @@ -29,15 +40,8 @@ def call_function( return main.handle_scheduled_run(flask.request) -def scheduler_headers() -> dict[str, str]: - return { - "X-CloudScheduler-JobName": f"test-job-{uuid.uuid4().hex[:8]}", - "X-CloudScheduler-ScheduleTime": "2026-11-06T09:00:00Z", - } - - def test_first_delivery_is_processed(capsys: pytest.CaptureFixture[str]) -> None: - body, status = call_function(scheduler_headers()) + body, status = call_function(HEADERS) assert status == 200 assert body == "Run processed" @@ -45,16 +49,26 @@ def test_first_delivery_is_processed(capsys: pytest.CaptureFixture[str]) -> None def test_duplicate_delivery_is_skipped(capsys: pytest.CaptureFixture[str]) -> None: - headers = scheduler_headers() - call_function(headers) + call_function(HEADERS) - body, status = call_function(headers) + body, status = call_function(HEADERS) assert status == 200 - assert body == "Run already processed" + assert body == "Duplicate run skipped" assert "Skipped duplicate run" in capsys.readouterr().out +def test_failed_run_releases_claim_for_retry() -> None: + with mock.patch.object(main, "process_run", side_effect=RuntimeError("boom")): + with pytest.raises(RuntimeError): + call_function(HEADERS) + + body, status = call_function(HEADERS) + + assert status == 200 + assert body == "Run processed" + + def test_missing_scheduler_headers_are_rejected() -> None: body, status = call_function() diff --git a/cloud_scheduler/snippets/tests/create_http_job_oauth_test.py b/cloud_scheduler/snippets/tests/create_http_job_oauth_test.py index fa980f82e48..df5f1cedbdb 100644 --- a/cloud_scheduler/snippets/tests/create_http_job_oauth_test.py +++ b/cloud_scheduler/snippets/tests/create_http_job_oauth_test.py @@ -67,7 +67,7 @@ def test_create_http_job_oauth_success( out, err = capsys.readouterr() assert f"Resource ID: {JOB_NAME}" in out assert "State: ENABLED" in out - assert "Target URI: https://dataproc.googleapis.com" in out + assert "Target URI: https://dataproc.googleapis.com/v1/x\n" in out assert err == "" diff --git a/cloud_scheduler/snippets/tests/integration_test.py b/cloud_scheduler/snippets/tests/integration_test.py index 6dfaf5fd875..166b43a4edb 100644 --- a/cloud_scheduler/snippets/tests/integration_test.py +++ b/cloud_scheduler/snippets/tests/integration_test.py @@ -180,9 +180,12 @@ def test_create_http_job_oauth(capsys: pytest.CaptureFixture[str], job_id: str) ) out, err = capsys.readouterr() + expected_uri = ( + f"https://dataproc.googleapis.com/v1/projects/{PROJECT_ID}/regions/" + f"{LOCATION_ID}/workflowTemplates/example-workflow-template:instantiate" + ) assert f"Resource ID: {job_name(job_id)}" in out - assert "dataproc.googleapis.com" in out - assert "example-workflow-template:instantiate" in out + assert f"Target URI: {expected_uri}\n" in out assert "error:" not in err