Your first workflow job

This workflow downloads a text or CSV file, counts its lines, and prints the result in the job logs. The complete example can also save the line count to a mounted volume when you provide volume arguments.

Complete script

View full process_file_workflow.py

#!/usr/bin/env python3
"""Download a text/CSV file in a CGC workflow job and count its lines."""

import argparse
import json
import shlex
import time
from pathlib import PurePosixPath

from cgc.sdk import App, Image, Volume


DEFAULT_INPUT_URL = "https://raw.githubusercontent.com/cs109/2014_data/master/countries.csv"


def wait_for_job(run, timeout_seconds: int, poll_seconds: int = 5):
    deadline = time.time() + timeout_seconds
    last_status = None

    while time.time() < deadline:
        status = run.status()
        if status != last_status:
            print("Job status:", status)
            last_status = status
        if status in {"Succeeded", "Failed"}:
            return status
        time.sleep(poll_seconds)

    return "Timeout"


def print_job_logs(run):
    response = run.logs(timestamps=True)
    logs = response.get("details", {}).get("logs", {})

    if not logs:
        print("No logs returned.")
        return

    print("Job logs:")
    for pod_name, containers in logs.items():
        for container_name, text in containers.items():
            print(f"[{pod_name}/{container_name}]")
            print(text.rstrip() or "<empty>")


def create_process_file_task(volume=None):
    app = App(
        name="file-processing-workflow",
        image=Image.from_registry("python:3.11-slim"),
        volumes=[volume] if volume else [],
    )


    @app.function(
        name="count-file-lines",
        cpu=1,
        memory=2,
        ttl_seconds_after_finished=3600,
    )
    def process_file(input_url: str, output_path: str = ""):
        code = f"""
from pathlib import Path
from urllib.request import urlopen

input_url = {json.dumps(input_url)}
output_path = {json.dumps(output_path)}

with urlopen(input_url, timeout=60) as response:
    text = response.read().decode("utf-8")

line_count = len(text.splitlines())
print(f"Downloaded: {{input_url}}")
print(f"Line count: {{line_count}}")

if output_path:
    path = Path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(str(line_count), encoding="utf-8")
    print(f"Saved line count to: {{path}}")
"""
        return f"python -c {shlex.quote(code)}"

    return process_file


def wait_for_volume(volume, timeout_seconds=300):
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        if volume.ready():
            return
        time.sleep(5)
    raise RuntimeError(f"Volume '{volume.name}' did not become ready")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Count lines in a remote text or CSV file using a CGC job."
    )
    parser.add_argument(
        "--input-url",
        default=DEFAULT_INPUT_URL,
        help="Public URL of the text or CSV file processed by the CGC job.",
    )
    parser.add_argument(
        "--volume",
        help="CGC volume name used to persist the result file.",
    )
    parser.add_argument(
        "--volume-size",
        type=int,
        default=20,
        help="Volume capacity in GB when a missing volume is created.",
    )
    parser.add_argument(
        "--storage-class",
        help="Storage class for a new volume; omit to use the configured default.",
    )
    parser.add_argument(
        "--mount-path",
        default="/data",
        help="Path where the volume is mounted inside the job container.",
    )
    parser.add_argument(
        "--output",
        help="Result file path. With a volume, it must be inside --mount-path.",
    )
    parser.add_argument(
        "--delete-job",
        action="store_true",
        help="Delete this job after successful completion and log retrieval.",
    )
    parser.add_argument(
        "--timeout-seconds",
        type=int,
        default=300,
        help="Maximum number of seconds to wait for the job.",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    if args.timeout_seconds <= 0:
        raise ValueError("--timeout-seconds must be positive")
    if args.volume:
        mount = PurePosixPath(args.mount_path)
        if not mount.is_absolute() or ".." in mount.parts:
            raise ValueError("--mount-path must be an absolute container path without '..'")
    volume = None
    output_path = args.output or ""

    if args.volume:
        volume = Volume.from_name(
            name=args.volume,
            mount_path=args.mount_path,
            size=args.volume_size,
            storage_class=args.storage_class,
            create_if_missing=True,
        )
        if not output_path:
            output_path = str(PurePosixPath(args.mount_path) / "line-count.txt")
        if ".." in PurePosixPath(output_path).parts or not PurePosixPath(output_path).is_relative_to(
            PurePosixPath(args.mount_path)
        ):
            raise ValueError("--output must be inside --mount-path")
        volume.ensure()
        wait_for_volume(volume, args.timeout_seconds)
    elif output_path:
        print(
            "Warning: no volume was selected. The output file will disappear "
            "when the job container is removed."
        )

    process_file = create_process_file_task(volume)
    run = process_file.submit(args.input_url, output_path)
    print("Created job:", run.name)

    status = wait_for_job(run, timeout_seconds=args.timeout_seconds)

    print("Final status:", status)

    if status == "Timeout":
        print("The job did not finish before the timeout.")
        print("Check platform events with:")
        print(f"cgc events {run.name}")
    else:
        print_job_logs(run)

    if status != "Succeeded":
        raise RuntimeError(f"Job '{run.name}' ended with status {status}; left for diagnosis")

    if args.delete_job:
        run.delete()
        print("Deleted job:", run.name)


if __name__ == "__main__":
    main()

Failed and Timeout raise an error and leave the job for inspection. --delete-job removes only a successful job after log retrieval.

Step 1: Choose image

image=Image.from_registry("python:3.11-slim")

python:3.11-slim is a public container image with Python 3.11. The remote command uses only the Python standard library, so no additional packages are required.

Step 2: Create an application

app = App(
    name="file-processing-workflow",
    image=...,
    volumes=[volume] if volume else [],
)

App stores settings shared by its tasks:

• name is the workflow name and resource-name prefix.
• image is the default image used by tasks in the workflow.
• volumes attaches persistent storage when the script creates or reuses a Volume.

This line does not create anything in CGC.

Step 3: Define your task

@app.function(
    name="count-file-lines",
    cpu=1,
    memory=2,
    ttl_seconds_after_finished=3600,
)
def process_file(input_url, output_path=""):
    ...

process_file is an example function name.

Value	Meaning
name="count-file-lines"	Prefix used for generated CGC job names.
cpu=1	Reserve one CPU core.
memory=2	Reserve 2 GB of memory.
ttl_seconds_after_finished=3600	Remove the finished job after one hour.
input_url	Your function argument. It identifies the file to process.
output_path	Optional path where the job writes the line count. Use a mounted volume path for persistent output.

Step 4: Return the remote command

The function builds Python code containing the selected URL and returns a shell command:

return f"python -c {shlex.quote(code)}"

json.dumps(input_url) safely places the URL inside Python source. shlex.quote(code) safely places that source inside a shell command.

The function itself runs locally. The returned python -c ... command runs in CGC.

Step 5: Create the job

run = process_file.submit("https://example.com/input.csv", "")

submit() passes the URL and optional output path to process_file(input_url, output_path), receives the generated command, and creates a CGC job. It returns a JobRun object representing that specific job.

Step 6: Wait and read logs

status = wait_for_job(run, timeout_seconds=300)
if status == "Timeout":
    print(f"Check platform events with: cgc events {run.name}")
else:
    print_job_logs(run)

• timeout_seconds=300 waits for up to five minutes.
• logs() returns output printed by the remote command.
• print_job_logs() extracts the useful container log text from the SDK response and prints it directly.
• timestamps=True includes log timestamps.

Possible final values include Succeeded, Failed, and Timeout. A timeout does not stop the CGC job. If a job stays in Pending, inspect platform events:

cgc events <job-name>

Pending means CGC accepted the job, but the pod has not successfully started yet. Common causes include image pulling, quota limits, volume mounting, or a platform startup issue.

Create and run the script

Create a new file in your own project, for example:

process_file_workflow.py

Paste the complete script from the beginning of this page into that file. The filename is your choice and has no special meaning to the CGC SDK.

Run the file from the directory where you created it:

python process_file_workflow.py

This command:

1. Runs the Python file you created on your computer.
2. Uses the default public CSV URL from DEFAULT_INPUT_URL.
3. Calls your process_file workflow function.
4. Creates a CGC job that downloads the file and counts its lines.
5. Waits for the job and prints its status and logs.

To save the line count on a persistent volume, add volume arguments:

python process_file_workflow.py \
  --volume file-processing-data \
  --storage-class ssd-rwx-ec \
  --output /data/line-count.txt

--storage-class must name a class available in your environment; supply it when no default is configured.

The script creates or reuses the volume, mounts it at /data, and validates that --output is inside that mount path.

To process a different public text or CSV file, pass --input-url with your own URL.

The only CGC-specific dependency is the installed package used by the imports:

from cgc.sdk import App, Image, Volume

See the complete process_file_workflow.py script on the Workflow examples page.

Next, follow Reusable workflow jobs to run your own Python file, mount persistent storage, upload files with Filebrowser, and process multiple images.