Wiring an App

A real app swaps the memory backend for a shared service and runs the patterns behind a web framework. This page wires one provider, then installs the app into FastAPI and FastStream with one call.

One provider, one line

A provider owns the connection. Pass it to uses= and grelmicro registers a default component for every kind the provider serves:

from grelmicro import Grelmicro
from grelmicro.providers.redis import RedisProvider

redis = RedisProvider("redis://localhost:6379/0")

micro = Grelmicro(uses=[redis])

Now Lock, Cache, and RateLimiter all resolve the Redis backend with no extra wiring.

Warning

Keep connection URLs in environment variables, not inline like the example above. The Configuration page shows the deployment story.

Add patterns

Build the patterns you need and use them inside the app scope:

from grelmicro.coordination import Lock

lock = Lock("cart")

async with micro:
    async with lock:
        ...

The lock finds the registered Redis backend through the active app. No backend= argument needed.

FastAPI

Call micro.install(app). One call wires both pieces:

import logging
from contextlib import asynccontextmanager

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.coordination import (
    Coordination,
    LeaderElection,
    Lock,
    TaskLock,
)
from grelmicro.coordination.memory import MemoryLeaderElectionAdapter
from grelmicro.log import configure
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import CircuitBreaker, CircuitBreakerRegistry
from grelmicro.resilience.circuitbreaker.memory import (
    MemoryCircuitBreakerAdapter,
)
from grelmicro.task import Tasks

logger = logging.getLogger(__name__)

# === grelmicro ===
tasks = Tasks()
leader_election = LeaderElection("leader-election")
tasks.add_task(leader_election)

redis = RedisProvider("redis://localhost:6379/0")

micro = Grelmicro(
    uses=[
        Coordination(lock=redis.lock(), election=MemoryLeaderElectionAdapter()),
        CircuitBreakerRegistry(MemoryCircuitBreakerAdapter()),
        tasks,
    ]
)


# === FastAPI ===
@asynccontextmanager
async def lifespan(app):
    configure()
    yield


app = FastAPI(lifespan=lifespan)
micro.install(app)


# --- Circuit Breaker: protect calls to an unreliable service ---
cb = CircuitBreaker("my-service")


@app.get("/")
async def read_root():
    async with cb:
        return {"Hello": "World"}


# --- Distributed Lock: synchronize access to a shared resource ---
lock = Lock("shared-resource")


@app.get("/protected")
async def protected():
    async with lock:
        return {"status": "ok"}


# --- Interval Task: run locally on every worker ---
@tasks.every(seconds=5)
def heartbeat():
    logger.info("heartbeat")


# --- Distributed Task: run once per interval across all workers ---
@tasks.every(seconds=60, lock=TaskLock(lease_duration=300))
def cleanup():
    logger.info("cleanup")


# --- Leader-gated Task: only the leader executes ---
@tasks.every(seconds=10, leader=leader_election)
def leader_only_task():
    logger.info("leader task")

The lifecycle is always required. install always wires it: it opens micro once at startup and closes it at shutdown, so every component is ready before the first request. A lifespan you already pass to FastAPI(lifespan=...) keeps running, chained around micro.

The per-handler ambient binding is optional. install wires it by default, so patterns like Lock("cart") and RateLimiter.sliding_window(...) resolve their backends inside route handlers with no backend= argument. Pass ambient=False when your handlers always pass an explicit backend= and do not need it:

micro.install(app, ambient=False)

Always call install, never hand-wire the lifespan alone

A request handler runs in its own task, so it only resolves ambient backends when install adds the middleware. If you open async with micro: in a hand-written lifespan but forget install (or pass ambient=False), the app starts up healthy and then every ambient call raises OutOfContextError on the first request that hits it. install(ambient=False) warns at startup when ambient components are registered (it raises under Grelmicro(strict=True)), and you can assert the wiring in a test before it ships:

def test_ambient_binding_is_wired() -> None:
    assert micro.check_ambient_binding(app)

FastStream

The same call wires a FastStream app:

from faststream import FastStream
from faststream.redis import RedisBroker

from grelmicro import Grelmicro
from grelmicro.coordination import Lock

broker = RedisBroker("redis://localhost:6379/0")
micro = Grelmicro(uses=[...])

app = FastStream(broker)
micro.install(app)


@broker.subscriber("orders")
async def handle(order: dict) -> None:
    async with Lock("orders"):
        ...

install opens micro on startup, closes it after shutdown, and binds the app around each consumed message so patterns resolve inside subscribers. Pass ambient=False to skip the per-message binding.