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.5.tar.gz (36.7 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.5-py3-none-any.whl (42.7 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: loom_core-0.1.5.tar.gz
  • Upload date:
  • Size: 36.7 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.5.tar.gz
Algorithm Hash digest
SHA256 a3402fd5db4b8cba0b8682db34c94205d1503789aca35a4735c19642a91d5fd9
MD5 4d79f80543292996b387855861ac5267
BLAKE2b-256 e7b3289b2f95bfc2bcf3d88617fe61ad165a76d6dc0071c3cd5940584bf34722

See more details on using hashes here.

File details

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

File metadata

  • Download URL: loom_core-0.1.5-py3-none-any.whl
  • Upload date:
  • Size: 42.7 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.5-py3-none-any.whl
Algorithm Hash digest
SHA256 3b3d43cde326abbe0e2670a074d0dfef87af158f5de05d71e0ea432b70cef58f
MD5 069e926242b0949708f1c71f9f231494
BLAKE2b-256 c007060c1dac103e39323df2210c3ad71884143e4436063f4bedd5f55aa3a20d

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