lexigram-workflow

Workflow orchestration for the Lexigram Framework (pipelines, bulk ops, sagas, graph engine)

Overview

lexigram-workflow provides workflow orchestration, state machines, and saga pattern for modeling complex, long-running business processes. It supports durable persistence of transition history, optimistic locking, multi-level approval chains, distributed transaction coordination with automatic rollback, and a graph engine for traversing directed graphs. All services are wired via WorkflowProvider, which registers workflow protocols with the DI container.

Full documentation: docs.lexigram.dev

Install

uv add lexigram-workflow

Quick Start

from lexigram import Application
from lexigram.di.module import Module, module

# Import the module from the package
from lexigram.workflow import WorkflowModule


@module(imports=[WorkflowModule.configure()])
class AppModule(Module):
    pass


async with Application.boot(modules=[AppModule]) as app:
    # use app.container to resolve services
    ...

Configuration

Zero-config usage: Call WorkflowModule.configure() with no arguments to use defaults.

Option 1 — YAML file

# application.yaml
workflow:
  batch_size: 10
  max_concurrency: 5
  timeout: 300.0
  retry_attempts: 3
  enable_progress_tracking: true
  pipeline_timeout: 60.0

Option 2 — Profiles + Environment Variables (recommended)

export LEX_WORKFLOW__ENABLED=true
# Environment variables for each field

Option 3 — Python

from lexigram.workflow.config import BulkOperationConfig
from lexigram.workflow import WorkflowModule

config = BulkOperationConfig(batch_size=10, max_concurrency=5, retry_attempts=3)
WorkflowModule.configure(config=config)

Config reference

Content-addressed checkpointing is configured via a standalone ContentCheckpointConfig passed to WorkflowModule.configure(..., content_checkpoint_config=...) — it is not a field of BulkOperationConfig and is not read from environment variables.

Module Factory Methods

Key Features

• State machine — Declarative states and transitions with entry/exit hooks
• Durable persistence — Transition history persisted to DB via StatePersistenceProtocol
• Optimistic locking — Prevents concurrent transition conflicts
• State recovery — Rebuild machine state from persisted history on restart
• Approval chains — Multi-level approval flows with role and threshold rules
• Sagas — Distributed transaction coordination with automatic rollback
• Content-addressed sagas — Idempotent stage caching keyed by sha256(stage_id, tenant_id, inputs, handler_version, config); skips already-completed work on resume
• Pipeline checkpointing — Content-addressed checkpoint stores (InMemory, Cache, Database) for durable saga resume
• Pipelines — Step-based sequential pipelines with error handling
• Bulk operations — Apply an operation to many entities in a supervised batch
• Guard conditions — Transition guards as async def can_confirm(self) -> bool
• Event hooks — on_enter_*, on_exit_*, on_transition lifecycle callbacks

Testing

async with Application.boot(modules=[WorkflowModule.stub()]) as app:
    # your test code
    ...

Key Source Files