livehop-minimal-mediator 1.0.0

A minimal, lightweight mediator implementation for Python applications supporting CQRS pattern with commands and queries

livehop-minimal-mediator

A minimal, lightweight mediator implementation for Python applications supporting the CQRS pattern with commands and queries.

Features

• ✨ Minimal and lightweight - Simple implementation with no unnecessary complexity
• 🚀 Easy to use - Clean, intuitive API
• 🎯 CQRS support - Perfect for implementing commands and queries
• ⚡ Async/Await - Full async support for modern Python applications
• 🔧 Type hints - Fully typed for better IDE support and type checking
• 📦 Zero dependencies - No external dependencies required

Installation

pip install livehop-minimal-mediator

Quick Start

1. Create a Query/Command

from livehop_minimal_mediator import IRequest, IRequestHandler

# Define your request
class GetUsersQuery(IRequest[list[dict]]):
    pass

# Define your handler
class GetUsersHandler(IRequestHandler[list[dict]]):
    async def handle(self, request: GetUsersQuery) -> list[dict]:
        # Your logic here
        users = [
            {"id": 1, "name": "John Doe"},
            {"id": 2, "name": "Jane Smith"}
        ]
        return users

2. Register and Use

from livehop_minimal_mediator import Mediator
import asyncio

# Create mediator instance
mediator = Mediator()

# Register handler
mediator.register_handler(GetUsersQuery, lambda: GetUsersHandler())

# Create sender
sender = mediator.create_sender()

# Use it
async def main():
    query = GetUsersQuery()
    users = await sender.send(query)
    print(users)

asyncio.run(main())

Usage Examples

Command with Parameters

from dataclasses import dataclass
from livehop_minimal_mediator import IRequest, IRequestHandler

@dataclass
class CreateUserCommand(IRequest[int]):
    name: str
    email: str

class CreateUserHandler(IRequestHandler[int]):
    def __init__(self, db_connection):
        self.db = db_connection

    async def handle(self, request: CreateUserCommand) -> int:
        # Insert user logic
        user_id = await self.db.insert_user(request.name, request.email)
        return user_id

# Register with dependencies
mediator.register_handler(
    CreateUserCommand,
    lambda: CreateUserHandler(db_connection)
)

# Use it
command = CreateUserCommand(name="Alice", email="[e-mail]")
user_id = await sender.send(command)

Query with Parameters

@dataclass
class GetUserByIdQuery(IRequest[dict | None]):
    user_id: int

class GetUserByIdHandler(IRequestHandler[dict | None]):
    def __init__(self, db_connection):
        self.db = db_connection

    async def handle(self, request: GetUserByIdQuery) -> dict | None:
        user = await self.db.get_user(request.user_id)
        return user

# Register and use
mediator.register_handler(
    GetUserByIdQuery,
    lambda: GetUserByIdHandler(db_connection)
)

query = GetUserByIdQuery(user_id=1)
user = await sender.send(query)

Integration with FastAPI

from fastapi import FastAPI, Depends
from livehop_minimal_mediator import Mediator, Sender

app = FastAPI()

# Create a global mediator instance
_mediator = Mediator()

# Register all your handlers
_mediator.register_handler(GetUsersQuery, lambda: GetUsersHandler())
_mediator.register_handler(CreateUserCommand, lambda: CreateUserHandler(db))

# Create dependency
def get_sender() -> Sender:
    return _mediator.create_sender()

# Use in routes
@app.get("/users")
async def get_users(sender: Sender = Depends(get_sender)):
    query = GetUsersQuery()
    users = await sender.send(query)
    return users

@app.post("/users")
async def create_user(name: str, email: str, sender: Sender = Depends(get_sender)):
    command = CreateUserCommand(name=name, email=email)
    user_id = await sender.send(command)
    return {"id": user_id}

Integration with Django

from django.http import JsonResponse
from livehop_minimal_mediator import Mediator

# In your settings or app config
mediator = Mediator()
mediator.register_handler(GetUsersQuery, lambda: GetUsersHandler())

# In your views
async def get_users_view(request):
    sender = mediator.create_sender()
    query = GetUsersQuery()
    users = await sender.send(query)
    return JsonResponse({"users": users})

Core Interfaces

IRequest[TResponse]

Marker interface for requests that return a response.

class MyQuery(IRequest[MyReturnType]):
    pass

IRequestHandler[TResponse]

Interface for handling requests.

class MyHandler(IRequestHandler[MyReturnType]):
    async def handle(self, request: MyQuery) -> MyReturnType:
        # Your logic here
        pass

Mediator

Manages registration and retrieval of request handlers.

mediator = Mediator()
mediator.register_handler(MyQuery, lambda: MyHandler())
sender = mediator.create_sender()

Sender

Dispatches requests to their registered handlers.

result = await sender.send(MyQuery())

Best Practices

1. Organize by Feature - Group requests and handlers together in feature modules
2. Use Dataclasses - Leverage dataclasses for clean, immutable request definitions
3. Async All the Way - Use async/await consistently throughout your handlers
4. Dependency Injection - Inject dependencies into handler constructors via factories
5. Single Responsibility - Keep handlers focused on one task

Why livehop-minimal-mediator?

• Simplicity: No complex pipeline behaviors or middleware - just requests and handlers
• Lightweight: No dependencies, minimal code footprint
• Pythonic: Idiomatic Python with type hints and async support
• Transparent: Simple, readable implementation you can understand and trust

Requirements

• Python 3.8 or later
• No external dependencies

License

MIT

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

If you encounter any issues or have questions, please file an issue on GitHub.