livehop-minimal-mediator 1.1.0
A minimal, lightweight mediator implementation for Python applications supporting CQRS pattern with commands and queries
Requires Python: >=3.8
Keywords: mediator, cqrs, commands, queries, minimal, livehop

# 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

```bash
pip install livehop-minimal-mediator
```

## Quick Start

### 1. Create a Query/Command

```python
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

```python
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

```python
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="alice@example.com")
user_id = await sender.send(command)
```

### Query with Parameters

```python
@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

```python
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

```python
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.

```python
class MyQuery(IRequest[MyReturnType]):
    pass
```

### IRequestHandler[TResponse]

Interface for handling requests.

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

### Mediator

Manages registration and retrieval of request handlers.

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

### Sender

Dispatches requests to their registered handlers.

```python
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.