fanferro

Fast, robust SNS + SQS messaging for Python, powered by a Rust core.

fan-out (SNS → SQS) · fanfare (announcing to a topic) · ferro (iron: the Rust core)

import fanferro
from pydantic import BaseModel                  # or dataclasses, msgspec, protobuf, ...


@fanferro.message_type("checkout.order_placed.v1")
class OrderPlaced(BaseModel):
    order_id: str
    amount_cents: int


client = fanferro.Client(
    source="urn:service:checkout",
    provisioning=fanferro.Provisioning.ensure(),  # create missing resources (default: must exist)
)
orders = client.topic("orders")
billing = client.queue("billing", dead_letter=fanferro.DeadLetterQueue("billing-dlq"))
client.subscription(orders, billing)
consumer = client.consumer(billing, concurrency=20)


@consumer.route(OrderPlaced)
async def charge(order: OrderPlaced) -> None:
    ...                                    # return → ack · raise → retry with backoff → DLQ


async def main() -> None:
    async with client:                     # connect + provision declared resources
        await orders.publish(OrderPlaced(order_id="o-1", amount_cents=1299))
        await consumer.run()               # SIGTERM → graceful drain

Status: 0.x - the API may still change between minor versions.

Install

uv add fanferro                   # pip install fanferro
uv add "fanferro[all]"            # + pydantic, protobuf, msgspec, opentelemetry-api

AWS Lambda

Consume a queue in a Lambda function with an SQS event source mapping: each record is decoded, routed and retried like a message a consumer received, and failed records come back as batchItemFailures.

client = fanferro.Client(source="urn:service:billing")   # module level: reused while warm
billing = client.queue("billing", dead_letter=fanferro.DeadLetterQueue("billing-dlq"))
router = fanferro.Router()


@router.route(OrderPlaced)
async def charge(order: OrderPlaced) -> None:
    ...                                    # return → done · raise → retried · Reject → DLQ


handler = fanferro.aws_lambda.sqs_handler(client, router, queue=billing)   # handler(event, context)

Turn on ReportBatchItemFailures on the event source mapping. Permissions, outcomes and testing with fanferro.testing.invoke_sqs_handler: Deployment → Lambda.

Testing

Test routing and handlers without a broker: fanferro.testing.dispatch runs one message through the consumer's routing, decoding, middleware, handler and retry and dead-letter policies, and returns how it was settled. build_message gives a message the transport metadata a handler may depend on (attempt, group_id, the SNS topic, sent_at...).

from fanferro.testing import build_message, dispatch


async def test_charge() -> None:
    result = await dispatch(consumer, build_message(OrderPlaced(order_id="o-1", amount_cents=1299), attempt=3))
    assert result.outcome == "acked"     # or "retried", "requeued", "dead_lettered", "discarded"

Flows run on an in-memory broker (fanferro.testing.memory_client()) or an emulator: Testing.

Highlights

• Handlers are plain typed functions. Return to ack, raise to retry. No receipt handles.
• Your own types. Pydantic, dataclasses, msgspec, protobuf, JSON or raw bytes.
• Routing on anything. Message type, attribute, body field or topic.
• Safe by default. Heartbeats, retries with backoff, dead-letter queues, graceful shutdown.
• Infrastructure in code. Declared queues, topics and subscriptions are checked or created.
• Long-running workers or AWS Lambda. The same handlers and policies for SQS event sources.
• Tests without AWS or Docker. An in-memory broker runs the same engine you deploy.
• Mistakes caught early. Bad names, sizes and signatures fail before any network call.
• Observable. Typed events, structured logs, hooks and OpenTelemetry.
• The hard parts included. Large payloads via S3, outbox, idempotency, FIFO ordering.
• Fast. A Rust core, full batches and little Python work per message.
• No required dependencies. Fully typed, Python 3.12-3.15, wheels for Linux, macOS, Windows.

Performance

• Rust does the heavy lifting. Signing, HTTP, the SQS and SNS protocols and batching.
• Full batches, overlapping. Messages are received ten at a time, the next batch prefetched.
• Acks don't block handlers. Deletes are batched in the background while the next message runs.
• Little Python per message. Only your codecs, routing and handlers run on the event loop.
• Lean defaults. Small messages, no redundant checksums, no tracing work unless enabled.

Benchmarks and how to run them: Performance.

Real AWS, emulators and the in-memory broker

fanferro runs on real AWS, on the AWS emulators MiniStack and LocalStack, and on its in-memory broker. Real AWS is the reference: probes run the same scenarios and raw API calls on each target, and the in-memory broker follows AWS wherever they found a difference, with AWS's error codes and texts.

Neither emulator enforces SQS's 60 seconds between deleting a queue and creating it again or between two purges, drops SNS deliveries to queues encrypted with alias/aws/sqs as AWS does, or refuses unknown credentials. Every finding, the recorded runs and how to probe an account of your own: Parity with AWS.

Learn more

Using an AI coding agent? Point it at fanferro guide (the usage guide shipped with the package) or copy skills/fanferro into your project's skills.

Contributing

See AGENTS.md (for humans and AI agents): uv sync, make check, conventional commits.

License

MIT