Metadata-Version: 2.4
Name: pymediate
Version: 0.1.1
Summary: A type-safe mediator pattern implementation for Python 3.12+
Keywords: mediator,cqrs,dependency-injection,type-safe,request-dispatch
Author: sina-al
Author-email: sina-al <27771210+sina-al@users.noreply.github.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: dependency-injector>=4.41.0 ; extra == 'di'
Requires-Python: >=3.12
Project-URL: Homepage, https://github.com/sina-al/pymediate
Project-URL: Documentation, https://sina-al.github.io/pymediate/
Project-URL: Repository, https://github.com/sina-al/pymediate
Project-URL: Issues, https://github.com/sina-al/pymediate/issues
Project-URL: Changelog, https://github.com/sina-al/pymediate/releases
Provides-Extra: di
Description-Content-Type: text/markdown

<p align="center">
  <img src="https://github.com/sina-al/pymediate/blob/main/assets/logo.svg?raw=true" alt="PyMediate logo" width="400"><br><br>
  <b>A type-safe request mediator for Python 3.12+</b><br><br>

  <!-- Badges -->
  <a href="https://github.com/sina-al/pymediate/actions/workflows/test.yml">
    <img src="https://github.com/sina-al/pymediate/actions/workflows/test.yml/badge.svg" alt="Tests">
  </a>
  <a href="https://github.com/sina-al/pymediate/actions/workflows/code-quality.yml">
    <img src="https://github.com/sina-al/pymediate/actions/workflows/code-quality.yml/badge.svg" alt="Code Quality">
  </a>
  <a href="https://www.python.org/downloads/">
    <img src="https://img.shields.io/badge/python-3.12+-blue.svg" alt="Python 3.12+">
  </a>
  <a href="https://badge.fury.io/py/pymediate">
    <img src="https://badge.fury.io/py/pymediate.svg" alt="PyPI version">
  </a>
  <a href="https://opensource.org/licenses/MIT">
    <img src="https://img.shields.io/badge/License-MIT-yellow.svg" alt="MIT License">
  </a>
</p>

---

## 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

```python
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="alice@example.com"))
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:

```python
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="alice@example.com"))
    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:

```python
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="alice@example.com"))
# 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](https://sina-al.github.io/pymediate/guide/pipeline-behaviors/) for more examples.

## Installation

```bash
# Core package
pip install pymediate

# With dependency injection support
pip install pymediate[di]
```

## Documentation

**[📚 Full Documentation](https://sina-al.github.io/pymediate/)**

- [Quick Start](https://sina-al.github.io/pymediate/getting-started/quick-start/)
- [User Guide](https://sina-al.github.io/pymediate/guide/requests-responses/)
- [Examples](https://sina-al.github.io/pymediate/examples/basic/)
- [API Reference](https://sina-al.github.io/pymediate/api/request/)

## Development

### Quick Start

```bash
# 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](https://poethepoet.natn.io/) for task running. Run `poe` to see all commands, or check [`tasks.toml`](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](LICENSE) for details.
