flow-med

A high-performance, type-safe Mediator pattern implementation for Python 3.13+, built on top of flow-res and injector.

Features

• Result-Driven Development: Built-in support for flow-res Result types, making error handling explicit and type-safe.
• Auto-Registration: Handlers are automatically registered using Python 3.13's __init_subclass__ and generic introspection.
• Dependency Injection: Seamless integration with the injector library for robust dependency management.
• Native Async Support: Designed from the ground up for asyncio with AwaitableResult support for elegant method chaining.
• Strict Type Safety: Fully compatible with pyright and mypy using modern Python 3.13 type parameters.

Installation

pip install flow-med

Quick Start

1. Define Request and Result

from flow_res import Result
from flow_med import Request

class GetUserRequest(Request[Result[str, Exception]]):
    def __init__(self, user_id: int):
        self.user_id = user_id

2. Implement Handler

Handlers are automatically registered when defined. Use @override to ensure correct implementation.

from typing import override
from flow_res import Ok, Result
from flow_med import RequestHandler

class GetUserHandler(RequestHandler[GetUserRequest, Result[str, Exception]]):
    @override
    async def handle(self, request: GetUserRequest) -> Result[str, Exception]:
        # Logic to get user
        return Ok(f"User {request.user_id}")

3. Initialize and Send

import asyncio
from injector import Injector
from flow_med import Mediator

async def main():
    # Initialize with an Injector
    Mediator.initialize(Injector())

    # Send request and chain results using flow-res
    result = await (
        Mediator.send_async(GetUserRequest(user_id=1))
        .map(lambda name: f"Hello, {name}!")
        .unwrap()
    )
    
    print(result) # Hello, User 1!

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

Requirements

• Python 3.13 or higher.
• flow-res
• injector

License

MIT License