pyuow 0.9.1

Python custom implementation of unit of work pattern

Unit of Work for Python — composable units, transactional work managers, and a domain model toolkit.

📖 Read the docs  ·  Quickstart  ·  PyPI

---

What is PyUoW?

PyUoW is a Python implementation of the Unit of Work behavioural pattern. It gives you small, typed building blocks for business logic that compose into a flow and run inside a Work Manager that handles transactions, batching, and domain events.

The result: business logic that reads top-to-bottom, separates orchestration from execution, and stays testable as it grows.

Highlights

• Composable units — write each step as a ConditionalUnit / RunUnit / FinalUnit, chain them with >>, validate the chain at build time.
• Transactional out of the box — a SQLAlchemy adapter ships in pyuow.contrib.sqlalchemy (sync + async), with nested-transaction support.
• Domain-first — Entity, AuditedEntity, SoftDeletableEntity, VersionedEntity, and an event-emitting Model base, all immutable dataclasses.
• Sync and async parity — every primitive has an aio/ twin; pick one per flow.
• Strict types — passes mypy --strict; runs on Python 3.10 through 3.14.

---

Install

pip install pyuow                # core
pip install "pyuow[sqlalchemy]"  # with SQLAlchemy integration

Python ≥ 3.10.

AI-agent skill

After installing, run pyuow install-skill to drop an AI-agent skill file into your project so Claude Code / OpenCode learn PyUoW's idioms. Add --global to install for your user account, or --check to dry-run.

See the step-by-step guide (or docs/install-skill.md).

At a glance

from dataclasses import dataclass

from pyuow import (
    BaseContext, BaseParams, ConditionalUnit, ErrorUnit,
    FinalUnit, Result, RunUnit,
)
from pyuow.work.noop import NoOpWorkManager


@dataclass(frozen=True)
class Greeting(BaseParams):
    name: str


@dataclass
class Ctx(BaseContext[Greeting]):
    params: Greeting


class IsNamePresent(ConditionalUnit[Ctx, str]):
    def condition(self, ctx: Ctx) -> bool:
        return bool(ctx.params.name)


class Greet(RunUnit[Ctx, str]):
    def run(self, ctx: Ctx) -> None:
        print(f"Hello, {ctx.params.name}!")


class Done(FinalUnit[Ctx, str]):
    def finish(self, ctx: Ctx) -> Result[str]:
        return Result.ok("greeted")


flow = (
    IsNamePresent(on_failure=ErrorUnit(exc=ValueError("name required")))
    >> Greet()
    >> Done()
).build()

result = NoOpWorkManager().by(flow).do_with(Ctx(params=Greeting(name="Alice")))
assert result.get() == "greeted"

What's in the box

Concept	What it gives you
Units & Flow	ConditionalUnit / RunUnit / FinalUnit / ErrorUnit, chained with >>
Result	ok / error / empty plus .map, .and_then, .unwrap_or
Context	Mutable, immutable, and domain-aware context bases
Work Manager	NoOp, transactional, and domain-transactional managers
Domain Model	Entity, AuditedEntity, Model, Batch, events, typed exceptions
DataPoints	Typed producer / consumer contracts between units
SQLAlchemy	Ready-made repositories, table mixins, and transaction manager
Async	A sync and an aio/ twin for every primitive

Contributing

PRs welcome — see CONTRIBUTING.md for the dev setup, or run:

poetry install --with docs
make tests        # pytest + coverage
make fmt          # ruff + mypy
make docs-serve   # live preview at http://127.0.0.1:8000