Quickstart¶
Requirements¶
- Python 3.12 or newer.
- Pydantic 2 (installed automatically).
- PostgreSQL 16+ (
asyncpg) or Redis 7+ (redis) only if you use those backends.
Install¶
For development from a checkout:
1. In-memory: one transaction¶
import asyncio
from statepact import TransactionManager
async def main() -> None:
manager = TransactionManager()
users = manager.resource("users", {"u1": {"credits": 5}})
await users.ensure(manager)
async with manager.session(agent_id="billing", resources=[users]) as session:
user = await session.read("u1")
user["credits"] -= 1
await session.write("u1", user)
result = await session.commit()
print(result.commit_version, await users.read("u1"))
await manager.close()
asyncio.run(main())
session.commit() is explicit. Leaving the async with block without it aborts
the transaction, releasing snapshots and any pessimistic locks.
2. Retry a conflicting transaction¶
from statepact import RetryPolicy
stock = await manager.run(
"agent-1",
[inventory],
reserve_one_unit, # async def reserve_one_unit(session): ...
isolation="optimistic",
retry_policy=RetryPolicy(max_attempts=100, base_delay=0.002, jitter=0.3),
)
run opens a fresh session per attempt. Only retryable conflicts are retried by
default. Use run_outcome when you need attempt counts, delays and the collected
errors:
outcome = await manager.run_outcome("agent-1", [inventory], reserve_one_unit)
print(outcome.value, outcome.attempts, outcome.errors)
To update an agent's plan between attempts, pass replan:
async def replan(error, attempt):
await agent.observe_conflict(error, attempt)
await manager.run("agent-1", [inventory], work, replan=replan)
3. Make a commit idempotent¶
async with manager.session(
agent_id="billing",
resources=[wallet],
idempotency_key="invoice-42",
) as session:
balance = await session.read("balance")
await session.write("balance", balance - 30)
result = await session.commit()
print(result.replayed) # False first time, True on replay
The default fingerprint hashes the agent id and the write footprint, not the
values. A retried read-modify-write therefore replays instead of raising a
mismatch. Pass idempotency_fingerprint="..." to bind the full payload
explicitly.
4. Use PostgreSQL¶
docker compose up -d postgres
export STATEPACT_DATABASE_URL=postgresql://statepact:statepact@localhost:5432/statepact
import os
from statepact import TransactionManager
from statepact.storage.postgres import PostgresBackend
dsn = os.environ["STATEPACT_DATABASE_URL"]
backend = await PostgresBackend.connect(dsn)
manager = TransactionManager(backend)
The schema is created automatically. Multiple processes can share the database; commits are validated under one advisory lock.
5. Use Redis¶
from statepact.storage.redis import RedisBackend
backend = await RedisBackend.connect("redis://localhost:6379/0", prefix="myapp")
manager = TransactionManager(backend)
Each commit is one Lua script, so validation and application are atomic. All keys
are namespaced under prefix.
6. Protect a leased job¶
lease = await manager.lease("job", owner="worker-a", ttl_seconds=30)
async with manager.session(agent_id="worker-a", resources=[job], fence=lease.token):
...
await session.commit()
await backend.renew_lease(lease) # heartbeat before the TTL expires
await backend.release_lease(lease)
A stale token is rejected with LeaseLostError; re-acquire the lease before
retrying the write.
7. Run the playground¶
The playground runs the real engine in-process and compares isolation modes on inventory, counter and write-skew scenarios.
Next steps¶
- Concepts — sessions, resources, versions, commit plans.
- Isolation — choose the right mode and understand its anomalies.
- Backends — deployment trade-offs and schemas.
- Limitations — read this before production.