Beyond — FastAPI Runtime Composition Library

One line per feature. Zero magic. A normal FastAPI instance at the end.

Beyond wraps FastAPI with a fluent, opt-in composition API for configuration loading, infrastructure adapters, and authentication bootstrapping. Every integration registers through standard FastAPI mechanisms — lifespan, dependency injection, included routers — and exposes the underlying raw client objects. Nothing is hidden.

📖 Full documentation →

from beyond_fastapi import Beyond, load_settings, require_user, DbSession, RedisClient

settings = load_settings(AppSettings, yaml="settings.yaml")

app = (
    Beyond(settings)
    .sqlalchemy(config=settings.postgres)
    .redis(config=settings.redis)
    .auth.jwt_header(jwt_config=settings.jwt)
    .build()
)

@app.get("/me")
async def me(payload: dict = require_user, db: DbSession = None, cache: RedisClient = None):
    return {"sub": payload.get("sub")}

---

Installation

pip install beyond-fastapi        # core (fastapi, pydantic-settings, pyyaml, python-jose)

# Optional adapters — install only what you need
pip install beyond-fastapi[sqlalchemy]   # SQLAlchemy + asyncpg
pip install beyond-fastapi[redis]        # Redis
pip install beyond-fastapi[rabbitmq]     # RabbitMQ (aio-pika)
pip install beyond-fastapi[beanie]       # Beanie + Motor (MongoDB)

Python ≥ 3.11 required.

---

Configuration Loading

Define your settings schema with Pydantic. Drop in Beyond's config fragments or roll your own. Environment variables always win — YAML is the base.

from pydantic_settings import BaseSettings
from beyond_fastapi import load_settings, PostgresConfig, RedisConfig, JwtConfig

class AppSettings(BaseSettings):
    debug: bool = False
    postgres: PostgresConfig
    redis: RedisConfig
    jwt: JwtConfig

settings = load_settings(AppSettings, yaml="settings.yaml")

settings.yaml

debug: true
postgres:
  dsn: postgresql+asyncpg://user:pass@localhost:5432/mydb
  pool_size: 20
  echo: false
redis:
  url: redis://localhost:6379
jwt:
  secret: super-secret-key
  algorithm: HS256
  access_token_expire_seconds: 7200

Priority (highest to lowest):

1. Environment variables
2. .env file
3. YAML file
4. Field defaults

export POSTGRES__DSN=postgresql+asyncpg://prod:pass@host:5432/db  # wins over YAML

Built-in config fragments

Fragment	Key fields
PostgresConfig	dsn, pool_size, max_overflow, echo
RedisConfig	url, max_connections
RabbitmqConfig	url
BeanieConfig	uri, database
JwtConfig	secret, algorithm, access_token_expire_seconds

All fragments are plain pydantic.BaseModel — embed them in your own BaseSettings subclass.

---

Adapters

Every adapter installs its resources during startup, cleans up on shutdown, and exposes a FastAPI dependency for per-request usage. Raw clients are always accessible on app.state.beyond.

SQLAlchemy

from beyond_fastapi import Beyond, DbSession
from beyond_fastapi.config.loader import PostgresConfig

kit = Beyond(settings)
kit.sqlalchemy(config=settings.postgres)  # or: kit.sqlalchemy(dsn="postgresql+asyncpg://...")

app = kit.build()

@app.get("/users")
async def list_users(session: DbSession):
    result = await session.execute(text("SELECT id, name FROM users"))
    return [{"id": r.id, "name": r.name} for r in result]

Raw access:

engine = app.state.beyond.sqlalchemy_engine          # AsyncEngine
factory = app.state.beyond.sqlalchemy_session_factory # async_sessionmaker

Redis

from beyond_fastapi import Beyond, RedisClient

kit = Beyond(settings)
kit.redis(config=settings.redis)  # or: kit.redis(url="redis://localhost")

app = kit.build()

@app.get("/cache/{key}")
async def get_cache(key: str, cache: RedisClient):
    value = await cache.get(key)
    return {"key": key, "value": value}

Raw access:

redis = app.state.beyond.redis_client  # redis.asyncio.Redis
await redis.ping()

RabbitMQ

from beyond_fastapi import Beyond
from beyond_fastapi.deps.providers import get_rabbitmq_channel

kit = Beyond(settings)
kit.rabbitmq(config=settings.rabbitmq)  # or: kit.rabbitmq(url="amqp://guest:guest@localhost/")

app = kit.build()

@app.get("/publish")
async def publish(channel = Depends(get_rabbitmq_channel)):
    await channel.default_exchange.publish(
        aio_pika.Message(body=b"Hello!"),
        routing_key="my_queue",
    )
    return {"status": "published"}

Raw access:

conn = app.state.beyond.rabbitmq_connection  # aio_pika AbstractRobustConnection

Beanie / MongoDB

from beanie import Document
from beyond_fastapi import Beyond

class UserDoc(Document):
    name: str
    email: str
    class Settings:
        name = "users"

kit = Beyond(settings)
kit.beanie(config=settings.beanie, document_models=[UserDoc])

app = kit.build()

Raw access:

client = app.state.beyond.beanie_client  # motor.motor_asyncio.AsyncIOMotorClient

---

Authentication

Beyond implements the transport + strategy pattern. A backend = how the token travels (transport) + how it's generated/validated (strategy). Multiple backends can coexist — they're tried in registration order.

JWT Bearer Header

kit = Beyond(settings)
kit.auth.jwt_header(
    secret="my-secret",
    algorithm="HS256",
    expire_seconds=3600,
)

# or from config:
kit.auth.jwt_header(jwt_config=settings.jwt)

JWT Cookie

kit.auth.jwt_cookie(
    secret="my-secret",
    cookie_name="session",
    cookie_max_age=3600,
    cookie_secure=True,
)

Multiple backends (tried in order)

kit.auth.jwt_header(name="api", secret="...")
kit.auth.jwt_cookie(name="web", secret="...")

Protecting routes

from beyond_fastapi import require_user, current_user

@app.get("/me")
async def me(payload: dict = require_user):
    # raises 401 if no valid token
    return {"sub": payload.get("sub")}

@app.get("/optional")
async def optional(payload: dict = current_user):
    # payload is None if not authenticated
    return {"authenticated": payload is not None}

Custom entity names

from beyond_fastapi import make_required_auth_dependency

require_admin = make_required_auth_dependency("admin")

@app.get("/admin")
async def admin_dashboard(payload: dict = require_admin):
    ...

Login / logout router

Beyond provides an optional, minimal login/logout router:

from beyond_fastapi.auth.router import make_auth_router
from fastapi import Depends, HTTPException

async def authenticate_user(username: str, password: str) -> dict:
    """Your custom credential check. Return a dict for the JWT or raise 401."""
    if username == "admin" and password == "secret":
        return {"sub": username, "role": "admin"}
    raise HTTPException(401, "Invalid credentials")

router = make_auth_router(
    backend_name="bearer",
    kit=kit,
    get_user_data=Depends(authenticate_user),
    prefix="/auth",
)
app.include_router(router)

---

Dependency Cheat Sheet

Import from one place:

from beyond_fastapi.deps.providers import (
    DbSession,              # Annotated[AsyncSession, Depends(get_db_session)]
    RedisClient,            # Annotated[Redis, Depends(get_redis)]
    get_rabbitmq_channel,   # async generator — yields per-request channel
    get_rabbitmq_connection,# returns the persistent connection
    get_motor_client,       # returns AsyncIOMotorClient
    current_user,           # optional auth — dict | None
    require_user,           # required auth — dict or 401
)

Or import directly from beyond_fastapi:

from beyond_fastapi import DbSession, RedisClient, require_user

---

Full Application Example

# ── settings.py ──────────────────────────────────────────────────────────
from pydantic_settings import BaseSettings
from beyond_fastapi import PostgresConfig, RedisConfig, JwtConfig

class AppSettings(BaseSettings):
    debug: bool = False
    postgres: PostgresConfig
    redis: RedisConfig
    jwt: JwtConfig

# ── main.py ──────────────────────────────────────────────────────────────
from fastapi import FastAPI, Depends, HTTPException
from beyond_fastapi import Beyond, load_settings, require_user, DbSession, RedisClient

settings = load_settings(AppSettings, yaml="settings.yaml")

kit = Beyond(settings)
kit.sqlalchemy(config=settings.postgres)
kit.redis(config=settings.redis)
kit.auth.jwt_header(jwt_config=settings.jwt)
kit.auth.jwt_cookie(jwt_config=settings.jwt, cookie_name="session")

app: FastAPI = kit.fastapi(title="My API", docs_url="/api/docs")

@app.get("/me")
async def me(
    payload: dict = require_user,
    session: DbSession = None,
    cache: RedisClient = None,
):
    return {"sub": payload.get("sub")}

@app.get("/health")
async def health(cache: RedisClient = None):
    await cache.ping()
    return {"status": "ok"}

@app.get("/debug/state")
async def debug_state(request):
    state = request.app.state.beyond
    return {
        "has_db": state.sqlalchemy_engine is not None,
        "has_redis": state.redis_client is not None,
    }

---

Documentation

Comprehensive documentation is available in the docs/ directory:

Section	Description
Getting Started	Installation, core concepts, first application
Configuration	load_settings, config fragments, YAML + env priority
Adapters	SQLAlchemy, Redis, RabbitMQ, Beanie — detailed usage
Authentication	JWT header/cookie, multi-backend, route protection, login/logout
Advanced Usage	Custom adapters, app.state internals, raw access, error handling
API Reference	Every public class, method, and function

---

Design Principles

• Produces a normal FastAPI instance. build() returns FastAPI, not a subclass.
• No globals. Each Beyond instance is self-contained.
• No magic. All registrations go through standard FastAPI mechanisms (lifespan, Depends, APIRouter).
• Escape hatches everywhere. Every adapter exposes its raw client on app.state.beyond.
• No middleware by default. No CORS, no logging, no health checks. You add what you need.
• No auto-detection. Your settings class is always explicit. Env vars win — that's it.

---

License

MIT