Pipekit allows you to manage your workflows at scale. The control plane configures Argo Workflows for you in your infrastructure, enabling you to optimize multi-cluster workloads while reducing your cloud spend. The team at Pipekit is also happy to support you through your Argo Workflows journey via commercial support.

Pipekit Python SDK

Installation

pip install pipekit-sdk

Usage

# The Pipekit SDK interacts with Hera Workflows classes
from hera.workflows import Container, Step, Steps, Workflow, script
from pipekit_sdk.service import PipekitService

# Create a Pipekit service that is used to talk to the Pipekit API
pipekit = PipekitService(token="<token>")

# List clusters and Pipes
clusters = pipekit.list_clusters()
pipes = pipekit.list_pipes()

@script()
def flip_coin() -> None:
    import random

    result = "heads" if random.randint(0, 1) == 0 else "tails"
    print(result)

# Create a Workflow using Hera
with Workflow(
    generate_name="coinflip-",
    annotations={
        "workflows.argoproj.io/description": (
            "This is an example of coin flip defined as a sequence of conditional steps."
        ),
    },
    entrypoint="coinflip",
    namespace="argo",
    service_account_name="argo",
) as w:
    heads = Container(
        name="heads",
        image="alpine:3.6",
        command=["sh", "-c"],
        args=['echo "it was heads"'],
    )
    tails = Container(
        name="tails",
        image="alpine:3.6",
        command=["sh", "-c"],
        args=['echo "it was tails"'],
    )

    with Steps(name="coinflip") as s:
        fc: Step = flip_coin()

        with s.parallel():
            heads(when=f"{fc.result} == heads")
            tails(when=f"{fc.result} == tails")

# Submit the Workflow to Pipekit
pipekit.submit(w, "<cluster-name>")

# Tail the logs
pipekit.print_logs(pipe_run.uuid)

Connecting to Pipekit

pipekit_url is the single base URL for the ID, Users, and UI APIs. This is correct when they sit behind one gateway, which is the default (https://api.pipekit.io).

When the services are reachable on separate hosts, set id_url and users_url (or the PIPEKIT_ID_URL and PIPEKIT_USERS_URL env vars). The SDK logs in against id_url and makes every other call against users_url.

pipekit = PipekitService(
    username="<user>",
    password="<password>",
    id_url="http://id.internal:8080",
    users_url="http://users.internal:8080",
)

insecure=True (or PIPEKIT_INSECURE=true) skips TLS verification. Use it only for testing against a cluster with a self-signed certificate, never in production.

timeout (or PIPEKIT_TIMEOUT) sets the per-request timeout in seconds, default 10. Raise it for slow links or for the cron lifecycle calls, which can take up to the server's notification timeout to return. The log stream is exempt and stays unbounded.

Managing CronWorkflows

You can create, update, and delete a CronWorkflow from Python. The namespace must match the one in the manifest on every call (the platform default is argo). A wrong namespace makes the cron look missing.

from hera.workflows import Container, CronWorkflow
from pipekit_sdk.service import PipekitService

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

with CronWorkflow(
    name="daily-demand-forecast",
    namespace="argo",
    entrypoint="main",
    # Argo Workflows 3.6 deprecated the singular spec.schedule. Use schedules.
    schedules=["*/5 * * * *"],
    service_account_name="argo",
) as cron:
    Container(name="main", image="alpine", command=["sh", "-c", "echo hello"])

# Create
pipekit.create(cron, "<cluster-name>")

# Update: the namespace is taken from the manifest when not passed
updated = pipekit.update_cron(cron, "<cluster-name>")

# Suspend / resume scheduling
pipekit.suspend_cron("<cluster-name>", "argo", "daily-demand-forecast")
pipekit.resume_cron("<cluster-name>", "argo", "daily-demand-forecast")

# Get the current state
current = pipekit.get_cron("<cluster-name>", "argo", "daily-demand-forecast")

# Delete
pipekit.delete_cron("<cluster-name>", "argo", "daily-demand-forecast")

Further help

Please refer to the Pipekit Documentation for more information.