slck-mediator 1.0.2

Cross-language Mediator pattern framework — send/stream/publish, 11-middleware pipeline, outbox, inbox, saga, and event bus

slck-mediator — Python

Cross-language Mediator pattern framework for Python.
Send/stream/publish · 11-middleware pipeline · outbox · inbox · saga · integration event bus · full observability.

📖 Full Documentation

Installation

pip install slck-mediator

With optional extras:

pip install "slck-mediator[otel]"    # OpenTelemetry adapters
pip install "slck-mediator[sql]"     # SQLAlchemy adapters
pip install "slck-mediator[mongo]"   # Motor (MongoDB) adapters
pip install "slck-mediator[all]"     # Everything

Quick Start

from mediator import Mediator, Request, ConsoleLogger, InMemoryMetrics, InMemoryTracer

class GreetCommand(Request[str]):
    def __init__(self, name: str) -> None:
        self.name = name

async def greet_handler(ctx, cmd: GreetCommand) -> str:
    return f"Hello, {cmd.name}!"

async def main():
    logger  = ConsoleLogger()
    metrics = InMemoryMetrics()
    tracer  = InMemoryTracer("app", logger, metrics)
    m = Mediator(logger=logger, metrics=metrics, tracer=tracer)
    m.register_request_handler(GreetCommand, greet_handler)

    result = await m.send(GreetCommand("World"))
    print(result)  # Hello, World!

Code Generation

Generate a build-time direct-dispatch mediator from your feature handlers:

python tools/generate_mediator.py

Development

pip install -e ".[dev]"
pytest

Key modules:

• tools/generate_mediator.py: build-time direct-dispatch generator for the example package with duplicate, missing-handler, polymorphic notification, generic-contract, and unsupported-constructor-or-factory diagnostics, and inherited-request-or-stream-contract diagnostics, and handler-or-message-discovery diagnostics
• example/generated_mediator.py: generated direct-dispatch, polymorphic notification resolution, service registration, typed mediator facade, and dependency-wiring factory used by the example composition root
• mediator/mediator.py: async runtime mediator with send, stream, and publish
• mediator/registry.py: runtime request, stream, and notification registry
• mediator/pipeline.py: middleware behaviors including rate limiting and bulkhead controls
• mediator/eventbus.py: integration event transports, including concrete Kafka producer, NATS client, RabbitMQ client, and SQS client-backed adapters, plus versioned JSON envelope serializers
• mediator/outbox.py: file-backed, in-memory, and adapter-backed outbox store, dead-letter store and replay manager, and worker
• mediator/inbox.py: file-backed, in-memory, and adapter-backed idempotent consumer inbox support
• mediator/saga.py: saga orchestration with durable file persistence and adapter-backed optimistic stores
• mediator/lifecycle.py: hosted worker lifecycle, health/readiness, and graceful shutdown coordination
• mediator/fastapi.py: FastAPI dependency helper
• example/main.py: runnable CQRS, stream, outbox, inbox, and saga demo using the generated service collection/provider integration and typed mediator facade

Built-in notification publishers: ForeachAwaitPublisher, TaskWhenAllPublisher, plus compatibility aliases for SequentialNotificationPublisher and ParallelNotificationPublisher.

Versioned integration messages use a canonical JSON envelope with specVersion, version, schema, occurredAt, and idempotencyKey, and the example logs both the serialized envelope and round-trip deserialization.