Skip to main content

Durable workflow orchestration engine for Python

Project description

Loom - Durable Workflow Orchestration

Python 3.12+ License: MIT PyPI

A Python-based durable workflow orchestration engine inspired by Temporal and Durable Task Framework. Loom provides event-sourced, deterministic workflow execution with automatic recovery and replay capabilities.

Features

  • Event Sourcing: All workflow state changes persisted as immutable events
  • Deterministic Replay: Workflows reconstruct from event history for recovery
  • Type Safe: Full generic typing support with Workflow[InputT, StateT]
  • Async First: Built on asyncio for high-performance concurrent execution
  • Durable Execution: Workflows survive process crashes and auto-recover
  • Beautiful CLI: Rich console interface with progress tracking
  • Well Tested: Comprehensive test suite with pytest

Quick Start

Installation

pip install loom-core

Or install from source:

git clone https://github.com/yourusername/loom.git
cd loom
pip install -e .

Define a Workflow

import asyncio
from typing import TypedDict
import loom


# Define your data types
class OrderInput(TypedDict):
    order_id: str
    customer_email: str


class OrderState(TypedDict):
    payment_confirmed: bool
    email_sent: bool


# Define activities (side effects)
@loom.activity(name="process_payment", retry_count=3, timeout_seconds=30)
async def process_payment(order_id: str) -> bool:
    # Call payment API
    return True


@loom.activity(name="send_email", retry_count=2)
async def send_confirmation_email(email: str, order_id: str) -> None:
    # Send email via service
    pass


# Define workflow
@loom.workflow(name="OrderProcessing", version="1.0.0")
class OrderWorkflow(loom.Workflow[OrderInput, OrderState]):
    
    @loom.step(name="process_payment")
    async def payment_step(self, ctx: loom.WorkflowContext[OrderInput, OrderState]):
        success = await ctx.activity(process_payment, ctx.input["order_id"])
        await ctx.state.set("payment_confirmed", success)
        ctx.logger.info(f"Payment processed: {success}")
    
    @loom.step(name="send_confirmation")
    async def notification_step(self, ctx: loom.WorkflowContext[OrderInput, OrderState]):
        if ctx.state["payment_confirmed"]:
            await ctx.activity(
                send_confirmation_email,
                ctx.input["customer_email"],
                ctx.input["order_id"]
            )
            await ctx.state.set("email_sent", True)
            ctx.logger.info("Confirmation email sent")

Note: For state updates, use:

  • await ctx.state.set("key", value) for single values
  • await ctx.state.update(key=lambda _: asyncio.sleep(0, value)) for batch updates (requires awaitable)

See STATE_MANAGEMENT.md for detailed examples.

Start a Workflow

async def main():
    db = loom.Database()
    async with db:
        # Initialize database
        await db.migrate_up()
        
        # Start workflow
        handle = await db.start_workflow(
            OrderWorkflow,
            workflow_input=OrderInput(
                order_id="ORD-12345",
                customer_email="customer@example.com"
            ),
            initial_state=OrderState(
                payment_confirmed=False,
                email_sent=False
            ),
        )
        
        print(f"Workflow started: {handle.workflow_id}")
        
        # Execute workflow tasks
        while True:
            task_executed = await loom.run_once()
            if not task_executed:
                break


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

Run the Worker

# Initialize database
loom init

# Start worker with 4 concurrent task processors
loom worker

# Custom configuration
loom worker --workers 8 --poll-interval 1.0

CLI Commands

# Initialize database
loom init

# Start distributed worker
loom worker [--workers 4] [--poll-interval 0.5]

# List workflows
loom list [--limit 50] [--status RUNNING]

# Inspect workflow details
loom inspect <workflow-id> [--events]

# Show database statistics
loom stats

🏗️ Architecture

Core Components

Architecture

Event Types

  • WORKFLOW_STARTED - Workflow initialization
  • WORKFLOW_COMPLETED - Successful completion
  • WORKFLOW_FAILED - Fatal error occurred
  • STATE_SET - Single state key updated
  • STATE_UPDATE - Batch state update
  • ACTIVITY_SCHEDULED - Activity queued for execution
  • ACTIVITY_COMPLETED - Activity finished successfully
  • ACTIVITY_FAILED - Activity permanently failed
  • TIMER_FIRED - Sleep/delay completed
  • SIGNAL_RECEIVED - External signal received

📚 Documentation

See .copilot-instructions.md for comprehensive development guidelines including:

  • Event sourcing patterns
  • Deterministic execution rules
  • Activity best practices
  • Testing strategies
  • Common pitfalls to avoid

🧪 Testing

# Run all tests
pytest

# Run with coverage
pytest --cov=src --cov-report=html

# Run specific test file
pytest tests/test_workflow.py

# Verbose output
pytest -v

Project Structure

loom/
├── src/
│   ├── common/         # Shared utilities
│   ├── core/           # Core engine (context, engine, runner, worker)
│   ├── database/       # Database layer
│   ├── decorators/     # @workflow, @step, @activity
│   ├── lib/            # Utilities and progress tracking
│   ├── migrations/     # Database migrations
│   └── schemas/        # Type definitions
├── tests/              # Test suite
├── examples/           # Example workflows
├── loom.py             # Main package interface
└── pyproject.toml      # Package configuration

Configuration

Loom uses SQLite by default for simplicity. For production:

  • Consider PostgreSQL/MySQL for scalability
  • Implement connection pooling
  • Add monitoring and alerting
  • Deploy multiple workers for high availability

Contributing

Contributions welcome! Please ensure:

  1. Tests pass: pytest
  2. Code formatted: black .
  3. Type checking: mypy .
  4. Linting: ruff check .

📝 License

MIT License - see LICENSE file for details

🙏 Acknowledgments

Inspired by:

📧 Contact

For questions and support, please open an issue on GitHub.


Built with ❤️ using Python 3.12+

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

loom_core-0.1.3.tar.gz (36.1 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

loom_core-0.1.3-py3-none-any.whl (42.0 kB view details)

Uploaded Python 3

File details

Details for the file loom_core-0.1.3.tar.gz.

File metadata

  • Download URL: loom_core-0.1.3.tar.gz
  • Upload date:
  • Size: 36.1 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.9

File hashes

Hashes for loom_core-0.1.3.tar.gz
Algorithm Hash digest
SHA256 1998062934f2c315643ec6b04e0bd97a7c4aa863499b06d681fc5f20439ced57
MD5 05263bdac31fc91360c1eeffa9bf744c
BLAKE2b-256 3744146ad908ae99cdd9a0f2848edf367f9334923a11df003ec8132921aab262

See more details on using hashes here.

File details

Details for the file loom_core-0.1.3-py3-none-any.whl.

File metadata

  • Download URL: loom_core-0.1.3-py3-none-any.whl
  • Upload date:
  • Size: 42.0 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.9

File hashes

Hashes for loom_core-0.1.3-py3-none-any.whl
Algorithm Hash digest
SHA256 5f6d2daabf2f76704a88b28c7f5530e65e9bb8c17b0143e0113f66c62ec1bdd7
MD5 e181a652cac3366b1bc384fc3e3114ef
BLAKE2b-256 9f9e24711401470c286b3770d8f27241a90256f183909137e0baeb51555df93f

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page