kestra 1.2.0

Kestra is an infinitely scalable orchestration and scheduling platform, creating, running, scheduling, and monitoring millions of complex pipelines.

Kestra Python Client

This Python client provides functionality to interact with the Kestra server for sending metrics, outputs, and logs, as well as executing/polling flows.

Installation

pip install kestra

Kestra Class

The Kestra class is responsible for sending metrics, outputs, and logs to the Kestra server.

Methods

• _send(map_: dict): Sends a message to the Kestra server.
• format(map_: dict) -> str: Formats a message to be sent to the Kestra server.
• _metrics(name: str, type_: str, value: int, tags: dict | None = None): Sends a metric to the Kestra server.
• outputs(map_: dict): Sends outputs to the Kestra server.
• counter(name: str, value: int, tags: dict | None = None): Sends a counter to the Kestra server.
• timer(name: str, duration: int | Callable, tags: dict | None = None): Sends a timer to the Kestra server.
• logger() -> Logger: Retrieves the logger for the Kestra server.

Flow Class

The Flow class is used to execute a Kestra flow and optionally wait for its completion. It can also be used to get the status of an execution and the logs of an execution.

Initialization

flow = Flow(
  wait_for_completion=True, # default is True
  poll_interval=1, # seconds. default is 1  
  labels_from_inputs=False, # default is False
  tenant=None # default is None
)

You can also set the hostname and authentication credentials using environment variables:

export KESTRA_HOSTNAME=http://localhost:8080
export KESTRA_USER=admin
export KESTRA_PASSWORD=admin
export KESTRA_API_TOKEN=my_api_token

It is worth noting that the KESTRA_API_TOKEN or KESTRA_USER and KESTRA_PASSWORD need to be used, you do not need all at once. The possible Authentication patterns are:

1. KESTRA_API_TOKEN
2. KESTRA_USER and KESTRA_PASSWORD
3. No Authentication (not recommended for production environments)

Methods

• _make_request(method: str, url: str, **kwargs) -> requests.Response: Makes a request to the Kestra server with optional authentication and retries.
• check_status(execution_id: str) -> requests.Response: Checks the status of an execution.
• get_logs(execution_id: str) -> requests.Response: Retrieves the logs of an execution.
• execute(namespace: str, flow: str, inputs: dict = None) -> namedtuple: Executes a Kestra flow and optionally waits for its completion. The namedtuple returned is a namedtuple with the following properties:
  • status: The status of the execution.
  • log: The log of the execution.
  • error: The error of the execution.

Usage Examples

1. Trigger a flow and wait for its completion:

   from kestra import Flow
   flow = Flow()
   flow.execute('mynamespace', 'myflow', {'param': 'value'})
   

2. Set labels from inputs:

   from kestra import Flow
   flow = Flow(labels_from_inputs=True)
   flow.execute('mynamespace', 'myflow', {'param': 'value'})
   

3. Pass a text file to an input of type FILE named 'myfile':

   from kestra import Flow
   flow = Flow()
   with open('example.txt', 'rb') as fh:
       flow.execute('mynamespace', 'myflow', {'files': ('myfile', fh, 'text/plain')})
   

4. Fire and forget:

   from kestra import Flow
   flow = Flow(wait_for_completion=False)
   flow.execute('mynamespace', 'myflow', {'param': 'value'})
   

5. Overwrite the username and password:

   from kestra import Flow
   flow = Flow()
   flow.user = 'admin'
   flow.password = 'admin'
   flow.execute('mynamespace', 'myflow')
   

6. Set the hostname, username, and password using environment variables:

   from kestra import Flow
   import os
   
   os.environ["KESTRA_HOSTNAME"] = "http://localhost:8080"
   os.environ["KESTRA_USER"] = "admin"
   os.environ["KESTRA_PASSWORD"] = "admin"
   flow = Flow()
   flow.execute('mynamespace', 'myflow', {'param': 'value'})
   

Error Handling

The client includes retry logic with exponential backoff for certain HTTP status codes, and raises a FailedExponentialBackoff exception if the request fails after multiple retries.

Kestra Class

Logging

The Kestra class provides a logger that formats logs in JSON format, making it easier to integrate with log management systems.

from kestra import Kestra

Kestra.logger().info("Hello, world!")

Outputs

The Kestra class provides a method to send key-value-based outputs to the Kestra server. If you want to output large objects, write them to a file and specify them within the outputFiles property of the Python script task.

Kestra.outputs({"my_output": "my_value"})

Counters

The Kestra class provides a method to send counter metrics to the Kestra server.

Kestra.counter("my_counter", 1)

Timers

The Kestra class provides a method to send timer metrics to the Kestra server.

Kestra.timer("my_timer", 1)

Gauges

The Kestra class provides a method to send gauge metrics to the Kestra server.

Kestra.gauge("my_gauge", 42.5)

Kestra Ion

The Kestra ION extra provides a method to read files and convert them to a list of dictionaries.

Installation

pip install kestra[ion]

Methods

• read(path_: str) -> list[dict[str, Any]]: Reads an Ion file and converts it to a list of dictionaries.

Usage Example

import pandas as pd
import requests
from kestra import Kestra

file_path = "employees.ion"
url = "https://huggingface.co/datasets/kestra/datasets/resolve/main/ion/employees.ion"
response = requests.get(url)
if response.status_code == 200:
    with open(file_path, "wb") as file:
        file.write(response.content)
else:
    print(f"Failed to download the file. Status code: {response.status_code}")


data = Kestra.read(file_path)
df = pd.DataFrame(data)
print(df.info())