ForgeAPI — Documentation

Table of Contents

1. Quick start
2. Project structure
3. Core
4. Exceptions
5. Auth
   • How it works — Guards
   • CurrentUser and OptionalUser
   • Auth facade — issuing tokens
   • Multiple guards
   • JWT strategy
   • Cookie strategy
   • Telegram strategy
6. Pagination
   • QuerySet .paginate()
   • Offset pagination
   • Cursor pagination
7. Broadcasting
   • BroadcastManager
   • Pub/Sub mode
   • Stream mode
   • Cross-project example
   • Registering handlers
   • Lifecycle
8. Controllers
   • Base pattern
   • schema class var
   • Route decorator
   • Auto-prefix and namespace
9. Schemas
   • Base classes
   • Schema directories
   • generate:schema
10. Policies
11. ModelMixin
12. Query Scopes
13. Model Observers
14. Permissions
    • Setup
    • PermissionsMixin
    • Dependencies
    • Role and Permission models
15. Cache
    • Basic operations
    • Common patterns
    • Drivers
    • Configuration
16. Storage
    • Configuration
    • Basic operations
    • Multiple disks
    • S3 / S3-compatible
    • ImageProcessor
17. Support
    • Number
    • Str
    • Time
18. Logger
19. Middleware
    • CORS
    • Rate limiting
    • Request ID
    • Access logging
20. Settings
21. Seeders
22. Scheduler
23. CLI reference
24. Configuration reference (config/)
25. Telescope
    • What Telescope captures
    • WebSocket live stream
    • Sensitive data masking
    • Recording jobs
26. MCP Server
    • Install
    • Global setup for Claude Code
    • Per-project setup
    • Available tools

---

1. Quick start

pip install forge-kits

Optional dependencies

Extra	Installs	Use when
auth	pyjwt	JWT or Cookie auth strategy
asyncpg	tortoise-orm, asyncpg	PostgreSQL
aiosqlite	tortoise-orm, aiosqlite	SQLite
aiomysql	tortoise-orm, aiomysql	MySQL / MariaDB
db	tortoise-orm	ORM only, bring your own driver
redis	redis	Cache Redis driver / Redis events
full-asyncpg	auth + asyncpg	PostgreSQL + JWT
full-aiosqlite	auth + aiosqlite	SQLite + JWT
full-aiomysql	auth + aiomysql	MySQL + JWT
full	auth + all three drivers	everything
s3	boto3	S3 / MinIO / Cloudflare R2 storage driver
images	Pillow	ImageProcessor helper
mcp	mcp	forge-kits MCP server for AI-assisted development

pip install forge-kits[full-asyncpg]   # PostgreSQL + JWT
pip install forge-kits[mcp]            # MCP server
pip install forge-kits[redis]          # Redis cache / events
pip install forge-kits[s3,images]      # S3 storage + image processing

forgeapi init my-project
cd my-project

forgeapi db:init && forgeapi db:makemigrations && forgeapi db:migrate
forgeapi runserver --reload

forgeapi init asks for auth strategy (jwt / cookie / telegram), DB driver (asyncpg / aiosqlite / aiomysql), and whether to generate the welcome boilerplate (User + Post + events).

---

2. Project structure

After forgeapi init my-project:

my-project/
  main.py                    # entry point — FastAPI app + Core(app)
  .env                       # secrets (JWT_SECRET, DB_* etc.)
  config/                    # Laravel-style config directory (see §24)
    project.py               # name, debug, extra providers
    structure.py             # directory layout + base_prefix
    http.py                  # cors, rate_limit, request_id, access_log, middleware
    auth.py                  # named guards (jwt / cookie / telegram)
    pagination.py            # default_limit, max_limit
    database.py              # TORTOISE_ORM dict lives here
    storage.py               # driver, root, base_url (or S3 config)
  app/
    controllers/             # *_controller.py files, auto-loaded by Core
    schemas/                 # Pydantic schemas
    events/                  # Event subclasses
    listeners/               # @broadcast.on(...) handlers
    policies/                # Policy classes, auto-discovered
  database/
    models/                  # Tortoise models
    migrations/              # migration files (tortoise CLI)
    seeds/                   # Seeder classes

main.py:

from fastapi import FastAPI
from forgeapi import Core
from tortoise.contrib.fastapi import register_tortoise
from config.database import TORTOISE_ORM

app = FastAPI()

core = Core(app)   # everything is wired from config/

register_tortoise(app, config=TORTOISE_ORM, generate_schemas=False, add_exception_handlers=True)

---

3. Core

Core(app) takes only the app. Everything else is config-driven — convention over configuration. Core is a thin orchestrator: each module ships a Provider with a register() phase (module wiring) and a boot() phase (user-code discovery); Core collects them, runs all register()s, then all boot()s.

from forgeapi import Core

core = Core(app)

What boots when

Module	Activated by
Middleware stack	config/http.py (cors, rate_limit, request_id, access_log, middleware)
Auth guards	config/auth.py exists
Storage	config/storage.py exists — configures the Storage facade
Controllers	controllers_dir exists — all *_controller.py imported recursively
Event listeners	listeners_dir exists — all files imported
Policies	policies_dir exists — all *_policy.py imported
Permissions	a model in models_dir inherits PermissionsMixin (no config needed)
Telescope	"debug": True in config/project.py — never in production
Pagination, Cache	always configured (from their sections or defaults)
Custom providers	"providers" list in config/project.py

Core(app) with no config files never raises — zero-config is a supported starting point.

Custom providers

# config/project.py
config = {"providers": ["app.providers.MetricsProvider"]}

from forgeapi import Provider

class MetricsProvider(Provider):
    def register(self) -> None: ...   # module wiring — must not import user code
    def boot(self) -> None: ...       # runs after all register()s

Accessing after setup

core.auth       # → Auth facade | None
core.config     # → KitConfig (validated config/ sections)
core.providers  # → list of active Provider instances

Including routers manually

core.include_router(admin_router)                    # prefix: /api/v1
core.include_router(admin_router, prefix="/admin")   # prefix: /api/v1/admin

Debug mode

# config/project.py
from forgeapi import env
config = {"debug": env("APP_DEBUG", False)}

Debug mode enables Telescope; all debug activity is logged as WARNING. Never enable in production.

---

4. Exceptions

from forgeapi import ForgeAPIError, ForgeAPIConfigError, ForgeAPIImportError

Class	When raised
ForgeAPIError	Base — all ForgeAPI errors
ForgeAPIConfigError	Misconfiguration: missing secret, unknown strategy, model not found
ForgeAPIImportError	Optional dependency not installed

Every exception includes a hint field with a fix suggestion.

try:
    core = Core(app)
except ForgeAPIConfigError as e:
    print(e.hint)

Auth has its own domain-exception hierarchy (ForgeAPIAuthError subclasses: TokenExpiredError, TokenInvalidError, SessionExpiredError, SessionInvalidError, UserNotFoundError). Strategies raise only these — Guard.authenticate is the single point translating them to HTTP 401 (see §5).

---

5. Auth

Auth is based on Guards. A Guard combines a strategy (how to verify credentials) with an optional user model (which Tortoise model to load from the DB).

Auth (facade)   — guard registry + strategy factories, pure delegation
 └─ Guard       — the only layer that speaks HTTP (401) and touches the DB
     └─ AuthStrategy — pure domain: extract/verify credentials, issue tokens

Step 1 — define guards in config/auth.py

Auth boots automatically when config/auth.py exists — nothing to pass to Core.

# config/auth.py
from forgeapi import env

config = {
    "default": "api",
    "guards": {
        "api": {
            "strategy": "jwt",              # jwt | cookie | telegram | custom
            "secret": env("JWT_SECRET"),
            "access_ttl": 30,               # minutes
            "refresh_ttl": 7,               # days
            "model": "database.models.user.User",   # optional — load user from DB
        },
    },
}

strategy picks the credential mechanism; the remaining keys are passed to the strategy's from_config() (see per-strategy tables below). model is a dotted path to a Tortoise model — when set, CurrentUser resolves to a real DB instance (and a valid token whose user is gone from the DB is a 401).

Step 2 — protect routes

from forgeapi.auth import CurrentUser, OptionalUser

@route.get("/me")
async def me(self, user: CurrentUser):
    return {"id": user.id}

@route.get("/feed")
async def feed(self, user: OptionalUser):
    if user:
        return await personalised_feed(user.id)
    return await public_feed()

Error semantics (uniform across strategies)

• credentials absent → 401 for CurrentUser, None for OptionalUser
• credentials present but invalid (expired / bad signature / user gone from the DB) → 401 always, even for OptionalUser
• every 401 carries WWW-Authenticate: Bearer error="<code>" with a machine-readable code: token_expired, token_invalid, session_expired, session_invalid, user_not_found, missing_credentials

What the user object contains

With "model" configured, the dependency returns your Tortoise model instance loaded from the DB — all its fields and methods are available.

Without "model", it returns a lightweight AuthUser built from the token payload:

Field	Type	Description
user.id	str | int	JWT sub claim — cast with int(user.id) for DB queries
user.username	str | None	From token payload
user.auth_method	str	"jwt" / "cookie" / "telegram"
user.extra	dict	Any extra claims (role, email, etc.)

Token claims — auth_claims()

Define auth_claims() on the user model to control what goes into tokens (sub is auto-filled from user.id):

class User(ModelMixin, Model):
    def auth_claims(self) -> dict:
        return {"username": self.username, "role": self.role}

Auth facade — issuing tokens

from forgeapi.auth import auth

access  = auth.token(user)           # access token — takes DB model instance
refresh = auth.refresh_token(user)   # refresh token — RefreshCapable strategies (jwt)

# Decode
payload = auth.decode(token, expected_type="access")  # raises TokenExpiredError | TokenInvalidError

# SessionIssuer strategies (cookie) only
auth.set_cookie(response, {"sub": str(user.id), "username": user.username})
auth.delete_cookie(response)

auth.token(user) accepts any model instance — sub is auto-filled from user.id, the rest comes from the model's auth_claims() hook (see above).

Capabilities are protocols (forgeapi.auth.contracts): TokenIssuer, RefreshCapable, SessionIssuer. The facade dispatches on isinstance(strategy, Protocol) — a custom strategy that implements a protocol gets token() / decode() / cookie helpers for free.

Login endpoint pattern

@route.post("/login")
async def login(self, payload: LoginPayload) -> dict:
    user = await User.get_or_none(email=payload.email)
    if not user or not user.verify_password(payload.password):
        raise HTTPException(401, "Invalid credentials")
    return {
        "access_token":  auth.token(user),
        "refresh_token": auth.refresh_token(user),
        "token_type":    "bearer",
    }

@route.post("/refresh")
async def refresh(self, payload: RefreshPayload) -> dict:
    try:
        data = auth.decode(payload.refresh_token, expected_type="refresh")
    except (TokenExpiredError, TokenInvalidError) as e:
        raise HTTPException(401, str(e))
    user = await User.find_or_fail(int(data["sub"]))
    return {"access_token": auth.token(user), "token_type": "bearer"}

Multiple guards

Define several named guards in config/auth.py; each gets its own strategy, secret, and user model:

# config/auth.py
config = {
    "default": "api",
    "guards": {
        "api":   {"strategy": "jwt", "secret": env("JWT_SECRET"),
                  "model": "database.models.user.User"},
        "admin": {"strategy": "jwt", "secret": env("ADMIN_JWT_SECRET"),
                  "model": "database.models.admin.Admin"},
    },
}

from forgeapi.auth import guard

CurrentAdmin = guard("admin").current_user()   # dependency bound to the admin guard

@route.get("/admin/stats")
async def stats(self, admin: CurrentAdmin): ...

token = guard("admin").token(admin)            # per-guard token operations

Manual registration remains as an escape hatch:

from forgeapi.auth import auth
from forgeapi.auth.guard import Guard
from forgeapi.auth.strategies import JWTStrategy

auth.register("api", Guard(name="api", strategy=JWTStrategy(secret_key="..."), user_model=User))
auth.set_default("api")

Custom strategies — auth.extend()

from forgeapi.auth import auth
from forgeapi.auth.strategies.base import AuthStrategy

class ApiKeyStrategy(AuthStrategy):
    @classmethod
    def from_config(cls, cfg: dict) -> "ApiKeyStrategy": ...

auth.extend("apikey", ApiKeyStrategy)
# now usable in config/auth.py: {"strategy": "apikey", ...}

Strategies raise only ForgeAPIAuthError subclasses — never HTTPException, never touch the DB. That structural rule is what keeps all strategies behaviourally identical at the edges.

JWT strategy

Reads Authorization: Bearer <token>. Implements TokenIssuer + RefreshCapable.

Config key	Default	Description
secret	—	Signing secret (use env("JWT_SECRET"))
secret_env	"JWT_SECRET"	Env var name to read when secret is not set
algorithm	"HS256"	JWT algorithm
access_ttl	30	Access token TTL, minutes
refresh_ttl	7	Refresh token TTL, days

Cookie strategy

Stores a signed JSON session in an HttpOnly cookie (HMAC). Implements SessionIssuer.

Config key	Default	Description
secret	—	HMAC secret (use env("COOKIE_SECRET"))
secret_env	"COOKIE_SECRET"	Env var name to read when secret is not set
cookie_name	"session"	Cookie name
max_age	3600	Session lifetime, seconds
httponly	True	HttpOnly flag
secure	True	Set False only for local HTTP development
samesite	"lax"	SameSite policy

auth.set_cookie(response, {"sub": str(user.id)})
auth.delete_cookie(response)

Telegram strategy

Validates initData from a Telegram Mini App. No login endpoint needed.

Config key	Default	Description
bot_token	—	Bot token, or a list of tokens (use env("BOT_TOKEN"))
bot_token_env	"BOT_TOKEN"	Env var name to read when bot_token is not set
max_age	86400	Max initData age, seconds

Client sends window.Telegram.WebApp.initData via X-Telegram-Init-Data header or Authorization: tma <initData>.

---

6. Pagination

Two flavors: QuerySet .paginate() (recommended), offset (classic), and cursor (stable on inserts/deletes).

QuerySet .paginate()

The cleanest way — call .paginate() directly on any QuerySet:

from fastapi import Request
from forgeapi.controllers import Controller, route

class PostController(Controller):
    prefix = "/posts"
    tags   = ["posts"]
    schema = PostResponse

    @route.get("/", response_model=None)
    async def index(self, request: Request):
        return await Post.all().order_by("-created_at").paginate(request, PostResponse)

    @route.get("/published", response_model=None)
    async def published(self, request: Request):
        return await Post.filter(is_published=True).paginate(request, PostResponse)

Query params: ?page=1&per_page=20. Response:

{
  "data": [...],
  "meta": {
    "current_page": 1,
    "per_page": 20,
    "total": 47,
    "last_page": 3,
    "from": 1,
    "to": 20
  },
  "links": {
    "prev": null,
    "next": "/posts?page=2&per_page=20"
  }
}

schema is optional — omit to get raw ORM objects.

Offset pagination

Classic DI-based pagination:

from forgeapi.pagination import Pagination

@route.get("/")
async def index(self, pagination: Pagination, request: Request):
    return await pagination.paginate(Post.all().order_by("-created_at"), PostResponse, request)

pagination.page, pagination.limit, pagination.offset are available as attributes.

Cursor pagination

No OFFSET — stable and O(1) on large datasets:

from forgeapi.pagination import CursorPagination

@route.get("/")
async def index(self, p: CursorPagination, request: Request):
    return await p.paginate(Post.all(), PostResponse, request, order_by="-id")

Query params: ?cursor=<token>&per_page=20.

Configuration

# config/pagination.py
config = {
    "default_limit": 20,
    "max_limit": 100,
}

Always configured by Core — defaults apply when the file is absent.

---

7. Broadcasting

BroadcastManager is the universal event bus for both in-process and cross-project messaging. It supports two modes and a pluggable driver system (Redis now, RabbitMQ planned).

BroadcastManager

from forgeapi import BroadcastManager

broadcast = BroadcastManager(
    driver="redis",
    url="redis://localhost:6379",
    namespace="shop",
    mode="stream",   # "pubsub" | "stream"
    maxlen=1000,     # stream mode: keep last N messages per key
)

Arg	Default	Description
driver	"redis"	Transport backend. RabbitMQ planned.
url	"redis://localhost:6379"	Broker connection URL
namespace	"forge"	Prefix for all channel/stream keys
mode	"pubsub"	"pubsub" = fire-and-forget, "stream" = persistent
maxlen	None	Stream mode: max messages per stream key

Registering handlers

@broadcast.on("order:created")
async def handle(data: dict) -> None:
    print(data["id"])

Core(app) auto-imports all modules in listeners_dir so decorators execute at startup.

Pub/Sub mode

Fire-and-forget fan-out. Fast, but messages are lost if subscriber is offline.

broadcast = BroadcastManager(driver="redis", url="redis://localhost:6379", namespace="shop")

@broadcast.on("order:created")
async def handle(data: dict) -> None:
    await telegram.send(f"Order #{data['id']}")

@asynccontextmanager
async def lifespan(app):
    await broadcast.connect()   # no group/consumer needed
    yield
    await broadcast.disconnect()

Stream mode

Persistent delivery via Redis Streams. Messages survive restarts. Each consumer group receives all messages independently.

broadcast = BroadcastManager(
    driver="redis",
    url="redis://localhost:6379",
    namespace="shop",
    mode="stream",
    maxlen=1000,
)

@broadcast.on("order:created")
async def handle(data: dict) -> None:
    await warehouse.fulfill(data["order_id"])

@asynccontextmanager
async def lifespan(app):
    await broadcast.connect(group="backend", consumer="worker-1")
    yield
    await broadcast.disconnect()

	mode="pubsub"	mode="stream"
Persistence	None	Stored until all groups ACK
Delivery	All subscribers simultaneously	Each group independently
Offline workers	Miss messages	Catch up on reconnect
Horizontal scaling	—	Multiple consumers per group share load

Cross-project example

Both projects use the same namespace — no shared Python classes needed, only plain dict.

# Project A — publisher (e.g. backend API)
broadcast = BroadcastManager(driver="redis", url="redis://...", namespace="shop", mode="stream")
await broadcast.emit("order:created", {"id": 42, "total": 99.0})

# Project B — consumer (e.g. Telegram bot, separate service)
broadcast = BroadcastManager(driver="redis", url="redis://...", namespace="shop", mode="stream")

@broadcast.on("order:created")
async def handle(data: dict) -> None:
    await bot.send(f"New order #{data['id']}")

await broadcast.connect(group="telegram-bot", consumer="worker-1")
await asyncio.sleep(float("inf"))

Lifecycle

# FastAPI
@asynccontextmanager
async def lifespan(app):
    await broadcast.connect(group="backend", consumer="worker-1")  # stream
    # await broadcast.connect()                                     # pubsub
    yield
    await broadcast.disconnect()

# Standalone script
async def main():
    await broadcast.connect(group="bot", consumer="w1")
    try:
        await asyncio.sleep(float("inf"))
    finally:
        await broadcast.disconnect()

---

8. Controllers

Controllers group routes. Core auto-discovers all *_controller.py files in controllers_dir (recursively) and registers their routers under base_prefix.

Base pattern

from fastapi import Request
from forgeapi.controllers import Controller, route
from forgeapi.auth import CurrentUser
from database.models.post import Post
from app.schemas.post import PostResponse, PostCreatePayload, PostUpdatePayload


class PostController(Controller):
    prefix = "/posts"
    tags   = ["posts"]
    schema = PostResponse   # auto response_model (see below)

    @route.get("/", response_model=None)
    async def index(self, request: Request):
        return await Post.all().order_by("-created_at").paginate(request, PostResponse)

    @route.post("/", status_code=201)
    async def create(self, payload: PostCreatePayload, user: CurrentUser):
        return await Post.create_from(payload, author_id=int(user.id))

    @route.get("/{id}")
    async def show(self, id: int):
        return await Post.find_or_fail(id)

    @route.patch("/{id}")
    async def update(self, id: int, payload: PostUpdatePayload, user: CurrentUser):
        post = await Post.find_or_fail(id)
        return await post.update_from(payload)

    @route.delete("/{id}", status_code=204)
    async def destroy(self, id: int):
        post = await Post.find_or_fail(id)
        await post.delete()

schema class var

Set schema on the controller class to auto-inject response_model on every route — no need to repeat it on each decorator:

class PostController(Controller):
    schema = PostResponse   # applied automatically to all routes

    @route.get("/")           # → response_model=PostResponse (auto)
    async def index(self): ...

    @route.get("/", response_model=None)   # override: disable for this route
    async def index(self, request: Request):
        return await Post.all().paginate(request, PostResponse)

    @route.delete("/{id}", status_code=204)  # skipped: no response body on 204
    async def destroy(self, id: int): ...

Rules:

• Routes with status_code=204 are skipped (no body)
• Routes with an explicit response_model=... keep their own value
• All other routes get response_model=schema

Route decorator

@route.get("/")
@route.post("/", status_code=201)
@route.put("/{id}")
@route.patch("/{id}")
@route.delete("/{id}", status_code=204)

# explicit form — multiple methods
@route("/", methods=["GET", "POST"])

All kwargs are forwarded to FastAPI (response_model, summary, dependencies, etc.).

Auto-prefix and namespace

If prefix is not set, it is derived from the class name:

Class	Auto prefix
PostController	/posts
AdminUserController	/admin/users
SuperAdminOrderController	/super/admin/orders

Final URL = base_prefix + controller.prefix + route path

forgeapi make:controller Post          # controllers/post_controller.py
forgeapi make:controller AdminUser     # controllers/admin/user_controller.py

---

9. Schemas

Base classes

from forgeapi import BaseSchema, BaseCreateSchema, BaseUpdateSchema

BaseSchema — response schemas. Has id, created_at, updated_at. from_attributes=True reads directly from Tortoise instances.

class PostResponse(BaseSchema):
    title: str
    body:  str

PostResponse.model_validate(post)

BaseCreateSchema — POST payloads. Plain BaseModel subclass.

class PostCreatePayload(BaseCreateSchema):
    title: str
    body:  str

BaseUpdateSchema — PATCH payloads. Enforces that every field is Optional at class definition time.

class PostUpdatePayload(BaseUpdateSchema):
    title: str | None = None
    body:  str | None = None

Schema directories

schemas/
  payload/
    post.py       # PostCreatePayload, PostUpdatePayload
  response/
    post.py       # PostResponse

generate:schema

forgeapi generate:schema Post --payload             # Create + Update payloads
forgeapi generate:schema Post --response            # PostResponse
forgeapi generate:schema Post --payload --response  # both
forgeapi generate:schema Post --payload -crud       # all four classes

---

10. Policies

Policies encapsulate authorization logic per resource.

Defining a policy

# app/policies/post_policy.py
from forgeapi.policies import Policy
from forgeapi import gate

@gate.policy(Post)
class PostPolicy(Policy):
    async def before(self, user, action: str) -> bool | None:
        if await user.has_role("admin"):
            return True   # admins bypass all checks
        return None       # continue to action method

    async def view(self, user, post) -> bool:
        return True

    async def create(self, user) -> bool:
        return user is not None

    async def update(self, user, post) -> bool:
        return post.author_id == int(user.id)

    async def delete(self, user, post) -> bool:
        return post.author_id == int(user.id)

@gate.policy(Post) registers the class — no separate step needed.

Using in controllers

from forgeapi import gate

@route.patch("/{id}")
async def update(self, id: int, payload: PostUpdatePayload, user: CurrentUser):
    post = await Post.find_or_fail(id)
    await gate.authorize(user, "update", post)   # raises HTTP 403 if denied
    return await post.update_from(payload)

Checking without raising

if await gate.allows(user, "delete", post): ...
if await gate.denies(user, "update", post): ...

Auto-discovery

gate.discover("app/policies")   # imports all *_policy.py files

Gate methods

Method	Description
gate.authorize(user, action, subject)	Raise HTTP 403 if denied
gate.allows(user, action, subject)	Return bool
gate.denies(user, action, subject)	Inverse of allows
gate.policy(ModelClass)	Decorator — register policy for a model
gate.register(ModelClass, PolicyClass)	Explicit registration
gate.discover(dir)	Auto-import all *_policy.py files

subject can be a model instance (update/delete) or a model class (create).

---

11. ModelMixin

ModelMixin adds ORM shortcuts and enables .paginate() on every QuerySet. Mix it alongside tortoise.Model:

from tortoise import Model, fields
from forgeapi import ModelMixin
from forgeapi.database import scope

class Post(ModelMixin, Model):
    id           = fields.IntField(pk=True)
    title        = fields.CharField(max_length=255)
    body         = fields.TextField()
    is_published = fields.BooleanField(default=False)
    author_id    = fields.IntField()
    created_at   = fields.DatetimeField(auto_now_add=True)
    updated_at   = fields.DatetimeField(auto_now=True)

    class Meta:
        table = "posts"

    @scope
    def published(qs):
        return qs.filter(is_published=True)

    @scope
    def by_author(qs, author_id: int):
        return qs.filter(author_id=author_id)

Use @scope (not @classmethod) for reusable filters — see §12 for details. Post.observe(PostObserver) registers lifecycle hooks — see §13.

Methods

find_or_fail(id, field="id") — get by field or raise HTTP 404:

post = await Post.find_or_fail(42)
post = await Post.find_or_fail("my-slug", field="slug")

create_from(payload, **extra) — create from a Pydantic schema:

post = await Post.create_from(payload)
post = await Post.create_from(payload, author_id=int(user.id))

None values are excluded — no accidental null writes.

update_from(payload, **extra) — partial update from schema, then save():

await post.update_from(payload)
await post.update_from(payload, updated_by=int(user.id))

QuerySet .paginate()

ModelMixin injects ForgeManager automatically so .paginate() is available on every QuerySet chain:

# Full paginated response
result = await Post.all().order_by("-created_at").paginate(request, PostResponse)

# With filters
result = await Post.published().paginate(request, PostResponse)
result = await Post.by_author(user_id).order_by("-created_at").paginate(request, PostResponse)

# Without schema — returns raw ORM objects
result = await Post.all().paginate(request)

Before vs after ModelMixin

# Before
data = payload.model_dump(exclude_none=True)
post = await Post.create(**data, author_id=int(user.id))

# After
post = await Post.create_from(payload, author_id=int(user.id))

# Before
for k, v in payload.model_dump(exclude_none=True).items():
    setattr(post, k, v)
await post.save()

# After
await post.update_from(payload)

---

12. Query Scopes

Laravel-style query scopes — reusable, chainable filters defined on the model.

from tortoise import Model, fields
from forgeapi.database import ModelMixin, scope

class Post(ModelMixin, Model):
    title = fields.CharField(max_length=255)
    is_published = fields.BooleanField(default=False)
    author_id = fields.IntField()

    @scope
    def published(qs):
        return qs.filter(is_published=True)

    @scope
    def by_author(qs, author_id: int):
        return qs.filter(author_id=author_id)

Using scopes

Call scopes directly on the queryset — they chain like regular filters:

# class-level (creates a fresh queryset)
posts = await Post.published()
posts = await Post.by_author(42)

# chaining
posts = await Post.published().order_by("-created_at").limit(20)
posts = await Post.all().published().by_author(42)

The @scope decorator registers the method in _scopes on the model class and makes it available on any queryset of that model via __getattr__. No base queryset class change required.

---

13. Model Observers

Lifecycle hooks fired by Tortoise signals — audit logs, cache invalidation, notifications — without scattering the logic across controllers.

from forgeapi.database import ModelObserver

class PostObserver(ModelObserver):
    async def creating(self, instance, **kw): ...   # before first save
    async def created(self, instance, **kw): ...    # after first save
    async def updating(self, instance, **kw): ...   # before update save
    async def updated(self, instance, **kw): ...    # after update save
    async def saving(self, instance, **kw): ...     # before any save
    async def saved(self, instance, **kw): ...      # after any save
    async def deleting(self, instance, **kw): ...   # before delete
    async def deleted(self, instance, **kw): ...    # after delete

Override only the hooks you need — ModelObserver provides empty defaults for all of them.

Registering

# class-based (instantiated automatically)
Post.observe(PostObserver)

# instance-based (useful for inline/test observers)
class _Audit(ModelObserver):
    async def created(self, instance, **kw):
        await AuditLog.create(action="post.created", target_id=instance.id)

Post.observe(_Audit())

Example — cache invalidation

class PostObserver(ModelObserver):
    async def saved(self, instance, **kw):
        await Cache.forget(f"post:{instance.id}")
        await Cache.forget("posts:index")

    async def deleted(self, instance, **kw):
        await Cache.forget(f"post:{instance.id}")

Post.observe(PostObserver)

Register observers at startup (e.g. in main.py or a custom Provider's boot()) — before any requests arrive.

---

14. Permissions

Spatie-style roles and permissions using polymorphic pivot tables.

How it works

Two shared tables store all assignments:

model_has_roles             model_has_permissions
──────────────────          ──────────────────────
model_type = "user"         model_type = "user"
model_id   = 42             model_id   = 42
role_id    = 1              permission_id = 3

model_type is the lowercase class name. Adding permissions to a new model requires zero new migrations.

Setup

1. Add PermissionsMixin to your model:

from tortoise import fields
from forgeapi.permissions import PermissionsMixin

class User(PermissionsMixin):
    id       = fields.IntField(pk=True)
    username = fields.CharField(max_length=150, unique=True)
    email    = fields.CharField(max_length=255, unique=True)

    class Meta:
        table = "users"

2. Add permissions models to TORTOISE_ORM:

TORTOISE_ORM = {
    "apps": {
        "models": {
            "models": ["database.models", "forgeapi.permissions.models"],
            ...
        },
    },
}

3. Nothing to register — Core(app) scans models_dir and activates permissions automatically when it finds the single PermissionsMixin subclass (silently skips when there is none).

4. Run migrations:

forgeapi db:makemigrations && forgeapi db:migrate

PermissionsMixin

# Checking
await user.can("edit:posts")                        # True if has any (direct or via role)
await user.can("edit:posts", "admin")               # OR logic
await user.cannot("delete:users")
await user.has_all_permissions("read", "write")     # AND logic
await user.has_role("admin")
await user.has_all_roles("admin", "editor")

# Granting / revoking
await user.give_permission("edit:posts", "delete:posts")
await user.revoke_permission("delete:posts")
await user.assign_role("admin", "editor")
await user.remove_role("editor")

# Listing
await user.get_all_permissions()   # → ["edit:posts", ...]
await user.get_role_names()        # → ["admin", "editor"]

Dependencies

from forgeapi.permissions import require_permission, require_role

@route.delete("/{id}")
async def destroy(self, id: int, user=require_permission("delete:posts")): ...

@route.post("/")
async def create(self, payload, user=require_permission("create:posts", "admin")): ...   # OR

@route.get("/admin/stats")
async def stats(self, user=require_role("admin")): ...

Both dependencies also check db_user.is_active when the field exists — inactive users receive 401.

Role and Permission models

from forgeapi.permissions.models import Role, Permission

role = await Role.find_or_create("editor")
await role.give_permission("edit:posts", "read:posts")
await role.has_permission("edit:posts")   # → bool

# Filtering by role
users = await (await User.with_role("admin"))
users = await (await User.without_role("admin"))

---

15. Cache

Async key-value cache. Two drivers: memory (default, no dependencies) and redis (persistent, shared across workers).

Core auto-configures Cache from config/cache.py on startup (memory driver by default).

from forgeapi import Cache

Basic operations

await Cache.set("key", value, ttl=60)        # store for 60 seconds
await Cache.get("key")                        # → value or None
await Cache.get("key", default="fallback")
await Cache.has("key")                        # → bool
await Cache.missing("key")                    # → bool (inverse)
await Cache.forget("key")                     # delete → bool
await Cache.flush()                           # clear all
await Cache.forever("key", value)             # store without TTL

Common patterns

remember() — get or compute and store:

posts = await Cache.remember(
    "posts:popular",
    fn=lambda: Post.filter(is_published=True).limit(10),
    ttl=300,
)

# async fn works too
async def fetch_stats():
    return await compute_heavy_stats()

stats = await Cache.remember("stats:global", fn=fetch_stats, ttl=3600)

pull() — get and immediately delete (one-time tokens):

token = await Cache.pull(f"reset:token:{user_id}")
if token != submitted_token:
    raise HTTPException(400, "Invalid token")

Counters:

await Cache.increment("views:post:42")         # → int
await Cache.increment("views:post:42", amount=5)
await Cache.decrement("stock:item:5")

Use in a controller:

@route.get("/popular", response_model=None)
async def popular(self, request: Request):
    cached = await Cache.get("posts:popular")
    if cached:
        return cached
    result = await Post.filter(is_published=True).paginate(request, PostResponse)
    await Cache.set("posts:popular", result, ttl=60)
    return result

Drivers

Driver	When to use
memory	Single process, development, testing. Resets on restart.
redis	Multiple workers, production. Requires pip install forge-kits[redis]

Cache configuration

# config/cache.py
config = {
    "driver": "memory",                        # "memory" | "redis"
    "prefix": "",                              # key prefix, e.g. "myapp:"
    "ttl": None,                               # default TTL in seconds (None = no expiry)
    "redis_url": "redis://localhost:6379/0",   # used when driver = "redis"
}

Programmatic setup (without Core):

from forgeapi import Cache

Cache.configure(driver="redis", prefix="myapp:", ttl=3600, redis_url="redis://localhost:6379/1")

---

16. Storage

File storage abstraction with Local and S3-compatible drivers. Configured via config/storage.py — the Storage facade boots automatically when the file exists.

Storage configuration

# config/storage.py
from forgeapi import env

config = {
    "driver": "local",        # "local" | "s3"
    "root": "storage/app",    # local: filesystem root
    "base_url": "/storage",   # local: public URL prefix
}

Install the storage extras when needed:

pip install forge-kits[s3]       # S3 / MinIO / Cloudflare R2
pip install forge-kits[images]   # ImageProcessor (Pillow)

Storage basic operations

from forgeapi import Storage

# write
path = await Storage.put("avatars/user-1.jpg", image_bytes)

# read
data: bytes | None = await Storage.get("avatars/user-1.jpg")

# check / delete
exists: bool = await Storage.exists("avatars/user-1.jpg")
await Storage.delete("avatars/user-1.jpg")

# list
files: list[str] = await Storage.list("avatars/")

# public URL
url: str = Storage.url("avatars/user-1.jpg")   # "/storage/avatars/user-1.jpg"

put() returns the stored path. get() returns None when the file is missing. All file I/O is non-blocking (asyncio.to_thread).

Multiple disks

from forgeapi.storage import LocalDriver, S3Driver

Storage.add_disk("public", LocalDriver(root="public/", base_url="/public"))
Storage.add_disk("backups", S3Driver(bucket="my-backups", region="eu-central-1",
                                     access_key="...", secret_key="..."))

# switch disk
await Storage.disk("backups").put("dump.sql.gz", data)
url = Storage.disk("public").url("logo.png")

S3 / S3-compatible

# config/storage.py
from forgeapi import env

config = {
    "driver": "s3",
    "bucket": "my-bucket",
    "region": "eu-central-1",
    "access_key": env("AWS_ACCESS_KEY_ID"),
    "secret_key": env("AWS_SECRET_ACCESS_KEY"),
    # MinIO / Cloudflare R2:
    # "endpoint_url": env("S3_ENDPOINT_URL"),
    # "acl": "public-read",
}

endpoint_url makes it work with MinIO, Cloudflare R2, or any S3-compatible service — same API, different endpoint.

ImageProcessor

Fluent image pipeline built on Pillow. Install: pip install forge-kits[images]

from forgeapi import ImageProcessor

# from raw bytes
pipeline = ImageProcessor.process(raw_bytes)

# from FastAPI UploadFile
pipeline = await ImageProcessor.from_upload(upload_file)

# transform
pipeline = (
    pipeline
    .resize(800, 600)           # fit into 800×600 (preserves ratio)
    .resize(800, 600, crop=True)# crop-fill to exact dimensions
    .thumbnail(256)             # shrink to 256px max side
    .quality(85)                # JPEG quality
    .convert("WEBP")            # format conversion
    .grayscale()                # desaturate
)

# get bytes
data: bytes = pipeline.to_bytes()

# store directly to Storage (returns stored path)
path = await pipeline.store("avatars/user-1", extension="webp")
path = await pipeline.store(directory="thumbs", extension="jpg")
path = await pipeline.store("hero.jpg", disk="public")

Pillow is imported lazily — ImportError with a clear hint if not installed.

---

17. Support

Utility helpers for formatting numbers, strings, and datetimes.

from forgeapi import Number, Str, Time

Number

Number.format(1234567.89)             # "1,234,567.89"
Number.format(1234.5, decimals=0)     # "1,235"
Number.format(1234.5, thousands=".", decimal=",")  # "1.234,50" (EU style)

Number.currency(99.9)                 # "99.90"
Number.currency(100.00044443)         # "100.00"  — messy float, clean output
Number.currency(1234.5)               # "1,234.50"

Number.file_size(1024)                # "1.0 KB"
Number.file_size(1048576)             # "1.0 MB"
Number.file_size(1073741824)          # "1.0 GB"
Number.file_size(1024, unit="MB")     # "0.0 MB"  — forced unit

Number.percent(0.754)                 # "75.4%"
Number.percent(0.5, decimals=0)       # "50%"

Number.abbreviate(1500000)            # "1.5M"
Number.abbreviate(2300)               # "2.3K"
Number.abbreviate(500)                # "500"

Number.clamp(150, 0, 100)             # 100
Number.clamp(-5, 0, 100)              # 0

Str

Str.limit("Hello world", 5)           # "Hello..."
Str.limit("Hello world", 5, " →")    # "Hello →"

Str.slug("Hello World!")              # "hello-world"
Str.slug("  Café & Bistro  ")        # "café-bistro"

Str.random(16)                        # "Xk9mR2pLqT8vNcYw"
Str.random(8, alphabet="0123456789") # "48291037"

Str.title("hello world")              # "Hello World"
Str.snake("HelloWorld")               # "hello_world"
Str.snake("hello-world")              # "hello_world"
Str.camel("hello_world")              # "helloWorld"
Str.pascal("hello_world")             # "HelloWorld"

Str.truncate_words("one two three four", 2)  # "one two..."

Str.strip_tags("<b>Hello</b> <i>world</i>")  # "Hello world"

Str.mask("1234567890", start=4)       # "1234******"
Str.mask("1234567890", char="•", start=0, length=4)  # "••••567890"

Str.contains("Hello World", "world", case_sensitive=False)  # True
Str.starts_with("Hello", "He")        # True
Str.ends_with("Hello", "lo")          # True

Time

Time.now()                            # datetime UTC
Time.now("Europe/Kyiv")               # datetime in timezone

Time.parse("2025-07-14")              # → datetime
Time.parse("2025-07-14T12:00:00Z")    # → datetime
Time.parse(1720958400)                # → datetime from Unix timestamp

Time.format(dt)                       # "2025-07-14 12:00:00"
Time.format(dt, "%d/%m/%Y")           # "14/07/2025"

Time.to_timezone(dt, "US/Eastern")    # convert timezone
Time.to_timezone(dt, "Europe/Kyiv")

Time.timestamp(dt)                    # → int (Unix timestamp)

Time.add(dt, days=1, hours=3)         # → datetime
Time.subtract(dt, days=7)             # → datetime
Time.diff_in_days(dt1, dt2)           # → int (absolute)
Time.diff_in_seconds(dt1, dt2)        # → int (absolute)

Time.human(dt)                        # "just now" / "5 minutes ago" / "in 3 hours"
Time.human(dt, relative_to=other_dt)  # relative to a specific point

Time.is_past(dt)                      # → bool
Time.is_future(dt)                    # → bool

Time.start_of_day(dt)                 # 00:00:00.000000
Time.end_of_day(dt)                   # 23:59:59.999999

---

18. Logger

forge-kits includes a structured logger so you don't need to call logging.getLogger(__name__) everywhere.

from forgeapi import Log

Log.info("Order created", order_id=order.id, user_id=user.id)
Log.warning("Payment retry", order_id=order.id, attempt=3)
Log.error("Stripe failed", order_id=order.id, reason=str(e))

Context is appended as key=value pairs:

INFO  Order created | order_id=42  user_id=7
ERROR Stripe failed | order_id=42  reason='Card declined'

Named channels

auth_log = Log.channel("auth")
auth_log.debug("Token decoded", user_id=42)

Methods

Method	Level
Log.debug(msg, **ctx)	DEBUG
Log.info(msg, **ctx)	INFO
Log.warning(msg, **ctx)	WARNING
Log.error(msg, **ctx)	ERROR
Log.critical(msg, **ctx)	CRITICAL
Log.channel(name)	Returns child Logger

Log wraps Python's standard logging — existing handlers keep working unchanged.

---

19. Middleware

Two extension points: global middleware wraps every request, guards scope to a route or controller.

Custom global middleware

from forgeapi import Middleware
from fastapi import Request, Response

class TimingMiddleware(Middleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        import time
        start = time.perf_counter()
        response = await call_next(request)
        response.headers["X-Process-Time"] = f"{time.perf_counter() - start:.3f}s"
        return response

Register via config (primary path) or core.use() (manual escape hatch):

# config/http.py
from app.middleware import TimingMiddleware, TenantMiddleware
config = {
    "middleware": [
        TimingMiddleware,
        (TenantMiddleware, {"default_tenant": "acme"}),   # (cls, kwargs) tuple
    ],
}

core.use(TimingMiddleware)
core.use(TenantMiddleware, default_tenant="acme")

Guards — per-route / per-controller

from forgeapi import Guard
from fastapi import HTTPException, Request

class ApiKeyGuard(Guard):
    def __init__(self, header: str = "X-API-Key"):
        self.header = header

    async def handle(self, request: Request) -> None:
        if not request.headers.get(self.header):
            raise HTTPException(403, "Missing API key")

Per-route:

@route.delete("/{id}", dependencies=[Depends(ApiKeyGuard())])
async def destroy(self, id: int): ...

Per-controller:

class AdminController(Controller):
    guards = [ApiKeyGuard("X-Admin-Key")]

Guards can use FastAPI dependencies directly in handle:

from forgeapi.auth import CurrentUser

class ActiveUserGuard(Guard):
    async def handle(self, user: CurrentUser) -> None:
        if not user.is_active:
            raise HTTPException(403, "Account disabled")

Built-in middleware

All configured in config/http.py:

# config/http.py
config = {
    "cors": ["*"],        # True → all origins; list → specific; False → off
    "rate_limit": 60,     # req/min per IP; True → 60; False → off
    "request_id": True,   # inject X-Request-ID header
    "access_log": True,   # log method/path/status/duration per request
    "middleware": [],     # custom classes or (cls, kwargs) tuples
}

Key	Default	Description
cors	False	True = allow all; list = specific origins
rate_limit	False	Sliding window per IP — returns 429 with Retry-After
request_id	False	Injects X-Request-ID header; access via request.state.request_id
access_log	True	Logs method, path, status, duration

---

20. Settings

BaseAppSettings wraps pydantic-settings with .env file loading:

from forgeapi.settings import BaseAppSettings

class Settings(BaseAppSettings):
    database_url: str
    redis_url: str | None = None
    jwt_secret: str
    debug: bool = False

settings = Settings()   # reads .env automatically

Sensitive field masking — fields containing password, secret, token, key, auth, credential show *** in repr:

>>> print(settings)
Settings(database_url='postgresql://...', jwt_secret='***', debug=True)

---

21. Seeders

Seeders populate the database with initial or test data.

forgeapi make:seed User
# → database/seeds/user_seeder.py

from forgeapi.database import Seeder
from database.models import User

class UserSeeder(Seeder):
    async def run(self) -> None:
        await User.get_or_create(
            username="admin",
            defaults={"email": "admin@example.com", "is_active": True},
        )

Running:

forgeapi db:seed              # run all seeders
forgeapi db:seed User         # run UserSeeder only
forgeapi db:seed User Post    # run in order

db:seed wraps each seeder in a transaction — rollback on error.

---

22. Scheduler

DB-backed task scheduler — Laravel-style. Jobs are defined in code (schedule.py); state (next run time, last status, errors) is persisted in the jobs table via Tortoise. No external queue or APScheduler dependency required.

pip install forge-kits   # Scheduler is included in the base package

Setup

Add forgeapi.scheduling to the models list in config/database.py so Tortoise creates the jobs table automatically:

# config/database.py
TORTOISE_ORM = {
    "connections": {...},
    "apps": {
        "models": {
            "models": [
                "database.models",
                "forgeapi.scheduling",        # ← add this
                "forgeapi.permissions.models",
            ],
            ...
        }
    },
}

If you use the CLI (schedule:work / schedule:run), the table is created automatically via generate_schemas(safe=True) — no manual migration step needed.

schedule.py

forgeapi init creates a schedule.py at the project root. Define all jobs there:

# schedule.py
from forgeapi import Scheduler

scheduler = Scheduler()

scheduler.call(send_daily_report).daily_at("09:00").name("newsletter")
scheduler.call(cleanup_temp_files).every(30).name("cleanup")
scheduler.call(weekly_backup).weekly_on("sunday", at="03:00").name("backup")
scheduler.call(sync_rates).hourly().name("sync-rates")
scheduler.call(health_ping).every_minute().name("health")

All callables — sync or async — are accepted:

async def send_daily_report():
    users = await User.filter(newsletter=True).all()
    ...

Timing methods

Method	When it runs
.every_minute()	Every minute
.every(n)	Every n minutes
.hourly()	Every hour
.every_hours(n)	Every n hours
.daily()	Daily at midnight
.daily_at("HH:MM")	Daily at the given time
.weekly()	Every Monday at midnight
.weekly_on("day", at="HH:MM")	Named weekday at the given time

.name("label") sets the unique key used in the DB and in CLI commands. Defaults to the function name.

Running via CLI

# Run due tasks once — use from cron (every minute)
forgeapi schedule:run

# Run a specific task manually, regardless of schedule
forgeapi schedule:run newsletter

# Start infinite dev loop — checks for due tasks, sleeps until next run
forgeapi schedule:work

# Show all registered tasks and their DB state
forgeapi schedule:list

schedule:work is the recommended dev runner. For production, add a cron job:

* * * * * cd /path/to/project && forgeapi schedule:run >> /var/log/scheduler.log 2>&1

FastAPI lifespan integration

If you prefer to run the scheduler in-process instead of a separate process:

# main.py
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
from forgeapi import Core
from schedule import scheduler   # your schedule.py

@asynccontextmanager
async def lifespan(app):
    task = asyncio.create_task(scheduler.run())
    yield
    task.cancel()
    await asyncio.gather(task, return_exceptions=True)

app = FastAPI(lifespan=lifespan)
core = Core(app)

scheduler.run() calls sync() on startup (upserts all jobs to DB), then loops indefinitely — sleeping until the soonest scheduled task, then calling run_due().

ScheduledTask model

The jobs table columns:

Column	Type	Description
name	VARCHAR(255) unique	Job name (.name("label"))
schedule_type	VARCHAR(20)	"interval" / "daily" / "weekly"
schedule_config	JSON	Schedule parameters (e.g. {"minutes": 30})
is_enabled	BOOLEAN	Toggle without removing from code
next_run_at	DATETIME	When the job will next run
last_run_at	DATETIME	When it last ran
last_status	VARCHAR(20)	"success" / "failed"
last_error	TEXT	Exception message on failure

---

23. CLI reference

forgeapi --help
forgeapi make:controller -h

Project scaffolding

forgeapi init my-project

Code generation

forgeapi make:controller Post          # controllers/post_controller.py
forgeapi make:controller AdminUser     # controllers/admin/user_controller.py
forgeapi make:controller Post --ms     # + model + schema stubs
forgeapi make:model Post
forgeapi make:model Post -cs           # + controller + schema
forgeapi make:seed User                # database/seeds/user_seeder.py
forgeapi make:schema Post              # stub schemas (3 classes with pass)

Typed schema generation (from existing model)

forgeapi generate:schema Post --payload             # Create + Update payloads
forgeapi generate:schema Post --response            # PostResponse
forgeapi generate:schema Post --payload --response  # both
forgeapi generate:schema Post --payload -crud       # all four incl. DeletePayload
forgeapi generate:schema Post --payload --cu        # Create + Update only

DB commands

Never run aerich directly — always use forgeapi db:*

forgeapi db:init
forgeapi db:makemigrations
forgeapi db:makemigrations -n add_email_field
forgeapi db:migrate
forgeapi db:downgrade
forgeapi db:history
forgeapi db:seed
forgeapi db:fresh             # TRUNCATE all tables (asks confirmation)
forgeapi db:fresh --force     # DROP all tables (irreversible)

Dev server

Never use uvicorn directly — always use forgeapi runserver

forgeapi runserver
forgeapi runserver --reload
forgeapi runserver --port 9000 --host 0.0.0.0 --reload

Scheduler

forgeapi schedule:run             # run all due tasks once (use from cron)
forgeapi schedule:run <name>      # run a specific task manually
forgeapi schedule:work            # start infinite dev loop
forgeapi schedule:list            # show all tasks and their DB state

Inspection

forgeapi routers   # list all registered routes (METHOD, PATH, HANDLER)
forgeapi models    # list all Tortoise model classes, tables, and fields

---

24. Configuration reference (config/)

The only config format is a config/ directory of Python dict files (Laravel-style). Each config/<section>.py defines a module-level config = {...}; the filename is the section name. All sections are optional — Core(app) works with no config files at all. (forgeapi.toml is no longer supported — load_config raises with a migration hint.)

env() helper

from forgeapi import env

env("APP_DEBUG", False)   # reads env var; casts "true"/"false"/"null"
env("JWT_SECRET")         # → str | None

Known sections

# config/project.py
from forgeapi import env
config = {
    "name": "my-app",
    "version": "0.1.0",
    "debug": env("APP_DEBUG", False),   # enables Telescope — never in production
    "providers": [],                    # extra Provider classes / dotted paths
}

# config/structure.py
config = {
    "models_dir": "database/models",
    "controllers_dir": "app/controllers",
    "schemas_dir": "app/schemas",
    "listeners_dir": "app/listeners",
    "policies_dir": "app/policies",
    "seeds_dir": "database/seeds",
    "base_prefix": "/api/v1",
    "schedule_file": "schedule.py",   # path to schedule.py; used by schedule:* CLI commands
}

# config/http.py
config = {
    "cors": ["*"],
    "rate_limit": 60,
    "request_id": True,
    "access_log": True,
    "middleware": [],
}

# config/auth.py — see §5 for guard keys
config = {
    "default": "api",
    "guards": {
        "api": {"strategy": "jwt", "secret": env("JWT_SECRET"),
                "model": "database.models.user.User"},
    },
}

# config/pagination.py
config = {"default_limit": 20, "max_limit": 100}

# config/cache.py
config = {"driver": "memory", "prefix": "", "ttl": None,
          "redis_url": "redis://localhost:6379/0"}

# config/storage.py
config = {
    "driver": "local",        # "local" | "s3"
    "root": "storage/app",    # local: filesystem root
    "base_url": "/storage",   # local: public URL prefix
    # S3 / MinIO / R2:
    # "bucket": "my-bucket",
    # "region": "us-east-1",
    # "access_key": env("AWS_ACCESS_KEY_ID"),
    # "secret_key": env("AWS_SECRET_ACCESS_KEY"),
    # "endpoint_url": "",     # MinIO / R2 endpoint
    # "acl": "public-read",
}

config/database.py

The TORTOISE_ORM dict lives here — that's all the file needs. The loader derives the importable dotted path (config.database.TORTOISE_ORM) for the tortoise CLI from the file location (config/ is a namespace package):

# config/database.py
import os
from dotenv import load_dotenv

load_dotenv()

TORTOISE_ORM = {
    "connections": {
        "default": os.getenv("DATABASE_URL", "sqlite://./db.sqlite3"),
    },
    "apps": {
        "models": {
            "models": ["database.models", "forgeapi.permissions.models"],
            "default_connection": "default",
            "migrations": "database.migrations",
        }
    },
}

An explicit config = {"tortoise_orm": "app.settings.TORTOISE_ORM"} is only needed when the dict lives somewhere non-standard.

Custom sections

Any extra file becomes a section, reachable with dot access:

# config/services.py
config = {"stripe": {"key": env("STRIPE_KEY")}}

core.config.get("services.stripe.key", default="")
core.config.get("auth.guards.api.strategy")

Known sections are validated by Pydantic models; misconfiguration raises ForgeAPIConfigError with a hint. Feature enablement is decided by section presence — e.g. auth boots only when the project provides config/auth.py.

---

25. Telescope

Debug-only request inspector activated by "debug": True in config/project.py. Never use in production.

# config/project.py
from forgeapi import env
config = {"debug": env("APP_DEBUG", False)}

Captures per request: SQL queries, log output, dispatched events, custom jobs. Up to 200 entries in a circular buffer.

What Telescope captures

Field	Description
method, path, status	HTTP basics
headers	All headers — sensitive ones replaced with "***"
payload	Request body — sensitive fields masked
response_body	Response body — same masking, capped at 64 KB
duration_ms	Handler time
queries	SQL queries: text, params, duration, source location
logs	All logging calls during the request
events	Events dispatched during the request
jobs	Custom jobs recorded via record_job()

WebSocket live stream

Connect to ws://<host>/_forge/telescope/ws:

Direction	Message
Server → Client	{"type": "init", "data": [...]} — all current entries on connect
Server → Client	{"type": "entry", "data": {...}} — after each request
Client → Server	{"type": "clear"} — clear the store

Sensitive data masking

Headers Authorization, Cookie, X-API-Key, X-Telegram-Init-Data → "***"

Body fields containing: password, secret, token, key, auth, credential, access_token, refresh_token → "***"

Recording jobs

from forgeapi.telescope import record_job

record_job("ProcessPayment", status="done", duration_ms=45.2)
record_job("SendEmail", status="failed", error=str(exc))

No-op when called outside a Telescope request context.

---

26. MCP Server

forge-kits ships an MCP server that gives AI assistants (Claude Code, Cursor, etc.) direct access to API docs, code generation tools, and project structure scanning — without reading source files.

Install

pip install forge-kits[mcp]

This installs forgeapi-mcp as a CLI entry point.

---

Quick setup via CLI

The easiest way — no JSON editing required:

# Per-project — adds to .mcp.json in the current directory
claude mcp add forge-kits forgeapi-mcp

# Global — adds to ~/.claude/settings.json, available in all projects
claude mcp add forge-kits forgeapi-mcp --scope global

---

Global setup for Claude Code (manual)

Install once and use across all your projects. Edit the global Claude Code settings file:

macOS / Linux:

~/.claude/settings.json

Windows:

C:\Users\<your-username>\.claude\settings.json

Add the MCP server:

{
  "mcpServers": {
    "forge-kits": {
      "command": "forgeapi-mcp"
    }
  }
}

After saving, restart Claude Code. The tools are now available in every project.

---

Per-project setup

Useful when different projects use different versions of forge-kits, or when you want to use a virtualenv-specific install.

Create .mcp.json in your project root (can be committed to the repo):

{
  "mcpServers": {
    "forge-kits": {
      "command": "forgeapi-mcp"
    }
  }
}

If forgeapi-mcp is installed in a virtualenv, point to it directly:

macOS / Linux:

{
  "mcpServers": {
    "forge-kits": {
      "command": "/path/to/project/.venv/bin/forgeapi-mcp"
    }
  }
}

Windows:

{
  "mcpServers": {
    "forge-kits": {
      "command": "C:\\path\\to\\project\\.venv\\Scripts\\forgeapi-mcp.exe"
    }
  }
}

Priority: per-project .mcp.json takes precedence over the global ~/.claude/settings.json when both exist.

---

Available tools

Tool	Description
get_docs(topic)	Full API reference for a topic
get_example(pattern)	Complete working code for a pattern
generate_controller(name, routes)	Generate a Controller class
generate_event(name, fields)	Generate an Event class + listener
generate_schema(name, fields, mode)	Generate Pydantic schemas
scan_project(path)	Deep AST scan: models, controllers, schemas, events, listeners, seeders, deps
project_info(path)	Read forgeapi.toml and list project files

get_docs topics

Topic	Content
cheatsheet	Controller + queries + auth + events quick reference
workflow	CLI rules — which commands to use and which to avoid
core	Core constructor, options, startup sequence
controllers	Controller, @route, schema class var, auto-prefix
events	BroadcastManager, pub/sub, Streams, cross-project
auth	Guards, CurrentUser, strategies, token operations
permissions	PermissionsMixin, require_permission, roles
policies	Policy, Gate, authorize, allows, discover
pagination	QuerySet .paginate(), offset, cursor, configuration
schemas	BaseSchema, BaseCreateSchema, BaseUpdateSchema
middleware	Middleware, Guard, built-in middleware
cli	All CLI commands reference
config	forgeapi.toml full reference
models	ModelMixin, Tortoise field types, relationships
cache	Cache facade, drivers, remember/pull/increment
storage	Storage facade, LocalDriver, S3Driver, ImageProcessor
scheduler	Scheduler, ScheduledJob, timing methods, lifespan integration
scopes	@scope decorator, queryset chaining
observers	ModelObserver, lifecycle hooks
support	Number, Str, Time helpers
tortoise	Basic CRUD, filter, order, async gather
tortoise_advanced	Q objects, prefetch_related, transactions, raw SQL

get_example patterns

Pattern	Content
crud_controller	Full CRUD with ModelMixin + schema class var
jwt_auth	Login, refresh, protected routes
redis_event	Redis pub/sub fan-out event
stream_event	Redis Streams consumer
rbac	Full RBAC — model, seeder, protected routes
pagination	QuerySet .paginate() with filters
guard	API key, active-user, admin guards
cache	remember, pull, counters in a controller
storage	upload, resize, store, S3 config
observer	audit log, cache invalidation on model events
scheduler	lifespan task worker with daily/weekly jobs

generate_controller

generate_controller("Post", ["GET /", "POST /", "GET /{id}", "PATCH /{id}", "DELETE /{id}"])

scan_project

scan_project(".")

Returns a structured report of all models, controllers, schemas, events, listeners, seeders, pyproject.toml dependencies, and .env keys (values hidden). Call this at the start of every session on a forge-kits project.