Reusable workflow jobs

After the first job works, use the same pattern for your own command:

1. Choose an image containing the required program.
2. Create an App.
3. Decorate a function with @app.function.
4. Accept the values that vary between runs as function arguments.
5. Return the command that should run in CGC.
6. Call submit() with real values.

Run image processing with PyTorch

The first workflow tutorial shows how to place a short program directly in the workflow function. For a larger program, keep the processing code in a normal Python file and use the workflow function only to start it with the correct arguments.

1. Create the image-processing program

Create process_image.py in your own project. This is a file written and maintained by you; it is not provided by CGC or cgcsdk.

View full process_image.py

process_image.py

import argparse

import numpy as np
import torch
from PIL import Image


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input",
        required=True,
        help="Path to the source JPEG or PNG inside the job container.",
    )
    parser.add_argument(
        "--output",
        required=True,
        help="Path where the processed image will be saved.",
    )
    args = parser.parse_args()

    image = Image.open(args.input).convert("RGB")
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    pixels = torch.from_numpy(np.array(image)).to(device)
    processed = 255 - pixels

    output = Image.fromarray(processed.cpu().numpy().astype(np.uint8))
    output.save(args.output)

    print(f"Processed image on {device} and saved it to {args.output}")


if __name__ == "__main__":
    main()

Pillow reads and writes the image file. PyTorch represents its pixels as a tensor and calculates 255 - pixels, which creates a color-negative image. The tensor uses the GPU when CUDA is available and otherwise uses the CPU.

In a real project, the same file can resize or normalize images, load a PyTorch model, run prediction, and draw or save the result.

Argument	Meaning
--input	Path to the source JPEG or PNG inside the job container.
--output	Path where the processed image will be saved.

2. Start Filebrowser

Filebrowser provides a web interface for uploading, downloading, and deleting files on a CGC volume. Start it once if it is not already running:

cgc compute filebrowser create

The workflow runner in the next step creates the volume and mounts it to Filebrowser for you. If you run it again, it reuses the existing volume and the existing Filebrowser mount.

You do not need to build a container image for this example. The selected public image already contains Python, PyTorch, NumPy, and CUDA. The workflow installs Pillow before running the script. The volume provides the files that are specific to your task.

3. Create the workflow runner

Create a second file named run_image_workflow.py in your own project:

View full run_image_workflow.py

run_image_workflow.py

import argparse
import shlex
import time
from pathlib import PurePosixPath

from cgc.sdk import App, Image, Volume, volume as volume_api
from cgc.sdk.exceptions import SDKException


VOLUME_NAME = "image-processing-data"
MOUNT_PATH = "/data"
FILEBROWSER_NAME = "filebrowser"


def is_mount_details(value):
    return (
        isinstance(value, dict)
        and value.get("pvc_name") == VOLUME_NAME
        and value.get("target_template_name") == FILEBROWSER_NAME
        and "mount_path" in value
    )


def get_mount_details_from_error(error):
    for value in error.args:
        if is_mount_details(value):
            return value
        if isinstance(value, dict) and is_mount_details(value.get("details")):
            return value["details"]
    return None


def prepare_volume():
    data = Volume.from_name(
        name=VOLUME_NAME,
        mount_path=MOUNT_PATH,
        size=2,
        storage_class="ssd-rwx-ec",
        create_if_missing=True,
    )
    data.ensure()
    deadline = time.monotonic() + 300
    while not data.ready():
        if time.monotonic() >= deadline:
            raise RuntimeError("Volume did not become ready: " + VOLUME_NAME)
        time.sleep(5)

    try:
        response = volume_api.volume_mount(
            name=VOLUME_NAME,
            target=FILEBROWSER_NAME,
        )
    except SDKException as error:
        details = get_mount_details_from_error(error)
        if details is None and error.reason == "PVC_MOUNT_ALREADY_MOUNTED":
            print("Volume is already mounted to Filebrowser.")
        elif details is not None:
            print(
                "Volume is mounted to Filebrowser. "
                f"Filebrowser mount path: {details['mount_path']}"
            )
        else:
            raise
    else:
        details = response.get("details", {})
        if response["code"] != 200 and not is_mount_details(details):
            raise RuntimeError(f"Could not mount volume to Filebrowser: {response}")
        if "mount_path" in details:
            print("Filebrowser mount path:", details["mount_path"])

    return data


def create_process_image_task(data):
    app = App(
        name="image-processing-workflow",
        image=Image.from_registry("pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime"),
        volumes=[data],
    )

    @app.function(
        name="process-image",
        cpu=4,
        memory=16,
        gpu=1,
        gpu_type="nvidia-rtx-a5000",
    )
    def process_image(input_path, output_path):
        return (
            "test -f /data/process_image.py"
            f" && test -f {shlex.quote(input_path)}"
            " && python -m pip install --quiet Pillow"
            " && python /data/process_image.py"
            f" --input {shlex.quote(input_path)}"
            f" --output {shlex.quote(output_path)}"
        )

    return process_image

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input",
        action="append",
        help="Input image path inside the job container. Repeat for multiple images.",
    )
    parser.add_argument(
        "--output",
        action="append",
        help="Output image path inside the job container. Repeat for multiple images.",
    )
    args = parser.parse_args()
    input_paths = args.input or ["/data/input.jpg"]
    output_paths = args.output or ["/data/output.jpg"]

    if len(input_paths) != len(output_paths):
        parser.error("Provide the same number of --input and --output values.")

    for path in input_paths + output_paths:
        candidate = PurePosixPath(path)
        if ".." in candidate.parts or not candidate.is_relative_to(MOUNT_PATH):
            parser.error("Input and output paths must be inside " + MOUNT_PATH)

    data = prepare_volume()
    process_image = create_process_image_task(data)

    print("Volume is ready and mounted to Filebrowser:", VOLUME_NAME)
    print("Open Filebrowser, open the mounted volume, and upload to its root:")
    print("- process_image.py")
    for input_path in input_paths:
        print(f"- {input_path.removeprefix(MOUNT_PATH + '/')}")
    print()
    print("Use this command to find the Filebrowser URL and app token:")
    print("cgc compute list -d")
    print("Sign in with username 'admin' and the displayed app token.")
    input("Press Enter after all files are uploaded...")

    runs = process_image.for_each(input_paths, output_paths)
    failed = []

    for run in runs:
        status = run.wait(timeout_seconds=600)
        print("Job name:", run.name)
        print("Final status:", status)
        print(run.logs(timestamps=True))

        if status != "Succeeded":
            failed.append(run.name)
        else:
            run.delete()
            print("Deleted successful job:", run.name)

    if failed:
        raise RuntimeError(f"Failed jobs: {', '.join(failed)}")


if __name__ == "__main__":
    main()

Run it:

python run_image_workflow.py

The script:

1. Creates or reuses the image-processing-data volume.
2. Mounts the volume to Filebrowser, or reuses the existing mount.
3. Asks you to upload process_image.py and the input image files.
4. Submits one CGC workflow job per input image.
5. Waits for each job and prints status and logs.

The workflow also mounts the same volume at /data, so the uploaded files are available inside the job container as:

/data/process_image.py
/data/input.jpg

Value	Meaning
VOLUME_NAME	Persistent volume used for the script, input image, and output image.
MOUNT_PATH="/data"	Path where the volume is mounted inside the job container.
FILEBROWSER_NAME="filebrowser"	Name of the Filebrowser compute resource.
image=...	Public PyTorch runtime image containing Python, PyTorch, NumPy, and CUDA.
volumes=[data]	Mounts the volume containing the script and data at /data.
test -f /data/process_image.py	Checks that the uploaded processing script exists before starting the job command.
test -f input_path	Checks that the selected input image exists before starting the processing script.
python -m pip install --quiet Pillow	Installs the library used to open and save JPEG or PNG files.
python /data/process_image.py	Runs the script uploaded through Filebrowser.
input_path	Image path passed to process_image.py.
output_path	Output image path passed to process_image.py.
cpu=4	Allocates four CPU cores.
memory=16	Allocates 16 GiB of memory.
gpu=1	Allocates one GPU.
gpu_type="nvidia-rtx-a5000"	Selects the required GPU type.

After the job finishes, /data/output.jpg remains on the volume and can be downloaded through Filebrowser.

Importing the script creates no resources. Only successful jobs are deleted after log retrieval; the volume is preserved.

Understand persistent and temporary paths

Files saved under /data are written to the mounted image-processing-data volume. They remain available after the job container is removed and can be used by later jobs.

Files saved outside the mounted volume belong only to the temporary job container. For example, /tmp/output.jpg disappears when that container is removed.

CGC currently supports a custom mount_path for only one volume in a workflow step.

If your program does not use a GPU, choose a CPU-compatible base image and omit both gpu and gpu_type from @app.function. The job then uses the default gpu=0 while still receiving the requested CPU and memory.

Build a custom image only when an existing image does not contain the packages, system libraries, or tools required by your program. For a short script whose dependencies are already available in the selected image, mounting it from a volume is enough.

Quote user-provided values

Do not insert paths, URLs, or names directly into shell commands:

# Unsafe
return f"python /data/process_image.py --input {input_path}"

# Safe
return (
    "python /data/process_image.py"
    f" --input {shlex.quote(input_path)}"
)

Use shlex.quote() for paths and other values inserted into shell commands.

Run the same task for many inputs

Upload photo-a.jpg and photo-b.jpg to the volume through Filebrowser. Then use the previously defined process_image task to process both images:

input_paths = [
    "/data/photo-a.jpg",
    "/data/photo-b.jpg",
]
output_paths = [
    "/data/photo-a-processed.jpg",
    "/data/photo-b-processed.jpg",
]

for_each() passes values at the same position to process_image:

• The first job receives /data/photo-a.jpg and /data/photo-a-processed.jpg.
• The second job receives /data/photo-b.jpg and /data/photo-b-processed.jpg.

This creates two independent CGC jobs, not four. Each job runs the same process_image.py script with different --input and --output arguments. The processed images remain on the mounted volume and can be downloaded through Filebrowser.

Delete the volume

Delete the volume only when its scripts, input images, and results are no longer needed:

data.delete()

warning

Deleting a volume permanently deletes all of its files. Removing a finished job does not remove its volume, and the volume should not be deleted automatically at the end of this workflow.

Use the API reference for all job configuration arguments and JobRun methods.

The complete process_image.py and run_image_workflow.py scripts are available on the Workflow examples page.