pipekit-sdk 2.1.2

Pipekit Python SDK

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)

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.