App

Start here: Wiring an App
The frameworks: Frameworks

Grelmicro is the container. uses=[...] registers what the service runs on, async with micro: opens it, and micro.install(app) wires the framework you picked.

GrelmicroMiddleware is the same per-request binding as pure ASGI, for a framework install does not know.

grelmicro

grelmicro is a lightweight framework/toolkit which is ideal for building async microservices in Python.

Usable

Usable = (
    AbstractAsyncContextManager[object]
    | type[AbstractAsyncContextManager[object]]
)

One item Grelmicro(uses=[...]), Bulkhead(uses=[...]), or micro.use() accepts.

Covers Component instances, Provider instances, first-party backends, the bare classes of any of those, and plain async context managers. Name it to annotate a list you build before passing it in:

from grelmicro import Grelmicro, Usable

components: list[Usable] = [Log(), health]
if settings.store_backend == "redis":
    components.append(RedisProvider())

micro = Grelmicro(uses=components)

Component names a narrower set. It excludes Provider instances and plain async context managers, both of which uses= accepts.

A uses= list also takes None entries and skips them, which this alias does not cover, since micro.use(None) is an error. Annotate a prebuilt list that carries its own conditionals as list[Usable | None].

Grelmicro

Grelmicro(
    *,
    uses: Iterable[Usable | None] | None = None,
    environment: Environment | None = None,
    strict: bool = False,
    allow_multiple: bool = False,
)

The grelmicro application container.

A Grelmicro is the user-owned root that holds every item attached to the app (components, task managers, custom async context managers, ...) and opens them as a single async context manager. Two Grelmicro instances in the same process are fully independent.

The conventional variable name is micro:

from grelmicro import Grelmicro
from grelmicro.coordination import Coordination
from grelmicro.task import Tasks

tasks = Tasks()

micro = Grelmicro(uses=[
    Coordination(lock=RedisLockAdapter()),
    tasks,
])

@tasks.every(seconds=5)
async def cleanup(): ...

async with micro:
    await asyncio.sleep(60)

Inside the async with micro: block, primitives that omit an explicit micro= argument resolve through Grelmicro.current() (per asyncio task).

opened `property`

opened: bool

Whether every registered item is open.

False until async with micro: has entered the last one, and False again once the block is left. Read it from something that answers before the app has finished starting, a probe endpoint above all: a component registered before another one runs while that other one is still connecting.

environment `property`

environment: Environment | None

The declared deployment tier, or None when nothing declares one.

Resolved once at construction from Grelmicro(environment=...), then GREL_ENVIRONMENT. A value naming no tier reads as None.

components `property`

components: tuple[Component, ...]

Registered Component instances in registration order.

Plain async context managers passed to use(...) are not included. Useful for /healthz-style introspection that prints what is wired up on the running app.

providers `property`

providers: tuple[Provider, ...]

Active Provider instances, in registration order, deduped by identity.

Includes Providers listed in uses= and those adopted from Components that borrow them (see _discover_shared_providers). One Provider feeding several Components appears once. Useful for introspection and for HealthChecks(auto_health=True), which registers a provider:{short_name} check per entry.

coordination `property`

coordination: Coordination

The registered Coordination component.

Resolves the default-named entry, or the sole entry of kind coordination.

cache `property`

cache: Cache

The registered Cache component (default-named, or sole entry of kind cache).

log `property`

log: Log

The registered Log component (default-named, or sole entry of kind log).

trace `property`

trace: Trace

The registered Trace component (default-named, or sole entry of kind trace).

metrics `property`

metrics: Metrics

The registered Metrics component (default-named, or sole entry of kind metrics).

health `property`

health: HealthChecks

The registered HealthChecks component.

Resolves the default-named entry, or the sole entry of kind health.

outbox `property`

outbox: Outbox

The registered Outbox component (default-named, or sole entry of kind outbox).

ratelimiter `property`

ratelimiter: RateLimiterComponent

The registered RateLimiterComponent.

Resolves the default-named entry, or the sole entry of kind ratelimiter.

circuitbreaker `property`

circuitbreaker: CircuitBreakerComponent

The registered CircuitBreakerComponent.

Resolves the default-named entry, or the sole entry of kind circuitbreaker.

check_backends

check_backends(
    environment: Environment = "production",
) -> None

Check every bound backend against what its component requires.

Asks the question the deployed app will ask, from a process that declares something else, so a test catches the wiring before a pod does:

def test_backends_hold_across_replicas() -> None:
    micro.check_backends()

PARAMETER	DESCRIPTION
`environment`	The tier to answer for. Defaults to `"production"`, so the answer holds for the deployment rather than for the process running the check. TYPE: `Environment` DEFAULT: `'production'`

RAISES	DESCRIPTION
`BackendScopeError`	If any bound backend reaches less far than its component requires, naming every one of them.

current `classmethod`

current() -> Grelmicro

Return the active Grelmicro app for the current asyncio task.

Use inside an async with micro: block to look up the active app:

from grelmicro import Grelmicro

micro = Grelmicro.current()

The lookup is per asyncio task, so concurrent tasks each see their own active Grelmicro.

RAISES	DESCRIPTION
`NoActiveAppError`	If called outside any `async with micro:` block in the current task scope.

use

use(item: Usable) -> None

Register an item to be lifecycled with the app.

Four shapes are accepted:

A Component instance: registered with (kind, name) lookup and exposed on micro.<kind>.
A first-party backend (e.g. RedisLockAdapter): auto-wrapped into the matching Component (Coordination for lock backends, Cache for cache backends) before registration.
A Provider (e.g. RedisProvider): always lifecycled. On an app with no Components, it also registers one default Component per kind it serves, outbox aside. Once any Component is registered, the Provider is lifecycle-only.
Any other async context manager: just lifecycled with the app, the caller keeps the reference.

A bare class (no parens) is instantiated first, in the spirit of FastAPI's Depends(dep), so use(MemoryLockAdapter) matches use(MemoryLockAdapter()).

# Auto-wrapped first-party backend
micro.use(RedisLockAdapter())          # registered as (coordination, default)
micro.use(RedisCacheAdapter())         # registered as (cache, default)

# Provider on an empty app: default Component per served kind
micro.use(RedisProvider("redis://localhost"))

# Explicit Component when a non-default name is needed
micro.use(Coordination(lock=RedisLockAdapter(), name="analytics"))

# Plain async context manager: lifecycled only, caller holds reference
tasks = Tasks()
micro.use(tasks)

Returns None. Mirrors FastAPI's app.include_router(router) pattern: pure side-effect registration. To access registered components, use the typed micro.coordination / micro.cache properties or micro.get(kind, name). For plain async context managers, the caller already holds the reference.

PARAMETER DESCRIPTION

item

The item to register and lifecycle with the app. A Component instance is indexed under (kind, name) and exposed on micro.<kind>. A first-party backend is auto-wrapped into its matching Component. A Provider on an app with no Components registers one default Component per kind it serves, outbox aside. A zero-arg class is instantiated first. Any other async context manager is just lifecycled, and the caller keeps the reference.

TYPE: Usable

RAISES	DESCRIPTION
`ComponentAlreadyRegisteredError`	A different component is already registered under the same `(kind, name)` key. Plain async context managers do not raise. They are appended.
`TypeError`	If `item` is `None`. `Grelmicro(uses=[...])` skips a `None` entry, a single call does not.

get

get(
    kind: type[ComponentT], name: str = "default"
) -> ComponentT

get(kind: str, name: str = 'default') -> Any

get(
    kind: str | type[Component], name: str = "default"
) -> Any

Resolve a registered component by (kind, name).

Pass the class to keep the type through resolution:

cache = micro.get(Cache)                          # -> Cache
limiter = micro.get(RateLimiterComponent, "api")  # -> RateLimiterComponent

Pass the kind string for a component grelmicro does not define, such as one a third-party package registers. That form returns Any, because the registration is dynamic and cannot be typed without a global registry:

mailer = micro.get("mailer")                      # -> Any

PARAMETER	DESCRIPTION
`kind`	The component class to resolve, such as `Cache` or `RateLimiterComponent`, which keeps the return type. Or the `kind` string on the registered component (`"coordination"`, `"cache"`, `"ratelimiter"`, `"circuitbreaker"`, `"log"`, `"trace"`, `"metrics"`, `"health"`), which returns `Any` and also resolves third-party kinds. TYPE: `str \| type[Component]`
`name`	Component instance name. `"default"` matches the entry that also backs `micro.<kind>`. Pass the explicit name to resolve a secondary registration such as `Coordination(lock=backend, name="analytics")`. TYPE: `str` DEFAULT: `'default'`

RAISES	DESCRIPTION
`ComponentNotRegisteredError`	If no component matches.

fake `async`

fake() -> AsyncIterator[None]

Swap every backed component onto an in-process store for a block.

Each registered Coordination, Cache, RateLimiterComponent, and CircuitBreakerComponent is replaced by one wired to a fresh MemoryProvider, under the same name. Everything is restored on exit. A test then runs the real code paths against real primitives, with no Redis and no Postgres:

async with micro:
    async with micro.fake():
        await checkout("cart-1")

Components with no backend to fake (Log, Trace, Metrics, HealthChecks) are left alone, and so is Outbox, which carries handlers and a relay that a swap would drop. Override those by hand with micro.override(...) when a test needs them.

RAISES	DESCRIPTION
`OutOfContextError`	If called outside an open `async with micro:` block, which is what `override` scopes to.

override `async`

override(*components: Component) -> AsyncIterator[None]

Swap component registrations for a block, restore them on exit.

Used in tests to substitute mock components:

async with micro:
    async with micro.override(Coordination(lock=MockLock())):
        await test_thing()

The override is scoped to the surrounding async with micro: block. The new components are entered when the override block opens and exited in reverse order when it closes.

Plain async context managers (registered via use(item) without a kind) cannot be overridden through this method. The caller already holds the reference and can substitute a mock at construction time.

PARAMETER	DESCRIPTION
`*components`	Components to install for the duration of the block. Each one shadows any component already registered under the same `(kind, name)` key. Original registrations are restored on exit, even if the block raises and even if one of these components fails to open. TYPE: `Component` DEFAULT: `()`

RAISES	DESCRIPTION
`OutOfContextError`	If called outside an open `async with micro:` block. The override needs an active app to scope to.

describe

describe(app: Any = None) -> AppReport

Return a structured report of what this app is wired with.

Answers what got wired, from what, reachable how far, and configured with what. Credential-like values are masked.

report = micro.describe()

assert report.ok
assert [c.kind for c in report.components] == ["cache", "coordination"]

Passing the app adds endpoints, one row per route saying what each registered component does to it, which answers "what happens to GET /products" without reading five registrations:

report = micro.describe(app)

for endpoint in report.endpoints:
    print(endpoint.method, endpoint.path, endpoint.applies)

python -m grelmicro check renders the same report and turns its checks into an exit code. Read more in the Wiring an App docs.

PARAMETER	DESCRIPTION
`app`	The Starlette, FastAPI, or FastStream application this app is installed on. Pass it to include the ambient binding check, which catches a forgotten `micro.install(app)`. Omit it for a service with no web framework. TYPE: `Any` DEFAULT: `None`

install

install(app: Any, *, ambient: bool = True) -> None

Wire the app lifecycle and ambient binding in one call.

Opens async with micro: alongside the framework's own lifecycle, so components are registered before any request or message is handled and closed on shutdown. A custom lifespan already passed to the framework keeps running, chained around this one.

When ambient is True (the default), each request handler or message subscriber runs with this app bound as Grelmicro.current(), so patterns that omit backend= resolve ambiently. It adds one middleware that sets a context variable and changes nothing a client can see.

Nothing else about how the framework answers a request changes. Register ErrorResponses() to have grelmicro's rejections rendered as RFC 9457 responses, IdempotentRequests() to have repeated requests replayed, and Trace() to have requests auto-instrumented. All three are wired here, and only because they were registered.

from fastapi import FastAPI

from grelmicro import Grelmicro

micro = Grelmicro(uses=[...])
app = FastAPI()
micro.install(app)

PARAMETER	DESCRIPTION
`app`	A Starlette, FastAPI, or FastStream application. The framework is detected from the object's shape, so the same call wires any of them. TYPE: `Any`
`ambient`	Wire per-handler ambient binding so `Lock(...)`, `@cached`, `RateLimiter...`, and the other patterns resolve through `Grelmicro.current()` inside request handlers and message subscribers. Default `True`. Pass `False` when handlers always pass an explicit `backend=` and the binding is not needed. TYPE: `bool` DEFAULT: `True`

RAISES	DESCRIPTION
`TypeError`	If `app` is not a recognized Starlette, FastAPI, or FastStream application.

check_ambient_binding

check_ambient_binding(app: Any) -> bool

Return whether ambient pattern resolution is wired for app.

Returns True when this app registers no ambient-resolving components (nothing needs binding) or when the binding middleware is installed on app by micro.install(app). Returns False when ambient components are registered but the middleware is missing, so a Lock(...), @cached, RateLimiter..., or CircuitBreaker(...) that omits backend= would raise OutOfContextError on every request.

Call it in a test to catch a forgotten micro.install(app) before it reaches production, where the failure only surfaces on the first affected request:

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

PARAMETER	DESCRIPTION
`app`	A Starlette, FastAPI, or FastStream application. TYPE: `Any`

RAISES	DESCRIPTION
`TypeError`	If `app` is not a recognized Starlette, FastAPI, or FastStream application and ambient components are registered.

GrelmicroMiddleware

GrelmicroMiddleware(app: ASGIApp, *, micro: Grelmicro)

Bind the active Grelmicro app for the duration of each request.

A request handler runs in its own task, outside the async with micro: block, so Grelmicro.current() and the ambient backend= resolution it powers do not see the app there. This middleware sets the active app for the request task, so Lock("cart"), RateLimiter.sliding_window(...), and @cached resolve ambiently inside the handler exactly as they do in a task.

from contextlib import asynccontextmanager

from fastapi import FastAPI

from grelmicro import Grelmicro, GrelmicroMiddleware

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

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with micro:
        yield

app = FastAPI(lifespan=lifespan)
app.add_middleware(GrelmicroMiddleware, micro=micro)

Open the app in the framework lifespan so its components are registered before any request arrives. The middleware is pure ASGI and works with any ASGI framework (Starlette, Litestar, ...). It binds on http and websocket scopes and passes the lifespan scope through untouched.

Initialize the middleware with the app to bind.

PARAMETER	DESCRIPTION
`app`	The next ASGI application in the middleware chain. TYPE: `ASGIApp`
`micro`	The `Grelmicro` app to bind for each request. Open it in the framework lifespan so its components are ready. TYPE: `Grelmicro`

app `instance-attribute`

app = app

micro `instance-attribute`

micro = micro

Component

Bases: AbstractAsyncContextManager['Component', bool | None], Protocol

A grelmicro component attached to a Grelmicro app.

Each grelmicro component wires one microservice pattern into the app (distributed lock, cache, rate limiter, circuit breaker, health check, ...). The user composes components into a Grelmicro application. The app opens every component in registration order and closes them in reverse order on exit.

ATTRIBUTE	DESCRIPTION
`kind`	Stable identifier for the component category (`"coordination"`, `"cache"`, `"ratelimiter"`, `"health"`, ...). The app exposes the component on `micro.<kind>` after registration. TYPE: `str`
`name`	Read-only registration name. Multiple components of the same `kind` may coexist under different names. The composite key for resolution is `(kind, name)`. TYPE: `str`
`singleton`	Optional class flag. When `True`, the app refuses to register a second component of the same `kind`. Set it on components that configure process-global state (the root logger, an OTel provider), and on any pair that cannot both be active: two components share a `kind` exactly when only one of them may answer. Absent means `False`. TYPE: `str`
`asgi_observes`	Optional class flag. `True` on a middleware that only watches a request, an access log above all. Those wrap outside the app's own middleware, so a request an outer layer refuses is still seen and still timed. Absent means `False`, which is where a middleware that may answer a request belongs: inside everything the app put in front of its handlers, so it can never be the reason a request skipped them. TYPE: `str`
`singleton_reason`	Optional class string saying why a second registration is refused, rendered into the error. Defaults to the process-global-state explanation, which is wrong for a pair excluded for any other reason. TYPE: `str`

Example

class Mailer:
    kind = "mailer"

    def __init__(self, *, name: str = "default") -> None:
        self._name = name

    @property
    def name(self) -> str:
        return self._name

    async def __aenter__(self) -> Self: ...
    async def __aexit__(self, exc_type, exc, tb) -> bool | None: ...

kind `class-attribute`

kind: str

name `property`

name: str

PARAMETER	DESCRIPTION
`uses`	Items registered at construction time, in the given order. Accepts `Component` instances (registered with `(kind, name)` lookup, exposed on `micro.<kind>`), bare adapter or component classes (instantiated for you), `Provider` instances (a lone Provider registers a default Component per kind it serves, `outbox` aside), and plain async context managers (lifecycled only, caller holds the reference). Two bare Providers with no Components raise `AmbiguousProviderError`. A `None` entry is skipped, so a component registered only for one backend stays a plain expression: `uses=[Log(), redis if backend == "redis" else None]`. Annotate a list you build beforehand with `Usable`. TYPE: `Iterable[Usable \| None] \| None` DEFAULT: `None`
`environment`	The deployment tier this app runs in: `"development"`, `"test"`, `"staging"` or `"production"`. A value outside those four raises `SettingsValidationError`. Falls back to `GREL_ENVIRONMENT`, which is read whatever `GREL_ENV_LOAD` says and which reports an unknown value instead of raising, and stays `None` when neither declares one. `"staging"` and `"production"` make an unmet backend scope a `BackendScopeError` at startup, `"development"` and `"test"` silence it, and an undeclared tier reports it once as a warning. The declared value also becomes the OpenTelemetry `deployment.environment.name` resource attribute. See the backend check. TYPE: `Environment \| None` DEFAULT: `None`
`strict`	Raise `LifecycleOrderError` instead of warning when a Component holds a Provider that is missing from `uses=` or listed after the dependent Component. Default `False` preserves the lenient warn-only behavior so existing apps keep starting. TYPE: `bool` DEFAULT: `False`
`allow_multiple`	Allow this app to run while another `Grelmicro` app is active in the same process. Off by default: components like `Log` and `Trace` own process-global state that two overlapping app lifecycles would restore out of order. Setting `True` opts out of the guard when you are sure no two active apps configure the same global state. TYPE: `bool` DEFAULT: `False`

App

grelmicro

Usable

Grelmicro

opened property

environment property

components property

providers property

coordination property

cache property

log property

trace property

metrics property

health property

outbox property

ratelimiter property

circuitbreaker property