Metadata-Version: 2.4
Name: fastmesh
Version: 0.2.0
Summary: Asyncio message bus library for integration and event-driven systems
License: MIT
Requires-Python: >=3.11
Provides-Extra: dev
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: respx>=0.21; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Provides-Extra: docs
Requires-Dist: mkdocs-material>=9.5; extra == 'docs'
Requires-Dist: mkdocstrings[python]>=0.25; extra == 'docs'
Provides-Extra: http
Requires-Dist: httpx>=0.27; extra == 'http'
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.20; extra == 'otel'
Provides-Extra: pydantic
Requires-Dist: pydantic>=2; extra == 'pydantic'
Description-Content-Type: text/markdown

# FastMesh

Python asyncio message bus for integration and event-driven systems.

**Zero mandatory dependencies · SDK-first · Composable**

> Pre-release — API not yet stable.

---

## Features

- **Three integration paradigms** — fire-and-forget, request/reply, pub/sub on the same bus
- **Transform pipeline** — `Pick`, `Rename`, `Cast`, `Enrich`, `JsonMap` (declarative, hot-reloadable config)
- **Content routing** — `ContentRouter` with predicate-based first-match dispatch
- **Resilience plugins** — `Retry`, `RateLimit`, `DeadLetterQueue`, `CircuitBreaker`
- **Service Registry** — TOML/JSON config for endpoints, auth, and resilience policies
- **HTTP Orchestration** — sequential and parallel multi-step HTTP workflows with rollback (saga)
- **Decorator API** — `@flow.source`, `@flow.transform`, `@flow.sink`

---

## Install

```bash
pip install fastmesh            # core — zero mandatory dependencies
pip install "fastmesh[http]"    # + httpx for orchestration
```

---

## Quickstart

```python
import asyncio
from collections.abc import AsyncGenerator
from fastmesh import Bus, Message
from fastmesh.adapters.base import BaseSource, BaseSink

class NumberSource(BaseSource):
    async def start(self) -> AsyncGenerator[Message, None]:
        for i in range(5):
            yield Message(payload={"n": i})

class PrintSink(BaseSink):
    async def send(self, msg: Message) -> None:
        print(msg.payload)

asyncio.run(Bus().source(NumberSource()).sink(PrintSink()).run())
```

---

## Core concepts

### Bus

Central coordinator. Fluent builder API:

```python
bus = (
    Bus()
    .source(my_source)          # one or more sources (fan-in)
    .pipe(Pick("id", "type"))   # transform chain (applied in order)
    .sink(my_sink)              # one or more sinks (fan-out)
    .plugin(my_plugin)          # middleware (ingress forward, egress reversed)
    .on("order.created", fn)    # pub/sub handler
)
await bus.run()
```

### Message

Immutable frozen dataclass. Transforms return new instances via `.update()`.

```python
msg = Message(payload={"id": 1, "type": "order"})
msg2 = msg.update({"status": "confirmed"})   # new Message, original unchanged
print(msg.id)          # uuid4 string
print(msg.timestamp)   # UTC datetime
print(msg.metadata)    # dict for tracing/context
```

### Three integration paradigms

All three share the same transform pipeline and plugin middleware.

```python
bus = Bus().sink(my_sink).on("order.created", handler)

# Fire-and-forget — caller does not wait
await bus.publish(Message(payload={"type": "order.created", "id": 42}))

# Request/reply — caller suspends until sink calls msg.reply(result)
result = await bus.call(Message(payload={"product": "widget"}), timeout=5.0)

# Pub/sub — handler called when payload["type"] matches
bus.on("order.created", async def handler(msg): ...)
```

### Transform pipeline

```python
from fastmesh.transforms.core import Pick, Rename, Cast, Enrich

bus = (
    Bus()
    .pipe(Pick("id", "type", "amount"))          # drop all other fields
    .pipe(Rename({"amount": "total"}))            # rename key
    .pipe(Cast({"total": float}))                 # coerce type
    .pipe(Enrich({"source": lambda p: "web"}))    # add computed field
)
```

#### JsonMap — declarative structural transform

Define field mappings in a JSON file. Supports dotpath output, string join,
file-backed lookup tables with mtime-based hot reload, and list expansion.

```json
{
  "customer.id":   "user_id",
  "customer.name": {"join": ["first_name", "last_name"], "sep": " "},
  "customer.tier": {"from": "status_code", "lookup": "mappings/status.json", "default": "STANDARD"},
  "order.lines":   {"from": "products", "each": {"sku": "code", "qty": "quantity"}}
}
```

```python
from fastmesh.transforms.json_map import JsonMap

bus = Bus().pipe(JsonMap("mappings/customer.json")).sink(my_sink)
```

### Content routing

`ContentRouter` is itself a `BaseSink` — composable at the bus level.

```python
from fastmesh.router.content import ContentRouter

router = (
    ContentRouter()
    .when(lambda p: p.get("type") == "order",   order_sink)
    .when(lambda p: p.get("type") == "invoice", invoice_sink)
    .otherwise(error_sink)     # optional fallback; drops silently if omitted
)

# Fan-out: router receives all messages AND audit_sink receives all messages
bus = Bus().source(source).sink(router).sink(audit_sink)
```

First-match semantics — evaluation stops at the first matching predicate.

### Resilience plugins

Plugins wrap individual sinks as `BaseSink` subclasses. Composable in any order.

```python
from fastmesh.plugins.retry          import Retry
from fastmesh.plugins.rate_limit     import RateLimit
from fastmesh.plugins.dlq            import DeadLetterQueue, MemoryDLQStore
from fastmesh.plugins.circuit_breaker import CircuitBreaker

dlq = MemoryDLQStore()

bus = Bus().source(source).sink(
    DeadLetterQueue(
        CircuitBreaker(
            Retry(my_sink, max_retries=3, initial_delay=0.5),
            failure_threshold=5,
            recovery_timeout=30.0,
        ),
        store=dlq,
    )
)
# Exception flow: sink → Retry (retries) → CircuitBreaker (counts) → DLQ (captures)
```

| Plugin | Behaviour |
|---|---|
| `Retry` | Exponential backoff; re-raises after `max_retries` exhausted |
| `RateLimit` | Token bucket; backpressure (waits, never rejects) |
| `CircuitBreaker` | CLOSED → OPEN → HALF_OPEN state machine |
| `DeadLetterQueue` | Captures failures; swallows exception |

### Service Registry

Centralise endpoint URLs, auth, and resilience policies in a TOML file.
`${VAR}` and `${VAR:-default}` are resolved from environment variables at load time —
the same config file works across dev/staging/prod.

```toml
# services.toml

[services.payments]
base_url = "${PAYMENTS_URL:-https://pay.internal}"

[services.payments.auth]
type      = "bearer"
token_env = "PAYMENTS_TOKEN"

[services.payments.retry]
max_retries   = 3
initial_delay = 0.5

[services.payments.circuit_breaker]
failure_threshold = 5
recovery_timeout  = 30.0

[services.audit]
base_url  = "${AUDIT_URL:-https://audit.internal}"
[services.audit.rate_limit]
rate  = 500.0
burst = 1000
```

```python
from fastmesh.config import ServiceRegistry

registry = ServiceRegistry.load("services.toml")

# registry.service("payments") returns:
#   CircuitBreaker(Retry(HttpSink("https://pay.internal", headers={"Authorization": "Bearer …"})))

bus = (
    Bus()
    .source(source)
    .pipe(JsonMap("mappings/payment.json"))
    .sink(registry.service("payments"))   # full plugin stack from config
    .sink(registry.service("audit"))
)
```

Supported auth types: `bearer`, `basic`, `api_key`.

### HTTP Orchestration

Multi-step HTTP workflows with context chaining, parallel execution, and rollback.

```python
from fastmesh.orchestration import HttpOrchestrator, ParallelOrchestrator, Step
from fastmesh.orchestration.step import Rollback

# Sequential — step 2 uses the token extracted by step 1
orch = HttpOrchestrator(
    steps=[
        Step(
            name="auth",
            url="https://auth.example.com/token",
            method="POST",
            body={"client_id": "$.payload.client_id"},
            extract={"access_token": "$.token"},   # copy to context
        ),
        Step(
            name="order",
            url="https://orders.example.com/orders",
            method="POST",
            body={"token": "{access_token}", "product": "$.payload.product"},
            rollback=Rollback(url="https://orders.example.com/orders/pending", method="DELETE"),
        ),
    ],
    reply_key="order",   # returned by bus.call()
)

result = await Bus().connect(orch).call(Message(payload={...}), timeout=10.0)
```

```python
# Parallel — all steps run concurrently; a failing step yields None (non-fatal)
orch = ParallelOrchestrator(
    steps=[
        Step(name="profile",     url="https://api.example.com/users/42"),
        Step(name="account",     url="https://api.example.com/accounts/42"),
        Step(name="preferences", url="https://api.example.com/prefs/42"),
    ],
    reply_key="profile",
)
```

URL placeholders: `{VAR}` (context → env → error), `$.field.nested` (dotpath).

### Decorator API

```python
from fastmesh import flow

@flow.source
async def my_source():
    for i in range(10):
        yield Message(payload={"n": i})

@flow.transform
async def double(msg):
    return msg.update({"n": msg.payload["n"] * 2})

@flow.sink
async def print_sink(msg):
    print(msg.payload)

await Bus().source(my_source()).pipe(double()).sink(print_sink()).run()
```

---

## Examples

| File | Covers |
|---|---|
| `examples/minimal.py` | Bus, Source, Sink primitives |
| `examples/integration_patterns.py` | Fire-and-forget, request/reply, pub/sub |
| `examples/routing.py` | Pick + ContentRouter + fan-out |
| `examples/plugins.py` | Retry, DLQ, CircuitBreaker, composition |
| `examples/service_registry.py` | ServiceRegistry — TOML config, auth, plugin stack |
| `examples/orchestration.py` | HttpOrchestrator, ParallelOrchestrator, JsonMap, rollback |

---

## Architecture

```
fastmesh/
├── bus.py               # Bus — central coordinator
├── message.py           # Message — immutable frozen dataclass
├── channel.py           # Channel — asyncio.Queue with backpressure
├── decorators.py        # @flow.source / .transform / .sink
├── adapters/
│   ├── base.py          # BaseSource, BaseSink (ABC)
│   └── http.py          # HttpSource, HttpSink
├── transforms/
│   ├── base.py          # BaseTransform (ABC)
│   ├── core.py          # Pick, Rename, Cast, Enrich
│   └── json_map.py      # JsonMap, LookupTable
├── plugins/
│   ├── base.py          # BasePlugin (ABC)
│   ├── retry.py         # Retry
│   ├── rate_limit.py    # RateLimit
│   ├── dlq.py           # DeadLetterQueue, MemoryDLQStore
│   └── circuit_breaker.py  # CircuitBreaker
├── router/
│   └── content.py       # ContentRouter
├── config/
│   └── service_registry.py  # ServiceRegistry, ServiceConfig, AuthConfig
└── orchestration/
    ├── http_orchestrator.py    # HttpOrchestrator
    ├── parallel_orchestrator.py # ParallelOrchestrator
    ├── resolver.py              # UrlResolver
    └── step.py                  # Step, Rollback
```

Core has **zero mandatory dependencies**. Optional:

| Extra | Dependency | Enables |
|---|---|---|
| `fastmesh[http]` | `httpx` | `HttpOrchestrator`, `ParallelOrchestrator` |

---

## Development

```bash
git clone https://github.com/your-org/fastmesh
cd fastmesh
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

pytest                                           # all tests
pytest --cov=fastmesh --cov-report=term-missing  # with coverage (≥ 80% required)
ruff check fastmesh/                             # lint
ruff format fastmesh/                            # format
python examples/minimal.py                       # smoke test
```

CI runs on Python 3.11, 3.12, 3.13.

---

## License

MIT
