business-workflow-manager

A reusable workflow engine for Python applications backed by SQLAlchemy ORM models. Define multi-step workflows in YAML, load them into a relational database, and drive execution at runtime with condition-based branching.

Features

YAML-based workflow definitions — version-controlled, human-readable workflow configurations
SQLAlchemy ORM models — ready-to-use Single Table Inheritance model hierarchy
Runtime engine — step progression with jmespath condition evaluation
Multilingual — built-in support for DE/FR/IT translated strings
GraphViz export — generate DOT diagrams from workflow configs
Session-friendly — works with your existing SQLAlchemy session; never commits

Installation

pip install business-workflow-manager

Or with uv:

uv add business-workflow-manager

Dependencies

Python ≥ 3.11
SQLAlchemy ~= 2.0
Pydantic ≥ 2.0
PyYAML
jmespath

Quick Start

1. Define a Workflow in YAML

workflow:
  title:
    de: Bewilligungsverfahren
    fr: Procédure d'autorisation
    it: Procedura di autorizzazione
  version: 1
  start_task_ref: review
  min_per_entity: 0
  max_per_entity: 1
  tasks:
    review:
      title: Review
      start_task: true
      steps:
        - type: form
          title: Assessment
          name: assessment
          fields:
            - name: approved
              type: boolean
              label: Approved?
        - type: document
          title: Upload Decision
      links:
        - task_ref: finalize
          condition: "approved == `true`"
        - task_ref: reject
          condition: "approved == `false`"
    finalize:
      title: Finalize
      steps:
        - type: document
          title: Final Document
      links:
        - task_ref: END
    reject:
      title: Reject
      steps:
        - type: document
          title: Rejection Letter
      links:
        - task_ref: END
    END:
      title: __END__
      steps: []
      links: []

2. Set Up the Database

The library ships its own DeclarativeBase. Create its tables alongside your application tables:

from sqlalchemy import create_engine
from business_workflow_manager.models import Base

engine = create_engine("postgresql://localhost/mydb")

# Create workflow tables (wf_config, wf_node, wf_link, wf_step, translations)
Base.metadata.create_all(engine)

3. Load a Workflow into the Database

from sqlalchemy.orm import Session
from business_workflow_manager import WorkflowManager

with Session(engine) as session:
    with WorkflowManager(session) as mgr:
        with open("workflows/review.yaml") as f:
            loaded = mgr.load_workflow_from_file(f)  # loaded.workflow, loaded.config
        session.commit()  # The library flushes but never commits — you control transactions

4. Start a Workflow Instance

from business_workflow_manager import WorkflowManager

with Session(engine) as session:
    with WorkflowManager(session) as mgr:

    # Create a workflow instance for entity "42"
    # (checks applicability, raises if max_per_entity exceeded)
    wf_node = mgr.start_workflow(workflow.wf_config_id, entity_id="42")
    # => WorkflowNodeInfo(wf_node_id=1, status=STARTED, ...)

    # Get available first steps (start tasks)
    next_steps = mgr.get_next_steps(wf_node.wf_node_id).available_options
    # => [StepOption(wf_config_id=5, name='review', type=TASK, ...)]

    # Start the first task
    task_node = mgr.start_next_step(wf_node.wf_node_id, next_steps[0].wf_config_id)
    # => TaskNodeInfo(wf_node_id=2, status=STARTED, ...)

    # Get the task's steps
    steps = mgr.get_next_steps(task_node.wf_node_id).available_options
    # => [StepOption(wf_config_id=8, name='assessment', type=FORM, ...)]

    # Start the form step
    form_node = mgr.start_next_step(task_node.wf_node_id, steps[0].wf_config_id)
    # => FormNodeInfo(wf_node_id=3, status=STARTED, ...)

    session.commit()

All public API methods return frozen dataclasses (NodeInfo subclasses, StepInfo and StepOption), not mutable ORM instances. Methods accept plain integer IDs (node_id, step_id) as arguments.

5. Conditional Branching

After completing all steps in a task, the engine evaluates link conditions to determine which task(s) come next:

    # After finishing all steps in the "review" task, get next tasks:
    next_tasks = mgr.get_next_steps(task_node.wf_node_id).available_options
    # If approved==True: => [StepOption(name='finalize', ...)]
    # If approved==False: => [StepOption(name='reject', ...)]

Conditions use jmespath expressions evaluated against node.data. Use backtick-quoted JSON literals for comparison values (`true`, `false`, `"string"`).

6. Updating a Workflow Node

from datetime import datetime
from business_workflow_manager import WorkflowManager, CascadeConflict, NodeStatus

with Session(engine) as session:
    mgr = WorkflowManager(session)

    # Update fields on a workflow node
    updated = mgr.update_workflow_node(
        wf_node.wf_node_id,
        title="Renamed workflow",
        deadline=datetime(2026, 6, 1),
        status=NodeStatus.FINISHED,
        cascade=False,  # default: raise if children conflict
    )

    # If children have later deadlines or are still open, CascadeConflict is raised:
    try:
        mgr.update_workflow_node(node_id, status=NodeStatus.FINISHED)
    except CascadeConflict as e:
        print(f"Conflicting children: {e.status_conflicts}")
        # Ask user to confirm, then retry with cascade=True
        mgr.update_workflow_node(node_id, status=NodeStatus.FINISHED, cascade=True)

    session.commit()

Status transitions have side effects:

STARTED: clears finished_at
INACTIVE: clears both deadline and finished_at

update_task_node works the same way but cascades upward — it checks parent/grandparent nodes:

If the task's deadline exceeds a parent's deadline → conflict
If setting the task to non-finished while a parent is already finished → conflict

With cascade=True, parent deadlines are extended and finished parents are reopened.

7. Creating Ad-hoc Documents

Documents can be created independently of a workflow step definition (no config required):

from datetime import datetime
from business_workflow_manager import WorkflowManager

with Session(engine) as session:
    mgr = WorkflowManager(session)

    doc = mgr.create_document_node(
        entity_id="42",
        title="Contract signed",
        started_at=datetime(2024, 6, 15),
        document_ref="DOC-2024-001",
        note="Original contract",
        is_public=True,
        parent_id=task_node.wf_node_id,  # optional: attach under a task or workflow
    )
    # => DocumentNodeInfo(wf_node_id=..., status=FINISHED, document_ref="DOC-2024-001")

    session.commit()

Ad-hoc documents are always created with status=FINISHED and all timestamps (started_at, finished_at, deadline) set to the same value. If parent_id is provided, the method walks up from that node to find the nearest TaskNode (preferred) or WorkflowNode ancestor. If no suitable parent is found, the document is created at the top level. This means creation never fails due to parent resolution.

create_note_node works the same way but without a document_ref field:

    note = mgr.create_note_node(
        entity_id="42",
        title="Follow-up required",
        started_at=datetime(2024, 6, 15),
        note="Discuss with legal team",
        parent_id=task_node.wf_node_id,
    )
    # => NoteNodeInfo(wf_node_id=..., status=FINISHED, ...)

8. Event Handling

Define domain events by subclassing Event, register handlers at startup, and dispatch them through the manager:

from sqlalchemy.orm import Session
from business_workflow_manager import Event, WorkflowManagerConfig, WorkflowManager, manager_factory


# Define your events
class InvoiceApproved(Event):
    def __init__(self, invoice_id: int) -> None:
        self.invoice_id = invoice_id


# Define handlers — receive the session and event
def notify_accounting(session: Session, event: Event) -> None:
    assert isinstance(event, InvoiceApproved)
    # ... use session to query/update the database ...


# Configure once at startup
def get_entity_data(session, entity_id):
    # Fetch and return a JSON-serializable dict for the entity
    return {"id": entity_id, "name": "Example"}

config = WorkflowManagerConfig(get_entity_data=get_entity_data)

# Option A: pass config directly
with WorkflowManager(session, config) as mgr:
    mgr.register_event_handler(InvoiceApproved, notify_accounting)
    mgr.handle_event(InvoiceApproved(invoice_id=123))

# Option B: use a factory (bind config once, create managers with just a session)
new_manager = manager_factory(config)
with new_manager(session) as mgr:
    mgr.register_event_handler(InvoiceApproved, notify_accounting)
    mgr.handle_event(InvoiceApproved(invoice_id=123))

Handlers are called in registration order with the active session and event instance.

API Reference

WorkflowManager

The primary API is the WorkflowManager class, which wraps a SQLAlchemy Session. Use it as a context manager to ensure the session reference is released:

from business_workflow_manager import WorkflowManager

with WorkflowManager(session) as mgr:
    ...

Method	Description
`load_workflow_from_file(f)`	Parse YAML file and load into DB, keeping edited translations. Returns `LoadedWorkflow` (`.workflow`, `.config`).
`load_workflow(config)`	Load a parsed `Workflow` config into DB, keeping edited translations. Returns the ORM `Workflow`.
`load_workflows_from_selection(selection, base_dir)`	Load all workflow files listed in a plain-text selection stream into DB, keeping edited translations. Returns `list[tuple[Path, LoadedWorkflow]]` in file order.
`dump_workflow(workflow)`	Export an ORM `Workflow` back to a pydantic config object.
`dump_workflow_to_file(key, f)`	Export a workflow (by UUID key) to YAML file.
`start_workflow(workflow_id, entity_id)`	Create a new workflow instance. Returns `WorkflowNodeInfo`.
`update_workflow_node(node_id, *, ...)`	Update a workflow node with business-rule enforcement. Returns `NodeInfo`.
`create_task_node(entity_id, title, started_at, *, ...)`	Create an ad-hoc task node (status=STARTED). Returns `TaskNodeInfo`.
`update_task_node(node_id, *, ...)`	Update a task node with business-rule enforcement. Returns `NodeInfo`.
`update_form_node(node_id, *, ...)`	Update a form node (propagates answers to node.data, sets FINISHED). Returns `FormNodeInfo`.
`create_document_node(entity_id, title, started_at, document_ref, *, ...)`	Create an ad-hoc document node. Returns `DocumentNodeInfo`.
`update_document_node(node_id, *, ...)`	Update a document node (always sets status to FINISHED). Returns `DocumentNodeInfo`.
`create_note_node(entity_id, title, started_at, *, ...)`	Create an ad-hoc note node. Returns `NoteNodeInfo`.
`update_note_node(node_id, *, ...)`	Update a note node (always sets status to FINISHED). Returns `NoteNodeInfo`.
`delete_node(node_id)`	Delete a node (fails if it has children or is readonly).
`get_next_steps(node_id)`	Get valid next steps for the node. Returns `list[StepOption]`.
`start_next_step(node_id, step_id)`	Create and link the next execution node. Returns `NodeInfo`.
`handle_event(event)`	Dispatch an event to all registered handlers.
`register_event_handler(event_type, handler)`	Register a handler function to be called when an event of `event_type` is dispatched.

Factory Function

from business_workflow_manager import manager_factory, WorkflowManagerConfig

config = WorkflowManagerConfig()
new_manager = manager_factory(config)

with new_manager(session) as mgr:
    mgr.register_event_handler(InvoiceApproved, notify_accounting)
    ...

Function	Description
`manager_factory(config)`	Returns a callable `(Session) -> WorkflowManager` with config bound.

Events

Class	Description
`Event`	Base class for user-defined events. Subclass to create concrete event types.
`WorkflowManagerConfig`	Frozen dataclass holding `get_entity_data: Callable[[Session, int], dict]`. Event handlers are registered via `WorkflowManager.register_event_handler`.
`EventHandler`	Type alias: `Callable[[Session, Event], None]`.
`EventHandlerMap`	Type alias: `dict[type[Event], list[EventHandler]]`. Handlers receive matching event instances.

Data Transfer Objects

All public API methods return frozen dataclasses (not ORM instances):

Node info (returned by start_workflow, start_next_step, etc.):

Class	Description
`NodeInfo`	Base class with all common node fields.
`WorkflowNodeInfo`	Workflow node (returned by `start_workflow`).
`TaskNodeInfo`	Task node.
`FormNodeInfo`	Form node (has `form_config`, `form_data`).
`DocumentNodeInfo`	Document node (has `document_ref`).
`NoteNodeInfo`	Note node.

Step options (returned by get_next_steps):

Class	Fields
`StepInfo`	`available_options` (`list[StepOption]`), `unavailable_reasons` (`list[NextStepUnavailableReason]`)
`StepOption`	`wf_config_id`, `title`, `type`, `name`, `is_optional`

unavailable_reasons lists every reason found while determining the next steps (deduplicated, ordered as defined in NextStepUnavailableReason, which is also the display priority). It can be non-empty even if available_options is not, e.g. when a conditional step was skipped or a valid link points to END. It is empty if the node, its parent task or its workflow is SKIPPED or INACTIVE.

`NextStepUnavailableReason`	Meaning
`TASK_CLOSED`	The parent task of a step is `FINISHED`, or the workflow is `FINISHED` before any task was started.
`END_OF_WORKFLOW`	A valid link points to `END`, or the task has no links.
`STEP_NOT_CLOSED`	The current step or started task is not finished yet, or a condition of a step or link was not met (e.g. a form selection is missing).
`TASK_WITHOUT_WORKFLOW`	The node is not part of a configured workflow (e.g. an ad-hoc task or note without a workflow parent).

Use step.wf_config_id when calling start_next_step, and node.wf_node_id when calling methods that take a node_id.

Config (Pydantic)

Class	Description
`Workflow`	Top-level workflow definition with tasks, versions, and limits.
`Task`	A task containing ordered steps and links to other tasks.
`Form`	A form step with named fields.
`Document`	A document upload step.
`TranslatedString`	A string with `de`, `fr`, `it` attributes.

Read/write YAML directly:

from business_workflow_manager.config import Workflow

with open("workflow.yaml") as f:
    config = Workflow.read(f)

with open("output.yaml", "w") as f:
    config.write(f)

GraphViz Export

Generate DOT diagrams for visualization:

from business_workflow_manager.graphviz import write_graph
from business_workflow_manager.types import Language

with open("workflow.dot", "w") as f:
    write_graph(config, f, Language.DE)

Render with GraphViz:

dot -Tpng workflow.dot -o workflow.png

Models

The library provides these ORM models:

Configuration models (table: wf_config, STI):

Model	Description
`WorkflowConfig`	Abstract base for all config items.
`Workflow`	A workflow definition with tasks, start_task, limits.
`Task`	A task with ordered steps and links to successor tasks.
`Form`	A form step definition (fields stored as JSONB).
`Document`	A document step definition.

Execution models (table: wf_node, STI):

Model	Description
`Node`	Abstract base for execution nodes.
`WorkflowNode`	A running workflow instance.
`TaskNode`	A running task within a workflow.
`FormNode`	A form step being filled (has `form_config`, `form_data`).
`DocumentNode`	A document step (has `document_ref`).
`NoteNode`	An ad-hoc note attached to the workflow.

Supporting models:

Model	Table	Description
`WorkflowLink`	`wf_link`	Directed edge between tasks (with optional condition).
`WorkflowStep`	`wf_step`	Ordered step within a task (with optional condition).
`Translation`	`translations`	i18n key-value store (key, locale, value).

Enums

from business_workflow_manager.types import Language, ItemType, NodeType, NodeStatus

Language.DE | Language.FR | Language.IT
ItemType.WORKFLOW | ItemType.TASK | ItemType.FORM | ItemType.DOCUMENT
NodeType.WORKFLOW | NodeType.TASK | NodeType.FORM | NodeType.DOCUMENT | NodeType.NOTE
NodeStatus.STARTED | NodeStatus.FINISHED | NodeStatus.SKIPPED | NodeStatus.INACTIVE

Exceptions

from business_workflow_manager.exceptions import (
    WorkflowException,     # Base exception
    WorkflowNotFound,      # Workflow key not in DB
    WorkflowExists,        # Duplicate workflow
    InvalidWorkflowStep,   # Step not valid for current state
    CascadeConflict,       # Update conflicts with child nodes
)

CascadeConflict has two attributes:

deadline_conflicts: list[NodeInfo] — child nodes with later deadlines
status_conflicts: list[NodeInfo] — child nodes still started/inactive

Integration Guide

Using with an Existing Application

The library uses its own DeclarativeBase. If your application has its own base, you have two options:

Option A: Shared metadata (recommended)

Map the workflow tables into your application's metadata:

from business_workflow_manager.models import Base as WorkflowBase
from myapp.models import Base as AppBase

# Point workflow models at your app's metadata
WorkflowBase.metadata = AppBase.metadata

Option B: Separate metadata

Keep them separate and create tables from both:

AppBase.metadata.create_all(engine)
WorkflowBase.metadata.create_all(engine)

Entity ID

The Node.entity_id column is a plain string with no foreign key constraint. Your application is responsible for ensuring it references a valid entity. This keeps the library decoupled from your domain model.

# In your app, create workflow instances tied to your entities:
wf_node = mgr.start_workflow(workflow.wf_config_id, entity_id=str(my_entity.id))

Transaction Management

The library never calls session.commit(). It uses session.flush() to obtain generated primary keys but leaves transaction boundaries entirely to you:

with Session(engine) as session:
    mgr = WorkflowManager(session)
    workflow = mgr.load_workflow(config)
    # ... do other work in the same transaction ...
    session.commit()  # You decide when to commit

Loading a Selection of Workflows

A selection file is a plain-text list of workflow files, relative to a workflows folder, one per line. Blank lines and lines starting with # are ignored. Globs are not supported; absolute paths are rejected.

# workflows for customer X
auskunft.yml
baugesuch/vorabklaerung.yml

try:
    with open("config/workflows.txt") as f:
        loaded = mgr.load_workflows_from_selection(f, "workflows/")
    session.commit()
except Exception:
    session.rollback()
    raise

loaded contains one (path, LoadedWorkflow) tuple per listed file. A listed file that does not exist raises WorkflowException; errors from invalid workflow files propagate unchanged. Files are loaded one after another; the function only flushes, so roll back on error to discard files already loaded.

Translations are handled per row (key + language). This applies to every import path: load_workflow, load_workflow_from_file and load_workflows_from_selection:

row exists and was never edited (updated_at is empty) → value is updated from the YAML, updated_at stays empty
row exists and was edited (updated_at is set) → kept unchanged
key is in the YAML but not in the database → inserted
rows no longer produced by the YAML are not deleted

Edited translations are never overwritten or deleted by an import. There is no option to force an overwrite.

YAML Schema Reference

workflow:
  key: <uuid>                    # Optional. Stable ID for updates.
  title: <string | translated>   # Workflow title
  version: <int>                 # Schema version
  start_task_ref: <task_ref>     # Reference to the default initial task
  min_per_entity: <int>          # Minimum instances per entity
  max_per_entity: <int | null>   # Maximum instances (null = unlimited)
  time_period: <int | null>      # Default deadline in days
  tasks:
    <task_ref>:                   # String key used for linking
      key: <uuid>                # Optional. Stable ID for updates.
      title: <string | translated>
      start_task: <bool>         # If true, the workflow can alternatively be started at this task
      time_period: <int | null>
      triggers:                  # Optional event triggers
        - type: <string>         # Must match a registered event class name
          value: <object>        # Mapping → event(**value); scalar → event(value); omit → event()
      steps:
        - type: document
          key: <uuid>            # Optional
          title: <string | translated>
          optional: <bool>       # Default: false
          condition: <jmespath>  # Optional, skip if false
          time_period: <int | null>
        - type: form
          key: <uuid>
          title: <string | translated>
          name: <string>         # Form identifier
          optional: <bool>
          condition: <jmespath>
          time_period: <int | null>
          fields:
            - name: <string>
              type: <string>     # e.g. "boolean", "text", "select"
              label: <string | translated>
              choices:           # Optional, for select fields
                - label: <string | translated>
                  value: <string>
      links:
        - task_ref: <task_ref>   # Target task
          condition: <jmespath>  # Optional, follow if true

Translated strings can be either a plain string (used for all languages) or an object with de, fr, it keys.

Development

# Clone and install
git clone <repo-url>
cd business-workflow-manager
uv sync --dev

# Run tests
uv run pytest

# Run tests with verbose output
uv run pytest -v

# Build the shpinx documentation
cd docs
make clean html
open _build/html/index.html

License

GNU Lesser General Public License v3.0 only (LGPL-3.0-only) — see the files LICENSE (LGPL) and COPYING (GPL, which the LGPL builds on).

Name		Name	Last commit message	Last commit date
docs		docs
src/business_workflow_manager		src/business_workflow_manager
tests		tests
.gitignore		.gitignore
.python-version		.python-version
COPYING		COPYING
LICENSE		LICENSE
README.md		README.md
RELEASES.md		RELEASES.md
justfile		justfile
pyproject.toml		pyproject.toml
uv.lock		uv.lock

Latest commit

History

Folders and files

Repository files navigation