Metadata-Version: 2.4
Name: pipekit-sdk
Version: 7.4.9
Summary: Pipekit Python SDK
License: BSD-3-Clause
License-File: LICENSE
Keywords: pipekit,hera,argo,workflows,orchestration
Author: Pipekit
Author-email: ci@pipekit.io
Requires-Python: >=3.9,<4
Classifier: License :: OSI Approved :: BSD License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Dist: hera (>=5.5.2)
Requires-Dist: pydantic (<3)
Requires-Dist: pyjwt (>=2.9.0,<3.0.0)
Requires-Dist: responses (>=0.25.3,<0.26.0)
Project-URL: Documentation, https://docs.pipekit.io/
Project-URL: Homepage, https://pipekit.io
Description-Content-Type: text/markdown

[![Pipekit Logo](https://helm.pipekit.io/assets/pipekit-logo.png)](https://pipekit.io)

[Pipekit](https://pipekit.io) 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

```bash
pip install pipekit-sdk
```

## Usage

```py
# 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`.

```py
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.

```py
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](https://docs.pipekit.io) for more information.

