feat(cloud_scheduler): modernize samples and cover documented job tasks

cloud_scheduler/receiver/.python-version

@@ -0,0 +1 @@
+3.13

cloud_scheduler/snippets/app.yaml → 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]

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]

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": {},
+}

cloud_scheduler/receiver/requirements-test.txt

@@ -0,0 +1 @@
+pytest==9.0.3

cloud_scheduler/receiver/requirements.txt

@@ -0,0 +1 @@
+functions-framework==3.10.2

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.

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:")

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.

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,
+    )