Kestra Python Client

Warnings

• breaking Starting from Kestra 0.23.0 (and confirmed in 0.24.0 and 1.x), a tenant context is required. The `defaultTenant` (null tenant) is no longer supported. For Kestra Open Source, a fixed tenant named 'main' is always used, and all API URIs now include the tenant, e.g., `/api/v1/...` became `/api/v1/main/...`.
  Fix: Ensure all API calls and client configurations specify a `tenant` ID. For Open Source, use `main`. Update any hardcoded API paths to include the tenant.
• breaking With Kestra 0.24.0, basic authentication became required. Additionally, IAM and API endpoint changes were introduced.
  Fix: Configure the `KestraClient` with either `username` and `password`, or an `access_token` (for service accounts). Avoid making unauthenticated API calls.
• breaking For users of S3 or GCS as internal storage, Kestra now removes the leading root slash in all storage paths to fix double slashes. This impacts how internal storage objects are accessed.
  Fix: Run the provided migration scripts for your storage provider to update object paths. Failure to do so will prevent Kestra from accessing files with the old path structure.
• gotcha When running Python scripts within Kestra flows, managing dependencies can be tricky. Installing packages with `beforeCommands` downloads and installs them on every run, increasing execution time.
  Fix: For efficient dependency management, consider building a custom Docker image that includes Python and all required pip packages, or install packages at server startup if using the `Process Task Runner`. Alternatively, use caching for virtual environments in `WorkingDirectory` tasks.
• gotcha The Python client package on PyPI is `kestra`, but the internal modules for programmatic client interaction (e.g., `Configuration`, `KestraClient`) are imported from `kestrapy`. For utilities used inside Kestra Python script tasks (e.g., `Kestra.outputs`), the import is `from kestra import Kestra`.
  Fix: Be mindful of the specific import paths based on whether you are writing an external Python script to interact with Kestra (use `kestrapy`) or writing a Python script to be executed within a Kestra flow (use `kestra`).

Install

• pip install kestra Install Kestra Python Client
• pip install python-dotenv # Optional, for managing environment variables Install python-dotenv

Imports

• Flow

  from kestra import Flow

  Used for programmatic execution and management of Kestra flows.
• Kestra

  from kestra import Kestra

  Used within Python script tasks inside Kestra flows for sending outputs, metrics, and logs.
• Configuration, KestraClient

  from kestrapy import Configuration, KestraClient

  While 'kestra' is the PyPI package, the internal SDK modules are exposed under 'kestrapy'. For programmatic client configuration, use 'kestrapy'.

Quickstart

This quickstart demonstrates how to programmatically create and execute a Kestra flow using the Kestra Python SDK. It configures the client using environment variables for host, username, and password (or API token), then defines a simple flow in YAML and submits it to the Kestra instance. It then triggers an execution and waits for its completion. For in-flow interactions, the `kestra.Kestra` class is used within Python script tasks.

import os
from kestrapy import Configuration, KestraClient

# Configure client using environment variables or hardcoded values (not recommended for production)
configuration = Configuration(
    host=os.environ.get('KESTRA_HOST', 'http://localhost:8080'),
    username=os.environ.get('KESTRA_USERNAME', 'root@root.com'), # or KESTRA_API_TOKEN
    password=os.environ.get('KESTRA_PASSWORD', 'Root!1234')
)
kestra_client = KestraClient(configuration)

tenant_id = os.environ.get('KESTRA_TENANT', 'main') # 'main' is default for OSS
namespace = "my_namespace"
flow_id = "my_flow"

flow_yaml = f'''
id: {flow_id}
namespace: {namespace}
tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "Hello from a Kestra Python SDK created flow!"
'''

try:
    # Create a flow
    created_flow = kestra_client.flows.create_flow(tenant=tenant_id, body=flow_yaml)
    print(f"Flow created: {created_flow.id}")

    # Execute the flow
    execution = kestra_client.executions.create_execution(tenant=tenant_id, flow_id=flow_id, namespace=namespace, wait=True)
    print(f"Execution finished with status: {execution.state.current}")
    
    # Example of using Kestra within a flow (not directly runnable in quickstart, but demonstrates usage)
    # from kestra import Kestra
    # Kestra.outputs({"message": "This is an output from within a flow!"})

except Exception as e:
    print(f"An error occurred: {e}")