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..12e2fbdaa6f --- /dev/null +++ b/cloud_scheduler/receiver/main.py @@ -0,0 +1,86 @@ +# 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 threading + +import flask +import functions_framework + +# 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: + """Writes a structured log entry that Cloud Logging parses from stdout.""" + 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. + + 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. + """ + 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}" + + # 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. + 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 + + +# [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..74e9552c195 --- /dev/null +++ b/cloud_scheduler/receiver/tests/main_test.py @@ -0,0 +1,76 @@ +# 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 + +import flask +import pytest + +from .. import main + +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" +) -> tuple[str, int]: + with app.test_request_context("/", method="POST", headers=headers, data=data): + return main.handle_scheduled_run(flask.request) + + +def test_first_delivery_is_processed(capsys: pytest.CaptureFixture[str]) -> None: + body, status = call_function(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: + call_function(HEADERS) + + body, status = call_function(HEADERS) + + assert status == 200 + 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() + + 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..2040605313c --- /dev/null +++ b/cloud_scheduler/snippets/create_appengine_job.py @@ -0,0 +1,110 @@ +# 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 argparse + import os + + # 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=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 new file mode 100644 index 00000000000..7cef27d08f2 --- /dev/null +++ b/cloud_scheduler/snippets/create_http_job_oauth.py @@ -0,0 +1,142 @@ +# 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 argparse + import os + + # 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 94e52c295d3..93b293dc6df 100644 --- a/cloud_scheduler/snippets/create_job.py +++ b/cloud_scheduler/snippets/create_job.py @@ -1,60 +1,136 @@ -# 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 can require authentication. - 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 argparse + import os + + # 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 new file mode 100644 index 00000000000..3120cd07dae --- /dev/null +++ b/cloud_scheduler/snippets/create_job_with_retry.py @@ -0,0 +1,146 @@ +# 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 argparse + import os + + # 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 new file mode 100644 index 00000000000..21a352f6901 --- /dev/null +++ b/cloud_scheduler/snippets/create_pubsub_job.py @@ -0,0 +1,113 @@ +# 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 argparse + import os + + # 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=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 3f456c5b83a..5852679ff1f 100644 --- a/cloud_scheduler/snippets/delete_job.py +++ b/cloud_scheduler/snippets/delete_job.py @@ -1,40 +1,83 @@ # 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 argparse + import os + + # 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=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 new file mode 100644 index 00000000000..cddbeead8b2 --- /dev/null +++ b/cloud_scheduler/snippets/get_job.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. + +# [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 argparse + import os + + # 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=args.project_id, + location_id=args.location_id, + job_id=args.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..990bba591aa --- /dev/null +++ b/cloud_scheduler/snippets/list_jobs.py @@ -0,0 +1,77 @@ +# 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 argparse + import os + + # 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=args.project_id, + location_id=args.location_id, + ) 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..0eb55c44e7d --- /dev/null +++ b/cloud_scheduler/snippets/pause_job.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. + +# [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 argparse + import os + + # 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=args.project_id, + location_id=args.location_id, + job_id=args.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..8455192a81b --- /dev/null +++ b/cloud_scheduler/snippets/resume_job.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. + +# [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 argparse + import os + + # 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=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 new file mode 100644 index 00000000000..bc1ad121316 --- /dev/null +++ b/cloud_scheduler/snippets/run_job.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. + +# [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 argparse + import os + + # 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=args.project_id, + location_id=args.location_id, + job_id=args.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..df5f1cedbdb --- /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/v1/x\n" 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/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/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..166b43a4edb --- /dev/null +++ b/cloud_scheduler/snippets/tests/integration_test.py @@ -0,0 +1,258 @@ +# 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() + 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 f"Target URI: {expected_uri}\n" 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..d9632a900f5 --- /dev/null +++ b/cloud_scheduler/snippets/update_job.py @@ -0,0 +1,95 @@ +# 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 argparse + import os + + # 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=args.project_id, + location_id=args.location_id, + job_id=args.job_id, + )