pymediate 0.1.1

A type-safe mediator pattern implementation for Python 3.12+

A type-safe request mediator for Python 3.12+

---

Features

• Type Safety: Full runtime validation with mypy support
• Zero Convention: No naming conventions - uses type inspection
• Async/Await Support: First-class async handlers and mediators via pymediate.aio
• DI Ready: Built-in dependency-injector integration
• Dataclass Friendly: Works seamlessly with @dataclass and Request[T] inheritance
• Well Tested: comprehensive test suite

Quick Example

from dataclasses import dataclass
from pymediate import Request, Handler, Mediator, Services

# Define response and request as pure dataclasses
@dataclass
class UserCreated:
    user_id: int
    username: str

@dataclass
class CreateUser(Request[UserCreated]):
    username: str
    email: str

# Handler automatically linked by type
class CreateUserHandler(Handler[CreateUser]):
    def __call__(self, req: CreateUser) -> UserCreated:
        return UserCreated(user_id=1, username=req.username)

# Set up and use
services = Services()
services.add(CreateUserHandler())
provider = services.provider()
mediator = Mediator(provider)

response = mediator.send(CreateUser(username="alice", email="[e-mail]"))
print(f"User {response.username} created with ID {response.user_id}")

Async Support

PyMediate provides first-class async/await support through the pymediate.aio package:

import asyncio
from dataclasses import dataclass
from pymediate import Request, Services
from pymediate.aio import Handler, Mediator

@dataclass
class UserCreated:
    user_id: int
    username: str

@dataclass
class CreateUser(Request[UserCreated]):
    username: str
    email: str

class CreateUserHandler(Handler[CreateUser]):
    async def __call__(self, req: CreateUser) -> UserCreated:
        # Perform async operations
        await asyncio.sleep(0.1)  # Simulate async database call
        return UserCreated(user_id=1, username=req.username)

async def main():
    services = Services()
    services.add(CreateUserHandler())
    provider = services.provider()
    mediator = Mediator(provider)

    response = await mediator.send(CreateUser(username="alice", email="[e-mail]"))
    print(f"User {response.username} created with ID {response.user_id}")

asyncio.run(main())

Key differences for async:

• Import from pymediate.aio instead of pymediate
• Handler's __call__ method must be async def
• Use await mediator.send(...) instead of mediator.send(...)
• Supports concurrent request handling with asyncio.gather()

Pipeline Behaviors

PyMediate supports pipeline behaviors (middleware) that automatically wrap request processing for cross-cutting concerns like logging, validation, caching, and more:

from pymediate import Request, PipelineBehavior

# Universal behavior - applies to all requests
class LoggingBehavior(PipelineBehavior[Request]):
    def __call__(self, request, next):
        print(f"Handling: {type(request).__name__}")
        response = next()
        print(f"Completed: {type(request).__name__}")
        return response

# Selective behavior - only applies to CreateUser requests
class ValidationBehavior(PipelineBehavior[CreateUser]):
    def __call__(self, request, next):
        # Validate before processing
        if not request.username:
            raise ValueError("Username is required")
        return next()

# Register behaviors and handlers
services = Services()
services.add(LoggingBehavior())       # Applied to ALL requests
services.add(ValidationBehavior())    # Only applied to CreateUser
services.add(CreateUserHandler())

mediator = Mediator(services.provider())

# Behaviors automatically wrap matching requests
response = mediator.send(CreateUser(username="alice", email="[e-mail]"))
# Output:
# Handling: CreateUser
# Completed: CreateUser

Behaviors can be universal (PipelineBehavior[Request]) or selective (PipelineBehavior[SpecificRequest]), applying only to matching request types or mixins. They are resolved per request, respecting DI container scopes (Transient, Scoped, Singleton). See the Pipeline Behaviors Guide for more examples.

Installation

# Core package
pip install pymediate

# With dependency injection support
pip install pymediate[di]

Documentation

📚 Full Documentation

• Quick Start
• User Guide
• Examples
• API Reference

Development

Quick Start

# Clone and install
git clone https://github.com/sina-al/pymediate.git
cd pymediate
uv sync --all-extras --group test

# Run tests
poe test

# Run all checks
poe check:all

# See all available tasks
poe

Available Commands

PyMediate uses Poe the Poet for task running. Run poe to see all commands, or check tasks.toml.

Note: uv sync alone only installs the default dev dependency group (ruff, mypy, poethepoet). Test dependencies (pytest and friends) live in the separate test group and won't be installed unless you pass --group test (or --all-groups) — otherwise poe test fails with Failed to spawn: pytest.

Requirements

• Python 3.12+
• Optional: dependency-injector>=4.41.0 for DI support

Contributing

Contributions are welcome! Check the docs for guidelines.

License

MIT License - see LICENSE for details.