Wiring an App

A real app runs its patterns behind a web framework. This page wires one provider, then installs the app into FastAPI, FastStream, and Litestar 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, Outbox aside:

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.

To serve only some kinds, list the components instead of the provider:

from grelmicro import Grelmicro
from grelmicro.cache import Cache
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import RateLimiterComponent

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

micro = Grelmicro(uses=[Cache(redis), RateLimiterComponent(redis)])

A bare provider registers a default component for each kind nothing else claims. micro.describe() names the provider each one came from, such as "redis". grelmicro check marks its line (provider default), or (default of redis) when the line names another provider or none.

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:

fragment

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.

Declare patterns at module load, next to your routes. Lock("cart"), TTLCache(ttl=60) and CircuitBreaker("svc") carry no backend reference, so the same Lock runs on Redis in production and on memory in tests with no rewiring.

async with micro: opens everything in uses=[...] in dependency order and closes it in reverse. A Provider opens before the components that borrow it, and Outbox then Tasks open after everything else, so a first run finds every backend open. import grelmicro imports no vendor SDK. A provider such as grelmicro.providers.redis loads redis only when you import it.

Register something conditionally

A component that exists only for one backend or one environment stays inline. A None entry in uses= is skipped:

import os

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

health = HealthChecks(provider_checks="require")
redis = RedisProvider("redis://localhost:6379/0")

micro = Grelmicro(
    uses=[
        health,
        redis if os.getenv("STORE_BACKEND") == "redis" else None,
    ]
)

With STORE_BACKEND unset, the app registers the health checks alone. Set it to redis and the provider joins them, with no change to the shape of the list.

When the list is long enough to deserve its own function, annotate it with Usable. It names everything uses= accepts, which Component does not: a Provider is not a Component, and neither is a plain async context manager.

import os

from grelmicro import Grelmicro, Usable
from grelmicro.health import HealthChecks
from grelmicro.providers.redis import RedisProvider


def build_components() -> list[Usable]:
    components: list[Usable] = [HealthChecks(provider_checks="require")]
    if os.getenv("STORE_BACKEND") == "redis":
        components.append(RedisProvider("redis://localhost:6379/0"))
    return components


micro = Grelmicro(uses=build_components())

Usable names one item, not the list. Keep the conditional in an if and the annotation stays list[Usable]. A prebuilt list that carries its own None entries is list[Usable | None], which uses= accepts just the same.

micro.use(item) registers one item after construction and rejects None, because a single call can be guarded with if instead:

if os.getenv("STORE_BACKEND") == "redis":
    micro.use(RedisProvider())

A Provider fills the kinds nothing else claims

A Provider registers a default component for every kind it serves that no component claims. Explicit wins, the provider fills the rest. uses=[...] and micro.use(...) follow the same rule, in any order:

micro = Grelmicro(uses=[redis])
micro.use(Coordination(redis))  # wins coordination, as in uses=[redis, Coordination(redis)]

The defaults are registered when the app opens, or the first time you call micro.components, micro.providers, micro.describe(), micro.check_backends(), micro.install(app) or micro.check_ambient_binding(app). Reading micro.<kind> or calling micro.get(kind) registers them too, when a provider fills that kind. After that, a component of a kind a default holds raises ComponentAlreadyRegisteredError, and so does a provider that would fill such a kind. A provider that fills only free kinds registers its defaults at once. The error names what registered the defaults.

Outbox is the exception, and it is always written out. It carries the handlers your messages are delivered to and the relay that delivers them, so it belongs where those are declared. See Outbox.

Two bare providers and no component raise AmbiguousProviderError when the app opens or first reads its components, because neither can be the default for a kind they share. Register the components explicitly when you want two providers to serve different kinds.

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 LeaderElection, Lock
from grelmicro.log import configure
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import CircuitBreaker
from grelmicro.task import Tasks

logger = logging.getLogger(__name__)

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

# One line says where the shared state lives.
redis = RedisProvider("redis://localhost:6379/0")

micro = Grelmicro(uses=[redis, 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(interval=5)
def heartbeat():
    logger.info("heartbeat")


# --- Distributed Task: run once per interval across all workers ---
@tasks.every(interval=60, gate="claim")
def cleanup():
    logger.info("cleanup")


# --- Leader-gated Task: only the leader executes ---
@tasks.every(interval=10, gate=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)

install also wires what you registered. ErrorResponses() has grelmicro's rejections answered as problem details, and IdempotentRequests() adds the idempotency middleware and describes it in the OpenAPI schema. Each happens only because the component is in uses=[...]. The Frameworks page lists what every framework gets.

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)

A mounted sub-application does not fail loudly

Install every app that owns components, mounted ones included. A mount is an ordinary call in the same task, so the host's request scope is still bound inside the sub-application. A sub-application that forgot install therefore resolves against the host's components instead of raising:

host_micro = Grelmicro(uses=[Cache(host_backend)])
sub_micro = Grelmicro(uses=[Cache(sub_backend)])

host = FastAPI()
host_micro.install(host)

sub = FastAPI()  # install(sub) forgotten
host.mount("/sub", sub)

A write from sub lands in host_backend. sub_backend stays empty, and nothing reports it. Two applications that look isolated share one store.

This is the one case where a forgotten install does not raise OutOfContextError, because a binding is present, just the wrong one. Assert each app separately:

def test_every_app_is_wired() -> None:
    assert host_micro.check_ambient_binding(host)
    assert sub_micro.check_ambient_binding(sub)  # False when install is missing

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: Order) -> 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.

Litestar

The same call wires a Litestar app. Litestar builds its middleware stack when you construct the app, so call install after it:

from litestar import Litestar, get

from grelmicro import Grelmicro
from grelmicro.coordination import Lock

micro = Grelmicro(uses=[...])


@get("/carts")
async def carts() -> dict[str, str]:
    async with Lock("cart"):
        return {"status": "ok"}


app = Litestar(route_handlers=[carts])
micro.install(app)

Hooks and lifespan managers already passed to Litestar(...) keep running. The Frameworks page lists every framework install supports.

See what got wired

micro.describe() answers what the app is wired with: every component, the backend behind it, the provider it borrows, and the checks that passed or failed. Credential-like values are masked.

report = micro.describe()

assert report.ok
assert [component.kind for component in report.components] == ["cache"]

The same report runs from the command line, which exits non-zero when a check fails so CI can gate on it:

python -m grelmicro check app:micro

Environment: production

Components
  coordination/default    RedisLockAdapter, RedisScheduleAdapter <- redis
  cache/default           RedisCacheAdapter <- redis

Tasks
  primary         -                     -
  purge-sessions  every 60s             claim           RedisLockAdapter <- redis
  nightly-report  cron 0 3 * * * (UTC)  leader primary  RedisScheduleAdapter <- redis
  heartbeat       every 10s             none

Providers
  redis  (RedisProvider)  redis://user:***@localhost:6379/0
    env:      REDIS_*
    serves:   lock, readwritelock, leaderelection, schedule, cache, ratelimiter, circuitbreaker
    declines: outbox

Checks
  ok    backend-scope: every bound backend reaches as far as its component requires
  ok    ungated-cron: every cron task names its gate
  ok    gate-backend: every gated task resolves its claim backend
  ok    leader-not-running: every leader election a task is gated on is registered

The Tasks block lists every task the registered Tasks holds, one row each: name, schedule, gate, and the backend it claims through. The gate reads claim, leader with the election's name, lock for a TaskLock, or none when every worker runs it. A row with - is not a schedule, such as the LeaderElection itself. The block is left out when no task is registered. See Check what is scheduled.

The declines line is the one to read when a pattern is not wired the way you expected. A provider fills a default component only for the kinds it serves, so uses=[redis] leaves the outbox unwired and this is where that shows up.

Pass the web application to include the ambient binding check, which catches the forgotten install described above, mounted sub-applications included:

def test_every_app_is_wired() -> None:
    assert micro.describe(app).ok

Check the app from the command line

Name the web application with --app and the report adds one row per route, saying what each registered component does to it:

python -m grelmicro check app:micro --app app:app

Environment: production

Components
  cache/default                MemoryCacheAdapter
  cached_responses/default     CachedResponses
  idempotent_requests/default  IdempotentRequests

Endpoints
  POST   /orders    idempotent 86400s
  GET    /products  cache 60s

Checks
  FAIL  backend-scope: IdempotentRequests('default') rides Cache('default'), which is bound to MemoryCacheAdapter and provides scope 'process', but requires scope 'cluster'
  FAIL  backend-scope: IdempotentRequests('default') is bound to InProcessLock, which provides scope 'process', but requires scope 'cluster'
  warn  path-patterns: cached_responses/default names /product, which no route matches
  ok    ambient-binding: patterns resolve their backend inside request handlers
  ok    health-not-served: every registered HealthChecks has its endpoints served

A row reads cache 60s for a cached read, idempotent 86400s for a write that replays, authenticated with its scopes or anonymous when AuthenticatedRequests is registered, and - when no component acts on the route.

Check	What it reports	Status
backend-scope	A backend that reaches less far than its component needs, such as a memory cache behind idempotency on several replicas.	FAIL in staging and production, warn with no environment, ok in development and test.
path-patterns	An include= pattern that names no route, which turns its rule off without a word. Needs --app.	warn
ambient-binding	An app or mounted sub-app that micro.install(app) never saw. Needs --app.	FAIL, or warn for a framework it does not recognize
health-not-served	A registered HealthChecks whose endpoints nothing serves, once an app is installed.	warn
gate-backend	A gated task that resolves no backend to claim through, so every fire runs nothing.	warn. Fails in staging and production from 0.45.0.
leader-not-running	A task gated on a LeaderElection that no registered Tasks holds.	warn
routes-not-declared	An integration that lists no routes while a component acting per endpoint is registered, so no endpoint row is shown and no include= pattern is checked. Needs --app.	warn

The command exits 0 when no check fails, 1 when one does, and 2 when the target does not resolve, for example a misspelled module or one that raises while it imports. Run it in CI before a deployment:

- run: python -m grelmicro check app:micro --app app:app
  env:
    GREL_ENVIRONMENT: production

Next

Read the per-pattern pages for cache, coordination, scheduling, and resilience. When you deploy, the Configuration page shows how to tune every pattern with GREL_* environment variables.