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-1.2.0.tar.gz (59.6 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-1.2.0-py3-none-any.whl (73.3 kB view details)

Uploaded Python 3

File details

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

File metadata

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

File hashes

Hashes for loom_core-1.2.0.tar.gz
Algorithm Hash digest
SHA256 0592bffe1b13e4f975f29758f4a684e9411d7e2b72aa0084010250f6e3746e50
MD5 9476d25d3f777e832a5355db28ced1af
BLAKE2b-256 1d79666b0bad9a1387f9f2963d9dce3ee45d921ca714474b79cad82c1520f6da

See more details on using hashes here.

File details

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

File metadata

  • Download URL: loom_core-1.2.0-py3-none-any.whl
  • Upload date:
  • Size: 73.3 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-1.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 e66699e39b05c5c07bf1333c26d74b7abb70ebb4c6ee167210ef55d17f5d1dc5
MD5 70a9f9e55a624e619482897043030af6
BLAKE2b-256 66977378ff4a53e214a740d20acefb4883cb57abaaaebc06c7784d6475438cb8

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