Python API guide

Discover processes, submit work, and inspect results using cuiman.Client. For installation and client concepts, start with Getting Started.

Start the local service

The examples use Eozilla's local test service, which needs no credentials. From a development checkout with pixi install completed, keep this command running in a separate terminal:

pixi run serve

In another terminal, run pixi shell from the repository root and start Python or Jupyter. Run the following blocks in order in the same session. Each helper definition is followed by its call; keep the returned client and job ID for subsequent steps. The complete source is api.py.

Create a client

from cuiman import Client
from gavicore.models import JobResults, JobStatus, ProcessDescription, ProcessRequest


def create_client() -> Client:
    """Connect to the local test service without authentication."""
    return Client(api_url="http://127.0.0.1:8008", auth={"auth_type": "none"})



client = create_client()

The explicit URL and auth_type="none" select the local service. To connect to your own deployment, use its configuration instead. Constructing a client does not log in. Requests authenticate using available credentials; call client.login() first when interactive sign-in is needed. See Authentication.

Discover processes

def inspect_process(client: Client, process_id: str) -> ProcessDescription:
    """List processes and return the selected process description."""
    print(client.get_processes().model_dump_json(indent=2))
    process = client.get_process(process_id)
    print(process.model_dump_json(indent=2))
    return process



process_id = "primes_between"
inspect_process(client, process_id)

The local primes_between process returns prime numbers between two input values. Other services expose different process IDs, inputs, and outputs; inspect their descriptions before preparing a request. You can also inspect the service with client.get_capabilities() and client.get_conformance().

Submit once

def submit_process(client: Client, process_id: str, request: ProcessRequest) -> str:
    """Submit once and return the server-assigned job ID."""
    job = client.execute_process(process_id, request=request)
    print(job.model_dump_json(indent=2))
    return job.jobID



request = ProcessRequest(inputs={"min_val": 10, "max_val": 80})
job_id = submit_process(client, process_id, request)

Submission starts an asynchronous job and returns its information. The returned job_id identifies this execution; running the submission again creates a new job. The request is a ProcessRequest; a request dictionary is also accepted by client.execute_process().

Monitor and retrieve results

def inspect_results(client: Client, job_id: str) -> JobResults | None:
    """Check once and retrieve results only for a successful job."""
    job = client.get_job(job_id)
    print(job.model_dump_json(indent=2))
    if job.status in (JobStatus.accepted, JobStatus.running):
        print("Check this job again later; do not submit it again.")
        return None
    if job.status != JobStatus.successful:
        print("Job did not succeed. Inspect its status and message before retrying.")
        return None
    results = client.get_job_results(job_id)
    print(results.model_dump_json(indent=2))
    return results



results = inspect_results(client, job_id)

If the job is still accepted or running, repeat only inspect_results(client, job_id) later. A successful result contains the prime numbers from 11 through 79. Failed or dismissed jobs have no successful result; inspect the job's status and message. client.get_jobs() lists jobs.

To try failure handling deliberately, define and call:

def submit_failure_example(client: Client) -> str:
    """Create a short job that deliberately fails halfway through."""
    return submit_process(
        client,
        "sleep_a_while",
        ProcessRequest(inputs={"duration": 2, "fail": True}),
    )



failed_job_id = submit_failure_example(client)

After roughly two seconds, use inspect_results(client, failed_job_id) to see the failure. To cancel a running job or delete a finished one, call client.dismiss_job(job_id) with the specific job you intend to dismiss.

For file or dataset outputs, Result openers explains the difference between retrieving result references and opening their data.

Close the client

When finished with this session:

client.close()

In a script, put the workflow in try and cleanup in finally, as shown by example_session() in the source file. Run that complete example with:

python -m examples.guides.cuiman.api

It submits one job, checks once, and closes the client. If results are not ready, retain its printed job ID and inspect it with a new client or the CLI.

See Getting Started for AsyncClient, and the API Reference for all methods. The original API notebook and Airflow notebook remain available as independent historical examples; their saved outputs and setup instructions may differ from the current client.