LogoLogo
PipekitPricingBlogPipekit StatusRelease Notes
  • Introduction
  • Getting Started
  • CLI
    • Cron Workflows
  • Pipekit Agent
    • Helm Install
  • Pipekit
    • Authentication
      • Okta
    • Runs
    • Pipes
      • Managing Pipes
        • Run Conditions
        • Secrets
        • Alerting
      • Pipe Runs
        • Run Graph (DAG)
        • Pod Logs
        • Workflow Logs
        • Workflow YAML
      • Cron Workflows
      • Externally Triggered Workflows
    • Metrics
    • Templates
    • Clusters
    • Organization
      • Creating an Organization
      • Managing Users
      • Managing Alert Providers
      • Settings
      • Permissions
  • Python SDK
    • Jupyter Notebooks
    • Python Scripts
  • Self-Hosting Pipekit
    • Dependencies and Pre-requisites
    • Container Images
    • Kubernetes Permissions
    • Self-Hosted Pipekit Helm Chart
    • License Key
    • Initial Login and Break Glass Account
    • Integrating with your Git Provider
    • Configuring SSO
  • Additional Information
    • Free Trial Cluster
  • REST API
Powered by GitBook
On this page
  1. Python SDK

Python Scripts

Using the Pipekit Python SDK in Python scripts

Last updated 9 months ago

Example

The following example runs the in the argo namespace using the argo-workflow service account via Pipekit.

Use the to generate a token using pipekit hera (ensure you are logged in first).

from hera.workflows import Container, Step, Steps, Workflow, script
from pipekit_sdk.service import PipekitService

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

@script(image="python:alpine3.6", command=["python"], add_cwd_to_sys_path=False)
def flip_coin() -> None:
    import random

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

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-workflow",
) 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")

pipekit.submit(w, "clusterB")
coinflip workflow
Pipekit CLI