rapids-streams 0.3.0

Rapids - A Python package

Rapids

A typed, async-first event router for Redis Streams. Build event-driven architectures with Pydantic-powered type safety, automatic event registration, and production-ready reliability features.

Table of Contents

• Features
• Installation
• Quick Start
• Events
  • BaseEvent
  • Event Inheritance
  • GenericEvent
  • Schema Versioning
• EventEmitter
  • Causation Chains
• Event Router
  • Basic Setup
  • Handler Matching
  • Multiple Handlers
• Correlation Tracking
• Error Handling and Retries
  • Two-Tier Retry System
  • Handler-Level Configuration
  • Backoff Strategies
  • Dead-Letter Queue (DLQ)
  • Exception Types
• Configuration Reference
• Complete Example
• Framework Integration
• Redis Streams Primer
• Development
• License

Features

• Typed Events - Define events as Pydantic models with automatic naming and registration
• Async-First - Built for asyncio with non-blocking I/O
• Flexible Routing - Match events by type, base class, union types, or glob patterns
• Consumer Groups - Distributed consumption with automatic load balancing
• Reliability - Retries, backoff strategies, timeouts, and dead-letter queues
• Correlation Tracking - Trace event chains with correlation and causation IDs
• Fast Serialization - Uses orjson for ~6x faster JSON encoding

Installation

pip install rapids-streams

Requirements: Python 3.12+, Redis 6.2+ (for XAUTOCLAIM support)

Quick Start

1. Define Events

from rapids import BaseEvent

class UserCreated(BaseEvent):
    user_id: str
    email: str

class OrderPlaced(BaseEvent):
    order_id: str
    user_id: str
    total: float

Events are automatically named based on class name:

• UserCreated → user.created
• OrderPlaced → order.placed
• S3ObjectUploaded → s3_object.uploaded

2. Emit Events

from redis.asyncio import Redis
from rapids import EventEmitter

redis = Redis()

# Create an emitter instance
emitter = EventEmitter(redis=redis, stream="events:all")

# Emit events
await emitter.emit(UserCreated(user_id="123", email="[e-mail]"))

# Batch emit (single Redis round-trip)
await emitter.emit([
    OrderPlaced(order_id="order-1", user_id="123", total=99.99),
    OrderPlaced(order_id="order-2", user_id="456", total=149.99),
])

3. Consume Events

from rapids import EventRouter

router = EventRouter(
    redis=redis,
    stream="events:all",
    group="order-service",
)

@router.on(OrderPlaced)
async def handle_order(event: OrderPlaced):
    print(f"Processing order {event.order_id} for ${event.total}")

# Start consuming (blocks until stopped)
await router.start()

Events

BaseEvent

All custom events inherit from BaseEvent, a frozen Pydantic model with automatic fields:

Field	Type	Description
id	str	Unique UUID (auto-generated)
timestamp	datetime	UTC creation time (auto-generated)
correlation_id	str | None	Groups related events together
causation_id	str | None	ID of the event that caused this one
metadata	dict	Arbitrary key-value data

from rapids import BaseEvent

class DocumentCreated(BaseEvent):
    doc_id: str
    title: str
    author: str

# Events are frozen (immutable)
event = DocumentCreated(
    doc_id="doc-123",
    title="Annual Report",
    author="alice",
    correlation_id="request-abc",
    metadata={"source": "api", "version": 2},
)

print(event.id)           # "550e8400-e29b-41d4-a716-446655440000"
print(event.timestamp)    # 2024-11-28T12:00:00+00:00
print(event.event_type)   # "document.created"

Event Inheritance

Create event hierarchies for shared fields and handler matching:

class DocumentEvent(BaseEvent):
    """Base class for all document events."""
    doc_id: str

class DocumentCreated(DocumentEvent):
    title: str
    author: str

class DocumentDeleted(DocumentEvent):
    deleted_by: str

class DocumentArchived(DocumentEvent):
    archive_location: str

GenericEvent

Unknown event types are deserialized as GenericEvent:

from rapids import GenericEvent

@router.on(GenericEvent)
async def handle_unknown(event: GenericEvent):
    print(f"Unknown type: {event.raw_type}")
    print(f"Raw data: {event.data}")

Schema Versioning

Events evolve over time. Rapids supports schema versioning to handle breaking changes without disrupting consumers.

Forward Compatibility

By default, events ignore unknown fields (extra="ignore"), so adding new optional fields is non-breaking:

# v1: Original event
class UserCreated(BaseEvent):
    user_id: str
    email: str

# v2: Added optional field - old consumers ignore it
class UserCreated(BaseEvent):
    user_id: str
    email: str
    display_name: str | None = None  # New optional field

Breaking Changes with Migrations

For breaking changes (new required fields, type changes, renamed fields), increment schema_version and use a Pydantic model_validator to migrate old data:

from typing import ClassVar
from pydantic import model_validator
from rapids import BaseEvent

class UserCreated(BaseEvent):
    schema_version: ClassVar[int] = 2  # Increment for breaking changes

    user_id: str
    email: str
    created_at: datetime  # New required field in v2

    @model_validator(mode="before")
    @classmethod
    def _migrate(cls, data: dict) -> dict:
        # _schema_version is injected during deserialization
        version = data.pop("_schema_version", cls.schema_version)

        if version < 2:
            # v1 → v2: Use event timestamp as created_at
            data["created_at"] = data.get("_timestamp")

        return data

The deserializer injects _schema_version (the version stored in the message) and _timestamp (the event timestamp) for use in migrations.

Multiple Version Migrations

For events with multiple version upgrades, chain migrations sequentially:

class OrderPlaced(BaseEvent):
    schema_version: ClassVar[int] = 3

    order_id: str
    user_id: str
    total: Decimal      # Changed from float in v2
    currency: str       # Added in v3

    @model_validator(mode="before")
    @classmethod
    def _migrate(cls, data: dict) -> dict:
        version = data.pop("_schema_version", cls.schema_version)

        # v1 → v2: Convert total from float to Decimal string
        if version < 2:
            data["total"] = str(data.get("total", 0))

        # v2 → v3: Add default currency
        if version < 3:
            data["currency"] = "USD"

        return data

Best Practices

1. Increment schema_version only for breaking changes
2. Keep migrations simple - transform data, don't add business logic
3. Test migrations with fixture data from each historical version
4. Document changes in the migration method docstring
5. Use _timestamp when new fields can derive from event metadata

EventEmitter

from rapids import EventEmitter

emitter = EventEmitter(
    redis=redis,
    stream="events:all",
    maxlen=100_000,      # Auto-trim stream (None to disable)
    max_retries=3,       # Retry on connection failures
    retry_delay=0.1,     # Base delay for exponential backoff
)

# Single event
msg_id = await emitter.emit(UserCreated(user_id="123", email="[e-mail]"))

# Batch emit (returns list of message IDs)
msg_ids = await emitter.emit([event1, event2, event3])

Causation Chains

Link events to track cause-effect relationships:

@router.on(DocumentCreated)
async def index_document(event: DocumentCreated):
    # Index the document...

    # Emit with causation link
    await emitter.emit(
        DocumentIndexed(doc_id=event.doc_id),
        caused_by=event,  # Sets causation_id to event.id
    )

Event Router

Basic Setup

from rapids import EventRouter, GroupStartPosition

router = EventRouter(
    redis=redis,
    stream="events:all",
    group="my-service",
    consumer="worker-1",  # Auto-generated if None
)

@router.on(UserCreated)
async def handle_user(event: UserCreated):
    print(f"New user: {event.email}")

# Start consuming
await router.start()

# Graceful shutdown
await router.stop()

Handler Matching

Rapids supports multiple ways to match events to handlers:

Exact Type Match

@router.on(UserCreated)
async def handle_user_created(event: UserCreated):
    pass

Base Class Match

Matches the base class and all its subclasses:

@router.on(DocumentEvent)
async def audit_all_documents(event: DocumentEvent):
    # Handles DocumentCreated, DocumentDeleted, DocumentArchived
    print(f"Document {event.doc_id}: {event.event_type}")

Union Types

Match multiple specific types:

@router.on(DocumentCreated | DocumentDeleted)
async def handle_lifecycle(event: DocumentCreated | DocumentDeleted):
    pass

Glob Patterns

Use wildcards for flexible matching:

@router.on("document.*")
async def handle_all_document_events(event):
    # Matches document.created, document.deleted, document.archived
    pass

@router.on("user.*")
async def handle_all_user_events(event):
    pass

@router.on("*")
async def log_everything(event):
    # Catch-all handler
    print(f"Event: {event.routing_key}")

Multiple Handlers

Multiple handlers can process the same event:

@router.on(OrderPlaced)
async def send_confirmation(event: OrderPlaced):
    # Send email
    pass

@router.on(OrderPlaced)
async def update_inventory(event: OrderPlaced):
    # Decrease stock
    pass

@router.on("*")
async def record_metrics(event):
    # Track all events
    pass

Correlation Tracking

Track related events across your system:

Correlation Scope

from rapids import correlation_scope, EventEmitter

emitter = EventEmitter(redis=redis, stream="events:all")

async def process_upload(file):
    async with correlation_scope() as correlation_id:
        # All events in this scope share the same correlation_id
        await emitter.emit(UploadStarted(filename=file.name))

        doc = await parse(file)
        await emitter.emit(DocumentCreated(doc_id=doc.id))

        await emitter.emit(UploadCompleted(doc_id=doc.id))

        print(f"All events have correlation_id: {correlation_id}")

Explicit Correlation

from rapids import get_correlation_id, get_causation_id

@router.on(OrderPlaced)
async def process_order(event: OrderPlaced):
    # Access correlation context
    print(f"Correlation: {event.correlation_id}")
    print(f"Causation: {event.causation_id}")

    # Continue the chain
    await emitter.emit(
        PaymentProcessed(order_id=event.order_id),
        caused_by=event,
    )

Error Handling and Retries

Two-Tier Retry System

Rapids provides two levels of retry protection:

1. Quick In-Process Retries - Fast retries within the handler
2. Redis-Level Retries - Automatic re-delivery via consumer groups

Handler-Level Configuration

from rapids.backoff import exponential, linear, constant

@router.on(
    ExternalApiEvent,
    timeout=60.0,                           # Handler timeout in seconds
    retries=3,                              # Quick retry attempts
    backoff=exponential(base=2, max_delay=30),  # Delay between retries
    retryable_exceptions=(ConnectionError, TimeoutError),
)
async def call_external_api(event: ExternalApiEvent):
    async with httpx.AsyncClient() as client:
        await client.post("https://api.example.com", json=event.model_dump())

Backoff Strategies

from rapids.backoff import exponential, linear, constant

# Exponential: 2, 4, 8, 16, 32, 60... (with jitter to prevent thundering herd)
exponential(base=2.0, max_delay=60.0, jitter=True)

# Linear: 1, 2, 3, 4, 5...
linear(delay=1.0, max_delay=60.0)

# Constant: 5, 5, 5, 5...
constant(delay=5.0)

Dead-Letter Queue (DLQ)

Failed messages are sent to a DLQ for later inspection:

router = EventRouter(
    redis=redis,
    stream="events:all",
    group="my-service",
    max_deliveries=5,              # Send to DLQ after 5 failed deliveries
    dlq_stream="events:dlq",       # DLQ stream name
    dlq_maxlen=10_000,             # Trim DLQ to 10k messages
)

Messages are sent to DLQ when:

• Delivery count exceeds max_deliveries
• Deserialization fails
• Handler raises a non-retryable exception

DLQ entry format:

{
  "original_msg_id": "1701234567890-0",
  "original_stream": "events:all",
  "original_group": "my-service",
  "reason": "max_deliveries_exceeded",
  "error": "Connection refused",
  "handler": "call_external_api",
  "failed_at": "2024-11-28T12:00:00+00:00",
  "event_type": "external_api.event",
  "data": "{...}"
}

Exception Types

from rapids.exceptions import (
    RapidsError,              # Base exception
    EmitError,                # Failed to emit after retries
    DeserializationError,     # Invalid event data
    HandlerTimeoutError,      # Handler exceeded timeout
    MaxRetriesExceededError,  # Quick retries exhausted
    NonRetryableError,        # Handler raised non-retryable exception
)

Configuration Reference

EventEmitter

Option	Default	Description
redis	required	Redis async client
stream	"events:all"	Target stream name
maxlen	100_000	Auto-trim stream (None to disable)
max_retries	3	Retry attempts on emit failure
retry_delay	0.1	Base delay for exponential backoff

EventRouter

Option	Default	Description
redis	required	Redis async client
stream	"events:all"	Source stream name
group	"default"	Consumer group name
consumer	auto	Consumer name (hostname-pid if None)
batch_size	10	Messages per XREADGROUP
block_ms	5000	Blocking read timeout (ms)
handler_timeout	30.0	Default handler timeout (seconds)
group_start_id	BEGINNING	Where new groups start
claim_interval	10	XAUTOCLAIM every N reads
claim_min_idle_ms	60_000	Claim messages idle > this
default_retries	0	Default quick retries
default_backoff	exponential()	Default backoff strategy
max_deliveries	5	Deliveries before DLQ
dlq_stream	None	DLQ stream (None to disable)
dlq_maxlen	10_000	DLQ trim size

GroupStartPosition

from rapids import GroupStartPosition

# Read entire stream history (for new groups that need historical data)
GroupStartPosition.BEGINNING  # "0"

# Only new messages after group creation
GroupStartPosition.LATEST     # "$"

Handler Decorator

@router.on(
    target,                    # Event class, union, or pattern
    timeout=30.0,              # Override router default
    retries=0,                 # Quick retry attempts
    backoff=exponential(),     # Backoff strategy
    retryable_exceptions=(),   # Exceptions to retry
)

Complete Example

import asyncio
from redis.asyncio import Redis
from rapids import (
    BaseEvent,
    EventEmitter,
    EventRouter,
    correlation_scope,
    GroupStartPosition,
)
from rapids.backoff import exponential


# Define events
class OrderPlaced(BaseEvent):
    order_id: str
    user_id: str
    items: list[str]
    total: float


class PaymentProcessed(BaseEvent):
    order_id: str
    payment_id: str
    amount: float


class OrderShipped(BaseEvent):
    order_id: str
    tracking_number: str


# Application setup
async def main():
    redis = Redis(host="localhost", port=6379)

    # Create emitter instance
    emitter = EventEmitter(redis=redis, stream="orders:events")

    # Setup router
    router = EventRouter(
        redis=redis,
        stream="orders:events",
        group="fulfillment-service",
        handler_timeout=60.0,
        max_deliveries=3,
        dlq_stream="orders:dlq",
    )

    @router.on(OrderPlaced)
    async def process_payment(event: OrderPlaced):
        print(f"Processing payment for order {event.order_id}")
        # Process payment...
        payment_id = f"pay-{event.order_id}"

        await emitter.emit(
            PaymentProcessed(
                order_id=event.order_id,
                payment_id=payment_id,
                amount=event.total,
            ),
            caused_by=event,
        )

    @router.on(
        PaymentProcessed,
        retries=3,
        backoff=exponential(base=2, max_delay=30),
    )
    async def ship_order(event: PaymentProcessed):
        print(f"Shipping order {event.order_id}")
        # Create shipment...

        await emitter.emit(
            OrderShipped(
                order_id=event.order_id,
                tracking_number="1Z999AA10123456784",
            ),
            caused_by=event,
        )

    @router.on("order.*")
    async def audit_log(event):
        print(f"[AUDIT] {event.routing_key}: {event.id}")

    # Start consuming
    print("Starting order fulfillment service...")
    await router.start()


if __name__ == "__main__":
    asyncio.run(main())

Framework Integration

FastAPI

Rapids supports deferred Redis setup, ideal for FastAPI's lifespan pattern:

import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
from redis.asyncio import Redis
from rapids import BaseEvent, EventEmitter, EventRouter

# Define events
class UserCreated(BaseEvent):
    user_id: str
    email: str

# Create at module level - no Redis yet
router = EventRouter(stream="events", group="my-service")
emitter = EventEmitter(stream="events")

@router.on(UserCreated)
async def handle_user(event: UserCreated):
    print(f"New user: {event.email}")

@asynccontextmanager
async def lifespan(app: FastAPI):
    redis = Redis()
    emitter.configure(redis)       # Configure emitter
    task = asyncio.create_task(router.start(redis))  # Start router
    yield
    await router.stop()
    task.cancel()
    await redis.close()

app = FastAPI(lifespan=lifespan)

@app.post("/users")
async def create_user(email: str):
    event = UserCreated(user_id="123", email=email)
    await emitter.emit(event)
    return {"status": "ok"}

Redis Streams Primer

Rapids uses Redis Streams consumer groups for reliable message delivery:

• At-least-once delivery: Messages are redelivered until acknowledged
• Consumer groups: Multiple consumers share the workload
• Pending messages: Failed messages stay pending for redelivery
• XAUTOCLAIM: Automatically reclaim messages from dead consumers

Key Redis commands used:

• XADD - Add events to stream
• XGROUP CREATE - Create consumer group
• XREADGROUP - Read messages for consumer
• XACK - Acknowledge message processing
• XAUTOCLAIM - Claim idle messages from dead consumers

Development

Setup

# Install uv if not already installed
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install dependencies
make install

Commands

make help          # Show all commands
make test          # Run all tests
make test-unit     # Run unit tests only (fast, no Docker)
make test-integration  # Run integration tests (requires Docker)
make lint          # Run linting
make format        # Format code
make typecheck     # Run type checking
make ci            # Run full CI checks

Testing

Unit tests use fakeredis for fast, isolated testing. Integration tests use testcontainers to spin up real Redis instances.

License

MIT