Python SDK

Using the Pipekit Python SDK

The Pipekit Python SDK lets you talk to the Pipekit API from Python. It's the right surface for use in Jupyter Notebooks, inside Python scripts, and for building automation against Pipekit programmatically.

The SDK wraps the REST API and is designed to pair well with Hera for authoring workflows in Python.

Install

pip install pipekit-sdk

The SDK targets Python 3.10+.

Authenticate

The SDK uses a bearer token, which you generate with the Pipekit CLI:

pipekit login          # one-time, opens browser for SSO or prompts for password
pipekit hera           # prints a token to stdout

Pass the token to a PipekitService:

from pipekit_sdk.service import PipekitService

pipekit = PipekitService(token="<token>")

For longer-lived programs, set PIPEKIT_TOKEN and pull it from the environment:

import os
from pipekit_sdk.service import PipekitService

pipekit = PipekitService(token=os.environ["PIPEKIT_TOKEN"])

Submit a Workflow

from hera.workflows import DAG, Workflow, script

@script(image="python:3.12")
def echo(message):
    print(message)

with Workflow(
    generate_name="dag-diamond-",
    entrypoint="diamond",
    namespace="argo",
    service_account_name="argo-workflow",
) as w:
    with DAG(name="diamond"):
        A = echo(name="A", arguments={"message": "A"})
        B = echo(name="B", arguments={"message": "B"})
        C = echo(name="C", arguments={"message": "C"})
        D = echo(name="D", arguments={"message": "D"})
        A >> [B, C] >> D

pipe_run = pipekit.submit(w, "<cluster-name>")
print(pipe_run["uuid"])

submit returns the pipe_run dictionary including the Run's UUID.

List Runs

runs = pipekit.list_runs(cluster_name="<cluster-name>")
for r in runs:
    print(r["uuid"], r["status"])

Filter by Pipe, namespace, or status. See the SDK source for the full parameter list.

Stop, terminate, or restart a Run

pipekit.stop(run_uuid="<run-uuid>")
pipekit.terminate(run_uuid="<run-uuid>")
pipekit.restart(run_uuid="<run-uuid>")

stop runs exit handlers; terminate does not. restart resubmits with the same parameters and returns a new pipe_run.

Stream logs

pipekit.print_logs(pipe_run["uuid"])

For non-blocking access, use the lower-level log fetcher (see SDK source).

CronWorkflows

The SDK supports the full CronWorkflow lifecycle:

from hera.workflows import Container, CronWorkflow

with CronWorkflow(
    name="cron-wf-example",
    namespace="argo",
    entrypoint="main",
    schedule="*/5 * * * *",
    concurrency_policy="Replace",
    service_account_name="argo-workflow",
) as cw:
    main = Container(
        name="main",
        image="alpine",
        command=["sh", "-c", "echo I am a CronWorkflow"],
    )

pipekit.create(cw, "<cluster-name>")
# Later:
pipekit.suspend(cw)
pipekit.resume(cw)
pipekit.update(cw)
pipekit.delete(cw)

What's next

• Jupyter Notebooks: a worked example with log streaming.

• Python Scripts: a worked example of a coinflip workflow.

• REST API: if you'd rather call the API directly.

• CLI > Using with Hera: token generation flow for Hera users.