petrus-engine 0.1.0

Petrus: an embedded Petri net workflow engine with colored tokens, timed transitions, and time travel.

Petrus: an embedded Petri net workflow engine for Python.

Petrus executes workflows defined as Petri nets. You describe the structure of a workflow — places, transitions, and arcs — and provide the behavior separately as plain Python callables. The engine takes care of token flow, enabling rules, guards, timed transitions, and a full state history with time travel.

What you can do with Petrus?

1. Define nets in JSON files (editable by tooling or a visual editor) or with a concise Python DSL.
2. Keep structure and behavior separate: transitions reference guards and handlers by name; you bind the callables at runtime.
3. Flow typed payloads through the net with colored tokens — guards and handlers declare typed parameters and the engine extracts the right payload automatically.
4. Compose nets from reusable subnets via file references, with input/output/awaiting interface places marking the boundaries.
5. Travel in time: every fire is recorded, so you can undo, redo, and branch from any past state.
6. Drive long-running workflows with split-phase firing, designed to await async handlers — with zero runtime dependency on any workflow platform.

How to use it?

Install from PyPI:

uv add petrus-engine        # or: pip install petrus-engine

Build a net with the DSL, bind behavior, and run it:

from petrus.dsl import Net
from petrus.engine import PetriNetRuntime

net = Net("approval")
p, t = net.p, net.t
p.pending(initial=1) > t.validate > p.validated
p.validated > t.approve(guard="is_approved") > p.approved
p.validated > t.reject(guard="is_rejected") > p.rejected

guards = {
    "is_approved": lambda d: d.value == "approve",
    "is_rejected": lambda d: d.value == "reject",
}

rt = PetriNetRuntime.from_schema(net.build(), guards=guards)

rt.fire(abs(net.t.validate))

A transition is enabled when its input places hold enough tokens, its inhibitor and read arcs are satisfied, and its guard (if any) returns True for the payloads peeked from the input tokens. Firing consumes the input tokens and deposits a merged output token in every output place.

Loading nets from JSON files

Nets can live in JSON files (fileVersion: "2" flat format) and reference other net files as subnets. Resolve a root file and hand the schema to the engine:

from petrus.schema import FlatNet

net = FlatNet.from_filename("nets/net_approval.json")
rt = PetriNetRuntime.from_schema(net.to_schema(), guards=guards, handlers=handlers)

CLI tools

# Generate a .pyi stub from a module exposing SCHEMA (for IDE autocomplete)
petrus-cli gen-stub myapp.nets -o nets/__init__.pyi

# Validate and canonically format net JSON files in-place (pre-commit friendly)
python -m petrus.cli.format_json --fail-on-change nets/net_*.json

Stub generation formats its output with black; install the cli extra to use it:

uv add "petrus-engine[cli]"

Timed transitions

petrus.timers.TimerManager syncs timer tasks with the net's enabled timed transitions. It is clock- and sleep-agnostic: by default it uses the system UTC clock and asyncio.sleep, and you can inject your own primitives — to control time in tests, or to integrate with a scheduling runtime:

from petrus.timers import TimerManager

tm = TimerManager()  # system UTC clock + asyncio.sleep
tm = TimerManager(now=my_clock, sleep=my_sleep)  # custom primitives

Thank you to Routable

Routable sponsored the development of this library. Working at Routable is an awesome experience, with a developer-first culture that fosters innovation and growth. If you're interested in joining a dynamic team, check out our job opportunities here!

Authors

• Henrique Bastos [e-mail]