Metadata-Version: 2.4
Name: py-mediatR
Version: 6.8.0
Summary: .NET MediatR-inspired CQRS/Mediator for Python: async pipeline, notifications, streaming, DI (singleton/scoped/transient), CancellationToken, auto-discovery; CI tests on free-threaded Python 3.13t/3.14t.
Author: Ayhan Öztemel
License: MIT License
        
        Copyright (c) 2026 Ayhan Öztemel
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Project-URL: Homepage, https://github.com/AyhanOztemel/py_mediatR
Project-URL: Source, https://github.com/AyhanOztemel/py_mediatR
Project-URL: Issues, https://github.com/AyhanOztemel/py_mediatR/issues
Project-URL: Changelog, https://github.com/AyhanOztemel/py_mediatR/blob/main/CHANGELOG.md
Keywords: mediatr,cqrs,mediator,fastapi,pipeline,di,dependency-injection,cancellation,nogil,free-threaded
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: Free Threading :: 2 - Beta
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: pydantic
Requires-Dist: pydantic>=1.10; extra == "pydantic"
Provides-Extra: test
Requires-Dist: pytest>=7; extra == "test"
Requires-Dist: pytest-asyncio>=0.23; extra == "test"
Provides-Extra: typecheck
Requires-Dist: mypy==2.3.1; extra == "typecheck"
Provides-Extra: coverage
Requires-Dist: pytest>=7; extra == "coverage"
Requires-Dist: pytest-asyncio>=0.23; extra == "coverage"
Requires-Dist: coverage[toml]>=7; extra == "coverage"
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: build; extra == "dev"
Requires-Dist: twine; extra == "dev"
Dynamic: license-file

# py_mediatR

[![PyPI](https://img.shields.io/pypi/v/py-mediatR.svg)](https://pypi.org/project/py-mediatR/)
[![Python](https://img.shields.io/pypi/pyversions/py-mediatR.svg)](https://pypi.org/project/py-mediatR/)
[![CI](https://github.com/AyhanOztemel/py_mediatR/actions/workflows/ci.yml/badge.svg)](https://github.com/AyhanOztemel/py_mediatR/actions/workflows/ci.yml)
[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](https://github.com/AyhanOztemel/py_mediatR/blob/main/LICENSE)

> Türkçe sürüm: [README.tr.md](https://github.com/AyhanOztemel/py_mediatR/blob/main/README.tr.md)

**CQRS / Mediator for Python**, inspired by
[.NET MediatR](https://github.com/jbogard/MediatR). Your code sends a small
request object; the mediator finds its handler and runs it. The caller never
imports or constructs the handler. Zero dependencies; tested on CPython 3.10 to
3.14, including the free-threaded 3.13t/3.14t builds.

## Install

```bash
pip install py-mediatR              # core (zero dependencies)
pip install "py-mediatR[pydantic]"  # optional pydantic model coercion
```

## Quick start

**1. A request and its handler** (`application/products/list_products.py`):

```python
from dataclasses import dataclass
from py_mediatR import IRequest
from domain.product_repository import IProductRepository

@dataclass
class ListProducts(IRequest):
    pass

class ListProductsHandler:                       # found automatically
    def __init__(self, repository: IProductRepository):
        self.repository = repository             # given by the container

    def handle(self, request: ListProducts) -> list[str]:
        return self.repository.list_products()
```

**2. `register/register.py`** - which class serves each contract, how long it lives:

```python
from domain.product_repository import IProductRepository
from infrastructure.memory_product_repository import MemoryProductRepository

def register(container):
    container.register_singleton(IProductRepository, MemoryProductRepository)
```

**3. `bootstrap/bootstrap.py`** - container, registrations and the mediator:

```python
from py_mediatR import Mediator, ServiceContainer
from register import register

def bootstrap() -> Mediator:
    container = ServiceContainer()
    register(container)
    return Mediator(scan_paths=["application"], handler_factory=container,
                    resolve_discovered_from_container=True)
```

**4. Use it anywhere** - a console program, a FastAPI/Flask route, a Django view:

```python
mediator = bootstrap()
mediator.send(ListProducts())        # ['Notebook', 'Pencil']
```

Complete, runnable projects for console, FastAPI, Flask and Django:
[examples/](https://github.com/AyhanOztemel/py_mediatR/blob/main/examples/README.md).

The smallest possible use, without discovery or a container:

```python
from py_mediatR import Mediator, IRequest

class Ping(IRequest):
    pass

class PingHandler:
    def handle(self, req: Ping) -> str:
        return "pong"

mediator = Mediator(auto_discover=False)
mediator.register_handler(Ping, PingHandler())
print(mediator.send(Ping()))  # pong
```

## How a handler is found

There is **no naming convention**. A class becomes a handler because of the
*type annotation* on its `handle()` parameter - the class name is irrelevant:

```python
class GetUser(IRequest):
    user_id: str

class ThisNameDoesNotMatter:          # not "GetUserHandler" - still found
    def handle(self, req: GetUser):   # <- the annotation is the contract
        return {"user_id": req.user_id}
```

| Kind | Detected by |
|---|---|
| Request handler | `handle()` has a parameter annotated with an `IRequest` subclass |
| Stream handler | same, but the annotation is an `IStreamRequest` subclass |
| Notification handler | `handle()` has a parameter annotated with an `INotification` subclass |

Details that bite:

- The annotation must be **resolvable** at import time. A wrong or unimportable
  forward reference makes the class invisible rather than raising.
- Exactly **one** `IRequest` parameter is allowed; two raises `TypeError`.
- Extra parameters are fine - `cancellation_token` is injected by name.
- A handler whose `__init__` takes arguments is registered *deferred* and
  resolved through `handler_factory` (your `ServiceContainer`) at dispatch time.
- One handler per request type; notifications may have many.

**Where discovery looks.** `Mediator()` scans the project folder for `*.py`
files. The project folder is found by walking up from the file that creates
the Mediator to the nearest folder with `requirements.txt`, `pyproject.toml`,
`setup.py`, `Pipfile`, `.git`, `main.py`, `app.py` or `manage.py` - so
`python main.py`, `uvicorn main:app` and test runners find the same project.
Pass `project_root=` to choose it yourself and `scan_paths=[...]` to narrow it.
Folders named `venv`, `.venv`, `env`, `site-packages`, `__pycache__`, `.git`,
`node_modules`, `.tox`, `.nox`, `.eggs`, `build`, `dist`, `migrations` and the
`.*_cache` folders are skipped **inside** the project. Import errors are logged
as warnings on the `mediatr.discovery` logger unless
`MEDIATR_DISCOVERY_WARNINGS=0`.

**Response coercion - the one place naming matters.** If a handler returns a
`dict`, py_mediatR coerces it into a response class found **by name in the same
module**: `GetUser` -> `GetUserResponse` (or `GetUserRequest` ->
`GetUserResponse`). An explicit `-> GetUserResponse` return annotation takes
priority. With neither, the `dict` is returned as-is.

## Registering services: `ServiceContainer`

| Call | Lifetime | Use it for |
|---|---|---|
| `container.register_singleton(IClock, SystemClock)` | one object for the whole application | configuration, clocks, shared clients |
| `container.register_scoped(IRepo, SqlRepo)` | one object per scope (for example a web request) | database sessions, unit of work |
| `container.register_transient(Service)` | a new object every time | stateless services |

The second argument creates the object: a class, or a function that takes no
argument or takes the container. Handlers and services receive their
dependencies through constructor type hints (`Optional[T]`, `T | None` and
`Annotated[...]` included). A singleton that would capture a scoped service is
rejected with `DIResolutionError`, as in .NET. This check is always on since
6.7.x - there is no `validate_scopes` switch - so a dependency graph accepted by
an older version can be rejected after upgrading.

## Frameworks

Every example is a complete project with the same layers and its own README
(English and Turkish).

| Framework | How the mediator is used | Example |
|---|---|---|
| Console | `mediator.send(...)` | [examples/console](examples/console) |
| FastAPI | `mediator.send(...)` in the endpoints | [examples/fastapi](examples/fastapi) |
| Flask | `mediator.send(...)` in the views | [examples/flask](examples/flask) |
| Django | `mediator.send(...)` in the views | [examples/django](examples/django) |

For one scope per HTTP request (a database session shared by the handlers of
one request), see *Per-request scope* below.

## Optional features

None of these is needed for the setup above. Each one solves a specific
problem; use it only when you have that problem.

### Async dispatch

**Why:** your handlers or behaviors are `async def`.
**How:** `await mediator.send_async(request)`, `await mediator.publish_async(event)`.
Sync and async handlers can be mixed; everything runs on the caller's event loop.

### Pipeline behaviors

**Why:** logging, validation, caching, retries or transactions around many
handlers, without repeating that code in each handler.
**Benefit:** one place for each cross-cutting concern.
**How:** every `send()` runs through the same onion. Behaviors are sorted by
`order`, **smallest first, outermost**:

```
send(request)
└─ behavior (order -100)          <- outermost
   └─ behavior (order -50)
      ├─ pre-processor            <- side effects only
      ├─ HANDLER                  <- your business logic
      └─ post-processor           <- sees the response, cannot change it
```

Eight behaviors are built in. Each reads an attribute off the **request**; if
it is absent the behavior is a no-op, so a request opts in to what it needs:

| Behavior | `order` | Opt-in on the request | Effect |
|---|---|---|---|
| `LoggingBehavior` | -100 | - (always) | logs entry/exit |
| `TracingBehavior` | -95 | - (always) | OpenTelemetry span per request (no-op without OpenTelemetry) |
| `PerformanceBehavior` | -90 | - (always) | warns past a duration threshold |
| `AuthorizationBehavior` | -85 | `requires_permission` | denies with `UnauthorizedError`; handler never runs |
| `ValidationBehavior` | -80 | `validate()` method | raises before the handler |
| `CachingBehavior` | -70 | `cacheable = True` | TTL + LRU; a hit skips the handler entirely |
| `RetryBehavior` | -60 | - (always) | re-runs on exception, exponential backoff + jitter |
| `TransactionBehavior` | -50 | `transactional = True` | commit on success, rollback on error |

```python
from dataclasses import dataclass
from py_mediatR import (IRequest, Mediator, LoggingBehavior, ValidationBehavior,
                        CachingBehavior, RetryBehavior, TransactionBehavior,
                        AuthorizationBehavior)

@dataclass
class PlaceOrder(IRequest):
    sku: str
    qty: int = 1
    transactional = True                  # TransactionBehavior engages
    requires_permission = "orders.write"  # AuthorizationBehavior checks this

    def validate(self) -> None:           # ValidationBehavior calls this
        if self.qty < 1:
            raise ValueError("qty must be positive")

mediator = Mediator(behaviors=[
    LoggingBehavior(),
    AuthorizationBehavior(lambda req, perm: current_user.has(perm)),
    ValidationBehavior(),
    CachingBehavior(ttl_seconds=60, max_size=1000),
    RetryBehavior(max_attempts=3, delay=0.1, backoff=2.0, jitter=0.05),
    TransactionBehavior(session_factory=lambda: SessionLocal()),
])
```

Your own behavior: `next_handler()` runs the rest of the chain; not calling it
short-circuits the handler (that is how `CachingBehavior` serves a hit).

```python
from py_mediatR import IPipelineBehavior

class AuditBehavior(IPipelineBehavior):
    order = -110                 # lower than LoggingBehavior -> runs outermost
    applies_to = PlaceOrder      # optional; None (default) means every request

    def handle(self, request, next_handler):
        audit.write(f"-> {type(request).__name__}")
        response = next_handler()
        audit.write(f"<- {type(request).__name__}")
        return response
```

In the async pipeline `handle()` may be `async def`, and `next_handler()`
returns an awaitable. Behavior **instances** are shared across dispatches;
mutable state in them needs its own concurrency control.

### Per-request scope

**Why:** the handlers of one HTTP request should share one database session,
and it should be closed when the request ends.
**How:** register the session as scoped and dispatch through a scope:

```python
from py_mediatR import scoped_mediator

container.register_scoped(DbSession)
mediator = Mediator(handler_factory=container)
with scoped_mediator(mediator, container) as m:
    m.send(CreateUser(name="Ada"))          # handlers share the scoped session
# scope disposed -> session.close()
```

FastAPI has a ready dependency for this:

```python
from py_mediatR import make_fastapi_mediator_dependency
get_mediator = make_fastapi_mediator_dependency(mediator, container)

@app.post("/orders")
async def create_order(cmd: CreateOrderDto, m=Depends(get_mediator)):
    return await m.send_async(CreateOrder(**cmd.model_dump()))
```

Behaviors given as a **class** (instead of an instance) are created for every
dispatch and, through a scope, receive scoped dependencies too.

### Pre/post processors

**Why:** a side effect before or after every handler (audit, metrics).
**How:** both support `order` and `applies_to`; a post-processor *sees* the
response but cannot replace it.

```python
from py_mediatR import IRequestPreProcessor, IRequestPostProcessor

class AuditPre(IRequestPreProcessor):
    def process(self, request):                 # may be async def
        audit.write(type(request).__name__)

class MetricsPost(IRequestPostProcessor):
    def process(self, request, response):       # may be async def
        metrics.increment(type(request).__name__)
```

### Exception handlers and actions

**Why:** turn a known error into a fallback response, or observe every error.

| | Purpose | Return value |
|---|---|---|
| `IExceptionHandler.handle` | replace the error with a fallback response | call `state.set_handled(response)`; otherwise the exception propagates |
| `IExceptionAction.execute` | observe only (log, alert, metric) | ignored - the exception keeps propagating |

```python
from py_mediatR import IExceptionHandler, IExceptionAction

class InvoiceNotFound(IExceptionHandler):
    exception_type = KeyError
    applies_to = FindInvoice
    def handle(self, request, exc, state):
        state.set_handled(FindInvoiceResponse(invoice_id=request.invoice_id, found=False))

class AlertAction(IExceptionAction):
    exception_type = Exception
    def execute(self, request, exc):
        alerting.notify(f"{type(request).__name__} failed: {exc}")
```

Both filter on `exception_type`, `applies_to` and `order`. All matching actions
run; the first matching handler that marks the error handled wins. A
two-parameter `handle(request, exc)` that returns a non-`None` value also counts
as handled.

### Validators

**Why:** validation that needs dependencies, outside the request class
(FluentValidation style).

```python
from py_mediatR import IValidator, ValidationBehavior

class CreateUserValidator(IValidator):
    applies_to = CreateUser          # or a tuple of types; None = all
    def validate(self, request):
        if "@" not in request.email:
            raise ValueError(f"invalid e-mail: {request.email}")

mediator = Mediator(behaviors=[ValidationBehavior(validators=[CreateUserValidator()])])
```

Matching validators run in `order`, then the request's own `validate()`. Async
validators are awaited in `send_async()` and rejected in `send()`.

### Notifications

**Why:** one event, several independent reactions (send an e-mail, update a
report...).

```python
from py_mediatR import PublishStrategy

mediator.publish(UserRegistered(user_id="U-1"))                       # SEQUENTIAL
await mediator.publish_async(evt, strategy=PublishStrategy.PARALLEL_WHENALL)
await mediator.publish_async(evt, strategy=PublishStrategy.PARALLEL_NOWAIT)
```

| Strategy | Semantics |
|---|---|
| `SEQUENTIAL` (default) | one after another, in `order`; first error propagates |
| `PARALLEL_WHENALL` | concurrent, waits for all |
| `PARALLEL_NOWAIT` | fire-and-forget; tasks are kept referenced so they cannot be garbage-collected |

Subscriber errors propagate by default; `Mediator(swallow_notification_errors=True)`
logs them instead. With `polymorphic_publish=True`, publishing a subclass also
triggers base-type subscribers.

### Streaming

**Why:** produce many items one by one (`IAsyncEnumerable<T>` in .NET).

```python
from py_mediatR import IStreamRequest, IStreamPipelineBehavior

class StreamOrders(IStreamRequest):
    count: int = 100

class StreamOrdersHandler:
    async def handle(self, req: StreamOrders):
        for i in range(req.count):
            yield {"seq": i}

async for order in mediator.create_stream(StreamOrders(count=10)):
    ...
```

`IStreamPipelineBehavior` wraps the generator like a pipeline behavior.

### Cancellation

**Why:** stop a slow request (timeout, user cancelled).

```python
from py_mediatR import CancellationTokenSource, OperationCancelledError

cts = CancellationTokenSource(cancel_after=2.0)   # timeout
try:
    result = await mediator.send_async(SlowQuery(), cancellation_token=cts.token)
except OperationCancelledError:
    ...
```

A handler that declares a `cancellation_token` parameter receives the token.
Linked tokens (`CancellationTokenSource.create_linked(...)`) are supported.

### See the call chain

**Why:** "why didn't my handler run?" - a mediator hides who calls what.

```python
with mediator.trace() as flow:
    mediator.send(SearchProducts(keyword="laptop"))
    mediator.send(SearchProducts(keyword="laptop"))   # served from cache
print(flow.render())
```

```
send(SearchProducts)   (0.35 ms)
└─ behavior: LoggingBehavior   (0.29 ms)
   └─ behavior: CachingBehavior   [cache miss]   (0.20 ms)
      └─ HANDLER: SearchProductsHandler   (0.07 ms)
send(SearchProducts)   (0.14 ms)
└─ behavior: LoggingBehavior   (0.11 ms)
   └─ behavior: CachingBehavior   [CACHE HIT - handler NOT called]
```

Tracing is off by default. `flow.steps()` and `flow.find(label)` expose the same
data for assertions; `render(unicode=False)` forces ASCII.

### Strict mode

**Why:** turn silent fallbacks into errors while developing.

```python
mediator = Mediator(strict_async=True, strict_discovery=True, strict_response=True)
```

- `strict_async` - an awaitable returned through synchronous `send()`/`publish()`
  raises `TypeError` naming the component (otherwise a WARNING on `mediatr.dispatch`).
- `strict_discovery` - all duplicate-handler and import problems of the scan
  raise one `DiscoveryError` (a `ValueError`); otherwise they are WARNING logs.
- `strict_response` - a failing dict -> dataclass/Pydantic response conversion
  re-raises (otherwise a WARNING on `mediatr.response` and the raw result).

### Explicit registration

**Why:** large codebases where scanning is not wanted.

```python
mediator = Mediator(auto_discover=False)
mediator.register_handler(GetUser, GetUserHandler())
```

or mark classes and let discovery collect only them:

```python
from py_mediatR import handler

@handler
class GetUserHandler:
    def handle(self, req: GetUser): ...
```

## Notes

- **Transactions:** a failing `rollback()`/`close()` is logged and attached to
  the original exception (`add_note`), or raised as `TransactionCleanupError`
  with `raise_on_cleanup_failure=True`.
- **Sync-over-async bridge:** waits at most 30 s (`MEDIATR_SYNC_BRIDGE_TIMEOUT`,
  `<=0` disables) and raises `SyncBridgeTimeoutError` instead of freezing.
- **Debug output:** `MEDIATR_DEBUG=1` prints discovery details; `flow.print()`
  writes to stdout. Library diagnostics otherwise use logging.
- **Known limits:** behavior instances are shared across dispatches and scope
  clones; stream cancellation and early-close cleanup through nested stream
  wrappers are limited; strict async checks cover `send`/`publish`, not streams.
  Scope cleanup failures emit both a `RuntimeWarning` and a WARNING log.
- **Imports:** `import py_mediatR` (also `import py_mediatr`). Everything in
  `__all__` is re-exported at the top level; the submodules (`contracts`,
  `mediator`, `di`, `discovery`, `behaviors`, `cancellation`, `tracing`,
  `coercion`) are an implementation detail.
- **Changes per version:** [CHANGELOG.md](https://github.com/AyhanOztemel/py_mediatR/blob/main/CHANGELOG.md).

## Test quality

The suite has 444 tests and runs on CPython 3.10, 3.11, 3.14 and the
free-threaded 3.14t build (GIL disabled). The whole package was mutation tested
with cosmic-ray: 2406 mutants, **100%** of those that can change behaviour
killed (92.5% of all; the rest are proven equivalent, each class with its
reason). Report: [docs/MUTATION_REPORT.md](https://github.com/AyhanOztemel/py_mediatR/blob/main/docs/MUTATION_REPORT.md).

## Examples

- [examples/](https://github.com/AyhanOztemel/py_mediatR/blob/main/examples/README.md)
  - four small, self-contained projects (console, FastAPI, Flask, Django). Start here.
- [examples/advanced/](https://github.com/AyhanOztemel/py_mediatR/blob/main/examples/advanced/README.md)
  - a bigger e-commerce demo that uses every feature, with a ten-step console
  walkthrough that prints each call chain.

The examples are not shipped in the wheel; they live in the repository.

## License

MIT - Ayhan Öztemel
