fastmesh 0.2.0

Asyncio message bus library for integration and event-driven systems

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

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

---

Quickstart

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:

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().

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.

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

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.

{
  "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"}}
}

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.

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.

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.

# 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

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.

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)

# 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

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

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