grelmicro 0.26.0

grelmicro is a lightweight toolkit for building Python applications that need to coordinate work across processes

Async-first toolkit. Microservice patterns inside.

A Python toolkit for distributed systems: microservices, modular monoliths, and self-contained systems.

Project status: Active development. grelmicro is pre-1.0. The public API is not yet stable. Breaking changes are allowed on MINOR bumps (0.14.0 → 0.15.0) and never on PATCH. Pin the minor: grelmicro>=0.14.0,<0.15.0. After 1.0.0, standard semver applies. See the versioning policy.

---

Documentation: https://grelinfo.github.io/grelmicro/

Source Code: https://github.com/grelinfo/grelmicro

---

Why grelmicro

grelmicro ships microservice patterns as small, composable modules with pluggable backends: locks, rate limits, circuit breakers, cache, logging, health checks, and task scheduling. Async-first, type-safe, Pydantic-validated. A local pre-commit hook gates commits on 100% pytest coverage (coverage --fail-under=100).

It is built for any Python application that coordinates work across processes, workers, or replicas. The same primitives serve every distributed system, whether you call it microservices, a modular monolith, or a self-contained system. A distributed lock is a distributed lock whether your system is one process or fifty. It fits naturally into cloud-native applications, containerized apps, and Kubernetes deployments.

• Micro: one focused primitive per module. Each covers a microservice pattern (distributed lock, leader election, rate limiter, circuit breaker, health check API, externalised configuration).
• Fast: small footprint by design. We keep the layers thin so your code stays quick.
• Async-first: every I/O call is async / await. Drops into FastAPI, FastStream, and any asyncio-based stack.
• Backend-agnostic: each primitive is a protocol. Swap Redis for PostgreSQL or SQLite without touching application code.
• Railguarded: 100% pytest coverage, ty-checked, ruff-linted, Pydantic-validated. Pre-1.0 API may shift on minor bumps. 1.x commits to standard semver.

Already using aiocache, slowapi, pybreaker, tenacity, or aioredlock? See the comparison page for a per-domain breakdown.

Modules

Module	Summary
Cache	@cached decorator with local, distributed, and early (XFetch) stampede protection. In-memory TTLCache or RedisCacheAdapter.
Synchronization	Distributed Lock, TaskLock, LeaderElection. Redis, PostgreSQL, SQLite, Kubernetes, in-memory.
Task Scheduler	Periodic task execution with optional distributed locking. Lightweight, not a Celery replacement.
Resilience	Circuit Breaker and Rate Limiter with pluggable algorithms (TokenBucketConfig, SlidingWindowConfig).
Logging	12-factor logging with JSON, LOGFMT, TEXT, or PRETTY output, structured error rendering, and OpenTelemetry trace context.
Tracing	Unified instrumentation. @instrument creates OpenTelemetry spans and enriches log records with structured context.
Health	Health check registry with concurrent runners and FastAPI liveness / readiness integration.
JSON	Fast JSON via orjson when available, with automatic fallback to stdlib json.

Installation

pip install grelmicro

See the Installation guide for uv and poetry commands, plus optional extras for Redis, PostgreSQL, SQLite, Kubernetes, OpenTelemetry, and structlog.

Example

Run the demo

Want to see every Pattern running against real Redis and Postgres? The FastAPI demo starts in three commands:

cd examples/fastapi-demo
docker compose up --wait
open http://localhost:8000/docs

It wires a cached endpoint, a rate-limited endpoint, a circuit-breaker-protected endpoint, a distributed lock, a leader-gated task, and /healthz / /readyz probes. Read app.py to see each one.

One route, one primitive

The smallest grelmicro program: a FastAPI route protected by a process-local rate limiter. No Grelmicro(...), no Redis, no lifespan.

from fastapi import FastAPI

from grelmicro.resilience import RateLimitExceededError, RateLimiter

app = FastAPI()
api_limiter = RateLimiter.sliding_window("api", limit=100, window=60)


@app.get("/ping")
async def ping() -> dict[str, str]:
    try:
        await api_limiter.acquire_or_raise(key="global")
    except RateLimitExceededError:
        return {"status": "throttled"}
    return {"status": "ok"}

That is the whole thing. Pick a primitive, name it, call it. Swap to a fleet-wide backend later by composing it inside Grelmicro(uses=[redis, RateLimiters(redis)]) as shown below.

Lifespan with one provider and one component

To make the rate limiter fleet-wide, wrap it in a Grelmicro container with one provider and one component, then bind it to FastAPI's lifespan.

from contextlib import asynccontextmanager

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import (
    RateLimitExceededError,
    RateLimiter,
    RateLimiters,
)

redis = RedisProvider("redis://localhost:6379/0")
micro = Grelmicro(uses=[redis, RateLimiters(redis)])

api_limiter = RateLimiter.sliding_window("api", limit=100, window=60)


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


app = FastAPI(lifespan=lifespan)


@app.get("/ping")
async def ping() -> dict[str, str]:
    try:
        await api_limiter.acquire_or_raise(key="global")
    except RateLimitExceededError:
        return {"status": "throttled"}
    return {"status": "ok"}

Adding more primitives is the same shape: one extra entry in uses=[...]. The full demo below shows what that looks like.

FastAPI integration

Create a file main.py with:

import logging
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException, Request

from grelmicro import Grelmicro
from grelmicro.cache import Cache, JsonSerializer, TTLCache, cached
from grelmicro.health import HealthChecks
from grelmicro.log import configure as configure_logging
from grelmicro.providers.redis import RedisProvider
from grelmicro.resilience import (
    CircuitBreaker,
    CircuitBreakers,
    RateLimitExceededError,
    RateLimiter,
    RateLimiters,
)
from grelmicro.resilience.circuitbreaker.memory import MemoryCircuitBreakerAdapter
from grelmicro.sync import LeaderElection, Lock, Sync
from grelmicro.task import Tasks

logger = logging.getLogger(__name__)

# === grelmicro app: one container, one lifespan ===
tasks = Tasks()
health = HealthChecks()
leader = LeaderElection("leader-election")
tasks.add_task(leader)

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

# Patterns used inside FastAPI request handlers take an explicit backend:
# handlers run in their own task, outside the app's ambient scope. Background
# tasks run inside that scope, so they resolve their backend ambiently.
sync_backend = redis.sync()
cache_backend = redis.cache()
ratelimiter_backend = redis.ratelimiter()
breaker_backend = MemoryCircuitBreakerAdapter()

micro = Grelmicro(uses=[
    redis,
    Sync(sync_backend),
    Cache(cache_backend),
    RateLimiters(ratelimiter_backend),
    CircuitBreakers(breaker_backend),
    tasks,
    health,
])

# === Patterns declared once at module load ===
ttl_cache = TTLCache(ttl=300, serializer=JsonSerializer(), backend=cache_backend)
lock = Lock("shared-resource", backend=sync_backend)
cb = CircuitBreaker("my-service", backend=breaker_backend)
api_limiter = RateLimiter.sliding_window(
    "api", limit=100, window=60, backend=ratelimiter_backend
)


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


app = FastAPI(lifespan=lifespan)


# --- Cache: avoid redundant database queries ---
@cached(ttl_cache)
async def get_user(user_id: int) -> dict:
    return {"id": user_id, "name": "Alice"}


@app.get("/users/{user_id}")
async def read_user(user_id: int):
    return await get_user(user_id)


# --- Circuit Breaker: protect calls to an unreliable service ---
@app.get("/")
async def read_root():
    async with cb:
        return {"Hello": "World"}


# --- Rate Limiter: protect endpoints from overload ---
@app.get("/api")
async def api_endpoint(request: Request):
    try:
        await api_limiter.acquire_or_raise(key=request.client.host)
    except RateLimitExceededError as exc:
        raise HTTPException(
            status_code=429,
            detail="Too many requests",
            headers={"Retry-After": str(int(exc.retry_after))},
        )
    return {"status": "ok"}


# --- Distributed Lock: synchronize access to a shared resource ---
@app.get("/protected")
async def protected():
    async with lock:
        return {"status": "ok"}


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


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


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

The key shape:

• One container, one lifespan. Grelmicro(uses=[...]) lists every Provider, Component, and active manager. async with micro: opens them all in order, closes in reverse.
• One Provider, many Components. Sync(redis), Cache(redis), RateLimiters(redis) all share the same RedisProvider pool. The Provider holds the connection, the Components attach to it.
• Patterns are declared at module load. Lock("cart"), TTLCache(ttl=60), CircuitBreaker("svc") carry no backend reference. They resolve through the active app inside async with. The same Lock works in production with Redis and in tests with MemorySyncAdapter, no rewiring.
• Pay only for what you import. import grelmicro does not pull in redis, psycopg, or any other vendor SDK. First-party Providers live under grelmicro.providers.{vendor} and load only when you import them.

For multiple Redis instances, separate names, or test overrides, see the docs.

License

This project is licensed under the terms of the MIT license.