Fastvia

Production-ready FastAPI backend toolkit with auth, DB, migrations, Redis, ARQ jobs, middleware, logging, pagination, rate limiting, and security helpers.

Fastvia does not take over your application.
You still create your own FastAPI app, routers, lifespan, database models, and business logic. Fastvia gives you clean reusable building blocks for the parts most production APIs need.

The PyPI distribution name is fastvia-kit, while the Python import package is fastvia.

Installation

pip install fastvia-kit

If you also want Uvicorn for running a FastAPI app:

pip install "fastvia-kit[server]"

For development:

make install-dev

What Fastvia Provides

• Request context middleware with request ID and process time
• Security headers middleware
• Logging context with text and JSON formatters
• API error classes and exception handlers
• Pagination helpers
• Redis-backed rate limiting with SlowAPI
• Async SQLAlchemy database helpers
• Redis client helpers
• ARQ job helpers
• JWT/FastAPI Users auth helpers
• Alembic migration helpers
• Bootstrap helper for existing FastAPI apps

Quick Start

You create your FastAPI app normally:

from fastapi import FastAPI

from fastvia import bootstrap_fastvia

app = FastAPI(
    title="My API",
    version="1.0.0",
)

bootstrap_fastvia(
    app,
    logger_name="myapi",
    log_format="json",
    cors_origins=["http://localhost:3000"],
    rate_limit_storage_uri="redis://localhost:6379/0",
)

This can configure:

• logging
• request context middleware
• security headers
• error handlers
• CORS
• GZip
• rate limiting

Example

A small runnable FastAPI example is available in:

examples/basic_app.py

Install Fastvia with server support first:

pip install "fastvia-kit[server]"

Run the example app:

uvicorn examples.basic_app:app --reload

Then open:

http://127.0.0.1:8000/docs

The example shows:

• Fastvia bootstrap setup
• request context headers
• security headers
• pagination
• API errors
• rate limiting

Examples

Fastvia includes small runnable examples:

examples/basic_app.py
examples/middleware_app.py
examples/cors_app.py
examples/error_handling_app.py
examples/db_pagination_app.py
examples/rate_limit_user_or_ip_app.py
examples/redis_app.py
examples/arq_jobs_app.py
examples/arq_worker.py
examples/auth_jwt_app.py
examples/logging_app.py
examples/alembic_migration_script.py

Run any FastAPI example with:

uvicorn examples.basic_app:app --reload

Some examples need extra services:

• redis_app.py needs Redis running.
• arq_jobs_app.py needs Redis and an ARQ worker.
• db_pagination_app.py uses SQLite with async SQLAlchemy.
• alembic_migration_script.py expects an existing Alembic setup.

Request Context Middleware

from fastapi import FastAPI

from fastvia.middleware import RequestContextMiddleware

app = FastAPI()
app.add_middleware(RequestContextMiddleware)

Each response includes:

x-request-id
x-process-time

The request context is also connected to Fastvia logging.

Security Headers Middleware

from fastvia.middleware import SecurityHeadersMiddleware

app.add_middleware(
    SecurityHeadersMiddleware,
    enable_hsts=True,
)

By default, Fastvia adds common security headers such as:

• X-Content-Type-Options
• X-Frame-Options
• Referrer-Policy
• Permissions-Policy
• Cache-Control
• Content-Security-Policy

You can customize or disable values when needed.

Logging

from fastvia.logging import setup_logging, log_extra

logger = setup_logging(
    level="INFO",
    log_format="json",
    logger_name="myapi",
)

logger.info(
    "Event processed",
    extra=log_extra(
        event_id="evt_123",
        event_type="demo.event.success",
    ),
)

Fastvia supports request-scoped log context:

request_id
method
path
event_id
event_type

API Errors

from fastapi import FastAPI

from fastvia.errors import NotFoundError, register_exception_handlers
from fastvia.middleware import RequestContextMiddleware

app = FastAPI()
app.add_middleware(RequestContextMiddleware)
register_exception_handlers(app)


@app.get("/items/{item_id}")
async def get_item(item_id: int):
    raise NotFoundError("Item not found.")

Example response:

{
  "error": "not_found",
  "message": "Item not found.",
  "details": {},
  "request_id": "..."
}

Pagination

from fastapi import Depends
from pydantic import BaseModel

from fastvia.pagination import (
    PageMeta,
    PaginatedResponse,
    PaginationParams,
    pagination_params,
)


class ItemOut(BaseModel):
    id: int
    name: str


@app.get("/items", response_model=PaginatedResponse[ItemOut])
async def list_items(
    pagination: PaginationParams = Depends(pagination_params),
):
    rows = [
        ItemOut(id=1, name="First"),
        ItemOut(id=2, name="Second"),
    ]
    total = 2

    meta = PageMeta.from_pagination(
        total=total,
        pagination=pagination,
    )

    return PaginatedResponse(
        items=rows,
        meta=meta,
    )

Response shape:

{
  "items": [],
  "meta": {
    "total": 42,
    "page": 2,
    "per_page": 10,
    "pages": 5,
    "has_next": true,
    "has_prev": true
  }
}

Rate Limiting

from fastapi import Request

from fastvia.rate_limit import register_rate_limiting

limiter = register_rate_limiting(
    app,
    storage_uri="redis://localhost:6379/0",
    default_limits=[],
)


@app.get("/ping")
@limiter.limit("10/minute")
async def ping(request: Request):
    return {"message": "pong"}

For authenticated user-based keys:

from fastvia.auth import get_user_id_from_bearer_token
from fastvia.rate_limit import user_or_ip_key

key_func = user_or_ip_key(
    user_id_resolver=lambda request: get_user_id_from_bearer_token(
        request,
        secret="secret",
        algorithm="HS256",
    )
)

limiter = register_rate_limiting(
    app,
    storage_uri="redis://localhost:6379/0",
    key_func=key_func,
)

Database Helpers

from fastvia.db import (
    create_async_db_engine,
    create_session_dependency,
    create_session_factory,
)

engine = create_async_db_engine(
    "postgresql+asyncpg://user:password@localhost/db",
)

SessionLocal = create_session_factory(engine)

get_db = create_session_dependency(SessionLocal)

Then use it in routes:

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession


@app.get("/items")
async def list_items(db: AsyncSession = Depends(get_db)):
    ...

Redis Helper

from fastvia.redis import close_all_redis, get_redis

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

await redis.set("key", "value")

Close clients during shutdown:

await close_all_redis()

ARQ Helpers

from fastvia.jobs import create_arq_pool

arq_pool = await create_arq_pool("redis://localhost:6379/0")

Fastvia also provides JSON job serializer/deserializer helpers.

Auth Helpers

Fastvia provides helpers for FastAPI Users JWT auth setup:

from fastvia.auth import create_jwt_auth_backend

auth_backend = create_jwt_auth_backend(
    secret="secret",
    lifetime_seconds=3600,
    algorithm="HS256",
    token_url="/auth/login",
)

It also provides helpers for wiring FastAPI Users dependencies while keeping your app-specific UserManager, email verification, reset password, and user model inside your own project.

Alembic Migration Helpers

from fastvia.migrations import create_alembic_config, upgrade_to_head

config = create_alembic_config(
    config_path="alembic.ini",
    database_url="postgresql+asyncpg://user:password@localhost/db",
)

upgrade_to_head(config)

Fastvia does not run migrations automatically.
Migrations should be explicit, especially in production.

Development

Install development dependencies:

make install-dev

Run tests:

make test

Run tests with coverage:

make test-cov

Format code:

make format

Lint code:

make lint

Clean build and cache files:

make clean

Build package:

make build

Check package:

make check

License

MIT