Metadata-Version: 2.4
Name: linkr
Version: 0.4.0
Summary: Async RPC framework with middleware support
License-Expression: MIT
License-File: LICENSE
Keywords: rpc,async,framework,rabbitmq,amqp,middleware
Author: Roman Koshel
Author-email: roma.koshel@gmail.com
Requires-Python: >=3.11,<4
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Topic :: System :: Distributed Computing
Provides-Extra: rabbitmq
Requires-Dist: pydantic (>=2.0)
Requires-Dist: rmqaio (>=0.18.0) ; extra == "rabbitmq"
Project-URL: BugTracker, https://github.com/levsh/linkr/issues
Project-URL: Homepage, https://github.com/levsh/linkr
Project-URL: Repository, https://github.com/levsh/linkr
Description-Content-Type: text/markdown

# linkr

Async messaging framework — RPC and publish/subscribe.

## Install

```
pip install linkr
```

## Quickstart

```python
from linkr import LocalTransport, App

transport = LocalTransport()
app = App(transport)

@app.method("add")
def add(x: int, y: int) -> int:
    return x + y

await app.init()
await app.consume()

result = await app.make("add", 2, 3).invoke()
print(result)  # 5

await app.close()
```

## Features

- Decorator-based handler registration
- Timeout, TTL, RTTL per call
- Fire-and-forget via publish()
- App-level middleware (`AppMiddleware`)
- Wire-level middleware (`WireMiddleware`) — compression, encryption
- Gzip compression via `GzipMiddleware`
- Dependency injection with `Depends[T]`
- Pydantic serialization
- JSON-RPC 2.0 support via `JsonRpcSerializer`
- Multi-serializer with auto-detection
- Local transport for in-process testing (`LocalTransport`)
- RabbitMQ transport

## App-level Middleware

```python
import logging

from typing import Any

from linkr import AppMiddleware, LocalTransport, App
from linkr.models import Request, Response


class LoggingMiddleware(AppMiddleware):
    async def dispatch_client(
        self,
        call_next,
        request: Request,
        *,
        kwds: dict[str, Any] | None = None,
    ) -> Response | None:
        logging.info("[%s] Calling %s", request.id, request.method)
        response = await call_next()
        if response:
            logging.info("[%s] Done", request.id)
        return response

    async def dispatch_server(
        self,
        call_next,
        request: Request,
        *,
        kwds: dict[str, Any] | None = None,
    ) -> Response | None:
        logging.info("[%s] Calling %s", request.id, request.method)
        response = await call_next()
        if response:
            logging.info("[%s] Done", request.id)
        return response


app = App(LocalTransport())
app.add_middleware(LoggingMiddleware())
```

## Wire-level Middleware

Compression, encryption and other wire transformations use `WireMiddleware`:

```python
from linkr.middleware.gzip import GzipMiddleware

app.add_middleware(GzipMiddleware())
```

Custom wire-level middleware inherits from `WireMiddleware` and works with raw bytes and wire headers:

```python
import gzip

from typing import Any

from linkr import WireMiddleware
from linkr.models import RawMessage, Request, Response


class CustomCompression(WireMiddleware):
    async def dispatch_client(
        self,
        call_next,
        request_raw_message: RawMessage,
        request: Request,
        *,
        kwds: dict[str, Any] | None = None,
    ) -> RawMessage | None:
        if len(request_raw_message.data) >= 1024:
            request_raw_message.data = gzip.compress(request_raw_message.data)
        raw_response = await call_next()
        if raw_response and raw_response.headers.get("content_encoding") == "gzip":
            raw_response.data = gzip.decompress(raw_response.data)
        return raw_response

    async def dispatch_server(
        self,
        call_next,
        request_raw_message: RawMessage,
        *,
        kwds: dict[str, Any] | None = None,
    ) -> tuple[RawMessage, Response] | tuple[None, None]:
        if request_raw_message.headers.get("content_encoding") == "gzip":
            request_raw_message.data = gzip.decompress(request_raw_message.data)
        result = await call_next()
        if result is None or result[0] is None:
            return None, None
        raw_response, response = result
        if len(raw_response.data) >= 1024:
            raw_response.data = gzip.compress(raw_response.data)
        return raw_response, response
```

## Dependency Injection

```python
from linkr import Depends, LocalTransport, App


class Database:
    def __init__(self, url: str) -> None:
        self.url = url


transport = LocalTransport()
async with App(transport) as app:
    app.dependencies.add_singleton(Database, lambda: Database("postgres://..."))

    @app.method("ping")
    def ping(db: Depends[Database]) -> str:
        return db.url

    await app.consume()
    result = await app.make("ping").invoke()
    print(result)  # postgres://...
```

## Error Handling

Type validation is enabled via ``validate_types=True``:

```python
from linkr import LocalTransport, App, RpcError

transport = LocalTransport()
async with App(transport) as app:
    @app.method("add", validate_types=True)
    def add(x: int, y: int) -> int:
        return x + y

    await app.consume()
    try:
        await app.make("add", x="not", y=3).invoke()
    except RpcError as e:
        print(e.error_code)     # ValidationError
        print(e.error_message)  # x: Input should be a valid integer
```

## Publish (Fire-and-Forget)

Send a message without waiting for a response:

```python
from linkr import LocalTransport, App

transport = LocalTransport()
async with App(transport) as app:
    req = app.make("event", text="hello")
    await app.publish(req)
```

## Transports

| Transport             | When to use                |
|-----------------------|----------------------------|
| LocalTransport        | Local dev, in-process app  |
| RmqTransport          | Production (RabbitMQ)      |

## License

MIT

