ferricstore 0.1.1

Python SDK for FerricStore and FerricFlow

FerricStore Python SDK

Python SDK for FerricStore and FerricFlow.

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

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 Redis-compatible port enabled:

ferricstore start

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

redis://127.0.0.1:6379/0

3. Create a durable queue item

from ferricstore import QueueClient

client = QueueClient.from_url("redis://127.0.0.1:6379/0")
emails = client.queue(type="email")

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

4. Run a queue worker

from ferricstore import QueueClient

client = QueueClient.from_url("redis://127.0.0.1:6379/0")
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("redis://127.0.0.1:6379/0")
order = client.workflow(
    type="order",
    initial_state="created",
    partition_by=("tenant_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",
    tenant_id="tenant-a",
    order_id="order-1",
    payload=b"order payload",
    idempotent=True,
)

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

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 the write.	Let the SDK retry/back off; reduce producer 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.
• RetryPolicy, WorkerConfig, ValueConfig, and ExceptionPolicy for runtime defaults.
• RawCodec by default, JsonCodec when you want JSON payloads.
• client.command(...) as the Redis/FerricStore escape hatch.

Async quickstart

import asyncio

from ferricstore import AsyncQueueClient


async def main():
    client = AsyncQueueClient.from_url("redis://127.0.0.1:6379/0")
    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, connection pools, lease duration, backpressure behavior, graceful shutdown, and value hydration caps.

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/native_commands.py: Redis/FerricStore command helpers.
• examples/dbos_style_benchmark.py: DBOS-style throughput benchmark.

Contributing

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