FerricStore Python SDK

Python SDK for FerricStore and FerricFlow.

Status: public alpha 0.8.0. APIs may change before 1.0, but the SDK is tested against command construction, queue/workflow handlers, leases, retries, history, indexed attributes, named values, idempotent create, worker loops, async flows, and local FerricStore integration scenarios.

Version 0.8.0 requires FerricStore 0.11.0 or newer. This is a breaking beta contract update; the native wire protocol remains v1.

FerricFlow keeps each workflow or job's state and history in one durable place. It is an explicit durable state pipeline, not a hidden deterministic replay engine:

create -> claim -> handler -> transition/complete/retry/fail

Handlers should be idempotent because work can be retried after lease expiry, worker crash, or explicit retry.

Durability is the default contract. A workflow command returns success only after the state change is accepted through FerricStore's quorum path and written to disk.

First 10 minutes

1. Install

pip install ferricstore

For local development from this repo:

python3 -m venv .venv
. .venv/bin/activate
pip install -e ".[dev]"

2. Start FerricStore

Use a local FerricStore server with the FerricStore protocol listener enabled:

ferricstore start

If you are running from the FerricStore source repo, use that repo's documented server command. The SDK examples assume:

ferric://127.0.0.1:6388

3. Query durable runs

Use parameterized FQL for bounded, partition-scoped reads. Cursors are opaque and must be reused with the same query and parameters.

from ferricstore import FlowClient

client = FlowClient.from_url("ferric://127.0.0.1:6388")
query = """
FROM runs
WHERE partition_key = @partition AND type = @type AND state = @state
ORDER BY updated_at_ms ASC
LIMIT 25
RETURN RECORDS
"""
params = {"partition": "partition-a", "type": "invoice", "state": "queued"}

result = client.query(query, params)
plan = client.explain(query, params)
indexes = client.query_indexes()

query_indexes() exposes each generation's bounded covering_fields and opaque format codec identities alongside build, validation, retirement, and statistics status. format.counter is None when an index has no exact-count prefix. A format change means the derived query projection must be rebuilt; authoritative Flow state is not rewritten.

For bounded request execution, pass deadline_ms as an absolute Unix timestamp in milliseconds to query, explain, or explain_analyze.

For reusable, composable queries, use the immutable query builder. It compiles to the same parameterized FQL1 and can be passed directly to query or explain:

from ferricstore import FlowFields, FlowQuery, flow_param

queued_invoices = (
    FlowQuery.runs()
    .where(
        FlowFields.partition_key.eq(flow_param("partition")),
        FlowFields.type.eq("invoice"),
        FlowFields.state.eq("queued"),
    )
    .order_by(FlowFields.updated_at_ms.desc())
    .limit(25)
    .return_records(FlowFields.run_id, FlowFields.state, FlowFields.updated_at_ms)
    .bind(partition="partition-a")
)

result = client.query(queued_invoices)

4. Create a durable queue item

from ferricstore import FlowClient, QueueClient

client = QueueClient.from_url("ferric://127.0.0.1:6388")
emails = client.queue(type="email")

emails.enqueue("email-1", payload=b"welcome:user-1", idempotent=True)

Use attributes for small indexed metadata you want to filter/count later:

emails.enqueue(
    "email-2",
    payload=b"welcome:user-2",
    partition_key="account-a",
    attributes={"account": "acme", "campaign": "summer"},
    idempotent=True,
)

flow = FlowClient.from_url("ferric://127.0.0.1:6388")
records = flow.list(
    "email",
    partition_key="account-a",
    attributes={"account": "acme"},
)
stats = flow.stats(
    "email",
    partition_key="account-a",
    attributes={"account": "acme"},
)

Attributes are not payload bytes. Use named values/value refs for large data.

FIFO Flow state policy is opt-in per state:

from ferricstore import FlowStatePolicy

flow.install_policy("email", states={"queued": FlowStatePolicy.fifo()})
emails.enqueue("email-3", payload=b"welcome", partition_key="account-a:email")

FIFO states require a partition_key; priority is for parallel states.

Policy writes are deep patches by default and return a typed snapshot with a monotonic generation. Use compare-and-swap when coordinating policy writers:

from ferricstore import StalePolicyGenerationError

snapshot = flow.policy_get("email")
try:
    snapshot = flow.install_policy(
        "email",
        expected_generation=snapshot.generation,
        states={"queued": FlowStatePolicy.fifo()},
    )
except StalePolicyGenerationError:
    snapshot = flow.policy_get("email")

Workflow install_policy() calls default to full replacement because the workflow definition is the source of truth. Pass replace=False to request a patch explicitly.

4. Run a queue worker

from ferricstore import QueueClient

client = QueueClient.from_url("ferric://127.0.0.1:6388")
emails = client.queue(type="email")


def send_email(job):
    print(f"send {job.id}: {job.payload!r}")
    return b"sent"


emails.worker(concurrency=10, batch_size=100).run(send_email)

If the handler raises, the default worker policy is retry.

5. Create a workflow/state machine

Use workflows when one durable flow moves through named states.

from ferricstore import WorkflowClient, complete, transition

client = WorkflowClient.from_url("ferric://127.0.0.1:6388")
order = client.workflow(
    type="order",
    initial_state="created",
    partition_by=("account_id", "order_id"),
)


@order.state("created")
def created(job):
    charge_card(job.payload)
    return transition("charged")


@order.state("charged")
def charged(job):
    send_receipt(job.id)
    return complete(result=b"ok")


order.start(
    "order-1",
    account_id="account-a",
    order_id="order-1",
    payload=b"order payload",
    idempotent=True,
)

order.worker(states=["created", "charged"], concurrency=10, batch_size=100).run()

Derived workflow partitions use collision-free fpk:<byte-length>:<value> encoding in 0.6.0. Drain flows created with the old colon-joined derived keys before upgrading. Explicit partition_key values are unchanged.

6. Store and fetch named values

Use named values when different states need different pieces of data. Values are stored as FerricFlow value refs and are only hydrated when requested.

emails.enqueue(
    "email-2",
    payload=b"small routing bytes",
    values={
        "template": b"welcome template bytes",
        "profile": b"user profile snapshot",
    },
    idempotent=True,
)

emails.worker(claim_values=["template"]).run(send_email)

Fetch one or many values directly when needed:

profile = client.value_get(owner_flow_id="email-2", name="profile")
values = client.value_mget(
    owner_flow_id="email-2",
    names=["template", "profile"],
)

Use ValueConfig or value_max_bytes in production to cap large value reads.

7. Inspect history

record = emails.get("email-1")
history = emails.history("email-1")

print(record)
for event in history:
    print(event)

History is for debugging and audit. Handlers should use claimed job data and requested values, not history replay.

8. Common errors

Error	Meaning	Usual fix
FlowAlreadyExistsError	The flow id already exists.	Use idempotent=True for safe producer retries or generate a new id.
FlowNotFoundError	The flow does not exist or was retained/expired.	Check id, partition inputs, and retention policy.
FlowWrongStateError	The command expected a different current state.	Check worker state filters and handler transitions.
StaleLeaseError	A worker tried to complete with an old lease.	Keep handlers under lease_ms or renew/retry safely.
OverloadedError	Server backpressure rejected a safe operation.	Let the SDK retry/back off; reduce request rate under sustained pressure.

What you use

• QueueClient / AsyncQueueClient for durable queues.
• WorkflowClient / AsyncWorkflowClient for explicit durable state machines.
• FlowClient / AsyncFlowClient for advanced command-level control.
• ScheduleResult, EffectResult, ApprovalResult, CircuitBreakerStatus, BudgetResult, and GovernanceOverview for typed admin/governance responses with dict fallback.
• RetryPolicy, WorkerConfig, ValueConfig, and ExceptionPolicy for runtime defaults.
• RawCodec by default, JsonCodec when you want JSON payloads.
• client.command(...) as the FerricStore low-level command escape hatch.

Async quickstart

import asyncio

from ferricstore import AsyncQueueClient


async def main():
    client = AsyncQueueClient.from_url("ferric://127.0.0.1:6388")
    emails = client.queue(type="email")

    async def handler(job):
        await send_email_async(job.payload)

    await emails.worker(concurrency=100, batch_size=500).run(handler)


asyncio.run(main())

Production shape

Use one process/service to create work and a separate long-lived worker service to claim and complete work.

web/serverless producer -> FerricStore -> worker service

Before production, configure timeouts, lease duration, backpressure behavior, graceful shutdown, and value hydration caps. The ferric:// transport defaults to one multiplexed connection with 8 request lanes; only raise connection or lane counts after profiling shows client-side saturation.

Docs

• Documentation index
• Quickstart
• SDK guide
• Configuration
• Production readiness
• Data in workflows
• Worker runtime
• Async APIs
• Use cases
• Testing
• Troubleshooting

Examples

• examples/order_workflow.py: two-state workflow.
• examples/queue_worker.py: queue producer and worker.
• examples/async_queue_worker.py: async queue producer and worker.
• examples/state_machine_workflow.py: explicit workflow runner.
• examples/protocol_commands.py: FerricStore command helpers.
• examples/protocol_kv_benchmark.py: protocol SET/GET benchmark.
• examples/protocol_dbos_benchmark.py: protocol DBOS-style queued workflow benchmark.
• examples/dbos_style_benchmark.py: DBOS-style throughput benchmark.

Contributing

See CONTRIBUTING.md, CODE_OF_CONDUCT.md, SECURITY.md, and RELEASE.md.