absurd 0.0.4

A Python client for the Absurd SQL-based durable execution workflow system

Absurd Python Client

A Python client for the Absurd SQL-based durable execution workflow system. This library provides a Python interface to interact with Absurd's PostgreSQL-based task queue and workflow engine.

About Absurd

Absurd is a SQL workflow system that implements durable execution using PostgreSQL as the backend. It provides features such as:

• Task queuing and processing
• Durable task execution that survives crashes
• Event-driven workflow coordination
• Checkpoint-based state management
• Retry strategies and failure handling
• Workflow run tracking

Installation

pip install absurd

Requirements

• Python 3.8+
• PostgreSQL with the Absurd extension installed

Quick Start

import psycopg
from absurd_client import AbsurdClient

# Create a client instance
client = AbsurdClient(queue_name="my_queue")

# Connect to your PostgreSQL database
with psycopg.connect("your_connection_string") as conn:
    # Spawn a new task
    task_id, run_id, workflow_run_id = client.spawn_task(
        conn=conn,
        task_name="process_data",
        params={"input": "data"},
        options={"headers": {"priority": "high"}}
    )
    
    print(f"Spawned task: {task_id}, run: {run_id}, workflow: {workflow_run_id}")
    
    # Claim and process tasks
    claimed_tasks = client.claim_task(conn, qty=1)
    for task_data in claimed_tasks:
        run_id, task_id, attempt, task_name, params, *_ = task_data
        
        try:
            # Process the task
            result = process_task(task_name, params)
            
            # Mark as completed
            client.complete_task(conn, run_id, result)
        except Exception as e:
            # Mark as failed
            client.fail_task(conn, run_id, str(e))

Features

Task Management

• spawn_task(): Create new tasks with parameters, options, headers, retry strategies
• claim_task(): Claim tasks for processing
• complete_task(): Mark tasks as completed
• fail_task(): Mark tasks as failed with error information
• cancel_task(): Cancel pending or sleeping tasks

Event Handling

• await_event(): Wait for specific events
• emit_event(): Emit events to wake sleeping tasks

Checkpoint Management

• set_checkpoint(): Set checkpoints for long-running tasks
• get_checkpoint(): Retrieve checkpoint data

Workflow Tracking

• create_workflow_run(): Track workflow execution
• update_workflow_run_status(): Update workflow status

Helper Functions

• spawn_retry_task(): Convenience function for tasks with retry strategies
• spawn_cancellable_task(): Convenience function for tasks with cancellation rules
• get_absurd_client(): Singleton client instance

Connection Management

The client works with psycopg3 connections. It's recommended to use connection pooling for production applications:

from psycopg_pool import ConnectionPool
from absurd_client import AbsurdClient

pool = ConnectionPool("your_connection_string")
client = AbsurdClient(queue_name="my_queue")

with pool.connection() as conn:
    task_id, run_id, workflow_run_id = client.spawn_task(
        conn=conn,
        task_name="example_task",
        params={"data": "value"}
    )

Error Handling

The client includes proper error handling for various scenarios:

• AbsurdSleepError: Raised when a run enters SLEEPING state waiting for an event
• Standard exception handling for database operations

License

This project is licensed under the terms specified in the LICENSE file.

Contributing

Contributions are welcome! Please open an issue or submit a pull request on GitHub.

Support

For support, please open an issue in the GitHub repository.