simple-saga 0.0.1

A lightweight implementation of the Saga pattern for managing distributed transactions in Python

Simple Saga

A lightweight implementation of the Saga pattern for managing distributed transactions in Python.

Overview

The Saga pattern breaks down distributed transactions into a series of local transactions, each with a compensating transaction that can undo the changes if a later step fails. This library provides a simple, type-safe implementation with support for both synchronous and asynchronous operations.

Features

• ✅ Simple API - Easy to use with a fluent interface
• 🔄 Automatic Compensation - Failed transactions are automatically rolled back
• ⚡ Sync & Async Support - Works with both synchronous and asynchronous functions
• 🔒 Type Safe - Full type hints with mypy support
• 🪶 Lightweight - Zero dependencies (uses only Python standard library)
• 📚 Well Documented - Comprehensive docstrings and examples

Installation

pip install simple-saga

Or with Poetry:

poetry add simple-saga

Quick Start

import asyncio
from simple_saga import SimpleSaga

# Define your actions
def create_order(order_id: str) -> dict:
    print(f"Creating order: {order_id}")
    return {"order_id": order_id, "status": "created"}

def cancel_order(order_result: dict) -> None:
    print(f"Cancelling order: {order_result['order_id']}")

async def reserve_inventory(product_id: str) -> dict:
    print(f"Reserving inventory for: {product_id}")
    return {"product_id": product_id, "reserved": True}

async def release_inventory(inventory_result: dict) -> None:
    print(f"Releasing inventory for: {inventory_result['product_id']}")

def charge_payment(amount: float) -> dict:
    print(f"Charging payment: ${amount}")
    # Simulating a payment failure
    raise Exception("Payment failed")

def refund_payment(payment_result: dict) -> None:
    print("Refunding payment")

# Create and execute saga
async def main():
    saga = SimpleSaga()

    # Add steps with actions and compensations
    saga.add_step(
        action=create_order,
        compensation=cancel_order,
        action_args=("ORDER-123",)
    )

    saga.add_step(
        action=reserve_inventory,
        compensation=release_inventory,
        action_args=("PRODUCT-456",)
    )

    saga.add_step(
        action=charge_payment,
        compensation=refund_payment,
        action_args=(99.99,)
    )

    try:
        results = await saga.execute()
        print("✅ Saga completed successfully!")
        for result in results:
            print(f"  Step {result.step_index + 1}: {result.result}")
    except Exception as e:
        print(f"❌ Saga failed: {e}")
        print("All completed steps have been compensated.")

if __name__ == "__main__":
    asyncio.run(main())

Output

Creating order: ORDER-123
✓ Step 1 completed: create_order
Reserving inventory for: PRODUCT-456
✓ Step 2 completed: reserve_inventory
Charging payment: $99.99
✗ Error at step 3: Payment failed
🔄 Starting compensation...
Releasing inventory for: PRODUCT-456
✓ Compensated step 2: release_inventory
Cancelling order: ORDER-123
✓ Compensated step 1: cancel_order
❌ Saga failed: Payment failed
All completed steps have been compensated.

Advanced Usage

Passing Multiple Arguments

saga.add_step(
    action=create_user,
    compensation=delete_user,
    action_args=("john_doe",),
    action_kwargs={"email": "[e-mail]", "age": 30}
)

Custom Compensation Arguments

By default, the action's result is passed to the compensation function. You can override this:

saga.add_step(
    action=allocate_resource,
    compensation=deallocate_resource,
    action_args=("resource-id",),
    compensation_args=("resource-id",),
    compensation_kwargs={"force": True}
)

Method Chaining

saga = (
    SimpleSaga()
    .add_step(action=step1, compensation=undo_step1)
    .add_step(action=step2, compensation=undo_step2)
    .add_step(action=step3, compensation=undo_step3)
)

results = await saga.execute()

Reusing Saga Definitions

saga = SimpleSaga()
saga.add_step(action=step1, compensation=undo_step1)
saga.add_step(action=step2, compensation=undo_step2)

# Execute multiple times
await saga.execute()  # First execution
await saga.execute()  # Second execution (automatically resets)

Logging Control

The library uses Python's standard logging module:

import logging

# Configure logging
logging.basicConfig(level=logging.INFO)

# Or disable saga logs
logging.getLogger("simple_saga").setLevel(logging.WARNING)

API Reference

SimpleSaga

Main class for defining and executing sagas.

add_step(action, compensation, *, action_args=(), action_kwargs=None, compensation_args=(), compensation_kwargs=None)

Add a step to the saga.

Parameters:

• action: Function to execute (can be sync or async)
• compensation: Function to compensate if this or later steps fail (can be sync or async)
• action_args: Positional arguments for the action
• action_kwargs: Keyword arguments for the action
• compensation_args: Positional arguments for the compensation
• compensation_kwargs: Keyword arguments for the compensation

Returns: Self (for method chaining)

async execute() -> list[StepResult]

Execute all steps in the saga. If any step fails, automatically runs compensation for all previously executed steps in reverse order.

Returns: List of StepResult objects

Raises: Re-raises the exception that caused the saga to fail

reset() -> None

Reset the saga state, clearing all executed steps. Called automatically by execute().

StepResult

Dataclass containing the result of a saga step execution.

Attributes:

• step_index: int - The index of the step
• step_name: str - The name of the action function
• result: Any - The result returned by the action

SagaStep

Dataclass representing a single step in the saga with action and compensation.

Attributes:

• action: Callable - The action function
• compensation: Callable - The compensation function
• action_args: tuple - Positional arguments for the action
• action_kwargs: dict - Keyword arguments for the action
• compensation_args: tuple - Positional arguments for the compensation
• compensation_kwargs: dict - Keyword arguments for the compensation

Design Decisions

Why async execute()?

Even though the library supports synchronous functions, execute() is async to handle mixed sync/async steps uniformly. This allows you to:

• Mix sync and async functions in the same saga
• Use async patterns for I/O-bound operations
• Keep the API simple and consistent

Compensation Behavior

• Compensations run in reverse order (LIFO)
• Compensation failures are logged but don't stop the chain
• Each compensation receives the action's result by default
• You can override compensation arguments if needed

Development

Setup

# Clone the repository
git clone https://github.com/yourusername/simple-saga.git
cd simple-saga

# Install dependencies
poetry install

# Run type checking
poetry run mypy simple_saga

# Run linting
poetry run ruff check simple_saga

Project Structure

simple-saga/
├── simple_saga/
│   ├── __init__.py      # Package exports
│   ├── saga.py          # Main SimpleSaga implementation
│   └── schema.py        # Data classes (StepResult, SagaStep)
├── pyproject.toml       # Project configuration
├── README.md            # This file
└── CLAUDE.md           # Development guide

License

MIT License - see LICENSE file for details

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Acknowledgments

This library implements the Saga pattern as described in:

• "Sagas" by Hector Garcia-Molina and Kenneth Salem (1987)
• Microservices Patterns by Chris Richardson