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
   • 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. Queue
    • Defining a job
    • Dispatching
    • Running the worker
    • Failed jobs
24. CLI reference
25. Configuration reference (config/)
26. Telescope
    • What Telescope captures
    • WebSocket live stream
    • Sensitive data masking
    • Recording jobs
27. MCP Server
    • Install
    • Global setup for Claude Code
    • Per-project setup
    • Available tools

---

1. Quick start

pip install forge-kits

Optional dependencies

SQLite + auth are included in the base package. Install extras only for production DB drivers or optional features.

Extra	Installs	Use when
asyncpg	asyncpg	PostgreSQL
aiomysql	aiomysql	MySQL / MariaDB
redis	redis	Cache Redis driver / BroadcastManager events
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               # SQLite included by default
pip install forge-kits[asyncpg]      # PostgreSQL
pip install forge-kits[aiomysql]     # MySQL / MariaDB
pip install forge-kits[redis]        # Redis cache / events
pip install forge-kits[mcp]          # MCP server
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 (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 (COOKIE_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 (cookie / telegram)
    pagination.py            # default_limit, max_limit
    database.py              # TORTOISE_ORM dict lives here
    storage.py               # driver, root, base_url (or S3 config)
    broadcast.py             # redis url, mode (pubsub/stream), namespace — activates BroadcastProvider
    scheduler.py             # enabled, schedule file path — activates SchedulerProvider
    queue.py                 # enabled, queue name — activates QueueProvider
  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
Broadcasting	config/broadcast.py exists — creates global broadcast, connects on startup, disconnects on shutdown
Scheduler	config/scheduler.py exists — loads schedule.py, starts worker task on startup
Queue worker	config/queue.py exists — starts worker task on startup
Controllers	controllers_dir exists — all *_controller.py imported recursively
Event listeners	config/broadcast.py exists — all files in listeners_dir 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).

Available strategies: cookie (HMAC-signed session cookie), telegram (Telegram Mini App initData).

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 sessions

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": "cookie",           # cookie | telegram | custom
            "secret": env("COOKIE_SECRET"),
            "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 session 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: 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 session payload:

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

Auth facade — session operations

from forgeapi.auth import auth

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

Login / Logout endpoint pattern (cookie)

@route.post("/login")
async def login(self, payload: LoginPayload, response: Response) -> 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")
    auth.set_cookie(response, {"sub": str(user.id), "username": user.username})
    return {"message": "Logged in"}

@route.post("/logout")
async def logout(self, response: Response) -> dict:
    auth.delete_cookie(response)
    return {"message": "Logged out"}

Multiple guards

# config/auth.py
config = {
    "default": "api",
    "guards": {
        "api":   {"strategy": "cookie", "secret": env("COOKIE_SECRET"),
                  "model": "database.models.user.User"},
        "admin": {"strategy": "cookie", "secret": env("ADMIN_COOKIE_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): ...

Manual registration remains as an escape hatch:

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

auth.register("api", Guard(name="api", strategy=CookieStrategy(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.

Cookie strategy

Stores a signed JSON session in an HttpOnly cookie (HMAC-SHA256).

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

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).

Config-driven setup (with Core)

Add config/broadcast.py — Core(app) does the rest: creates the global broadcast proxy, imports all listeners, and manages connect/disconnect via FastAPI startup/shutdown hooks.

# config/broadcast.py
from forgeapi import env

config = {
    "enabled": True,
    "driver": "redis",
    "url": env("REDIS_URL", "redis://localhost:6379"),
    "namespace": "shop",
    "mode": "pubsub",       # "pubsub" | "stream"
    # "maxlen": 1000,       # stream mode: keep last N messages per key
    # "group": "backend",   # stream mode: consumer group name
    # "consumer": "worker-1",
}

# app/listeners/order_listener.py
from forgeapi.broadcasting import broadcast

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

# main.py — no lifespan needed, BroadcastProvider handles connect/disconnect
from fastapi import FastAPI
from forgeapi import Core

app = FastAPI()
core = Core(app)

Modes

	mode="pubsub"	mode="stream"
Persistence	None — messages lost if subscriber is offline	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

Standalone / Cross-project usage

Use BroadcastManager directly when broadcasting outside a Core-managed app (e.g. a standalone bot, a worker process, a publisher-only script):

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

# Project A — publisher (e.g. backend API)
broadcast = BroadcastManager(driver="redis", url="redis://...", namespace="shop", mode="stream")
await broadcast.connect()
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"))

Manual lifespan (without Core)

@asynccontextmanager
async def lifespan(app):
    await broadcast.connect(group="backend", consumer="worker-1")  # stream
    # await broadcast.connect()                                     # pubsub
    yield
    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
    cookie_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://...', cookie_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

In-process with Core (config-driven)

Add config/scheduler.py — Core(app) starts the scheduler as a background task on startup and cancels it on shutdown. No lifespan boilerplate needed.

# config/scheduler.py
config = {
    "enabled": True,
    "schedule": "schedule.py",   # path to your schedule.py
}

# main.py — scheduler starts automatically
from fastapi import FastAPI
from forgeapi import Core

app = FastAPI()
core = Core(app)

Manual lifespan (without Core)

If you need to run the scheduler in-process without using config/scheduler.py:

import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
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)

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. Queue

DB-backed async job queue. Jobs are stored in the queued_jobs table; permanently-failed ones land in failed_jobs. No external broker — Tortoise ORM only.

Setup

Add forgeapi.queue.models to TORTOISE_ORM so the tables are created:

# config/database.py
TORTOISE_ORM = {
    "apps": {
        "models": {
            "models": ["database.models", "forgeapi.queue.models"],
            ...
        }
    },
}

The CLI creates tables automatically via generate_schemas(safe=True) — no manual migration step needed.

Defining a job

from forgeapi.queue import Job

class SendWelcomeEmail(Job):
    queue = "default"   # optional — queue name
    max_tries = 3       # optional — retry attempts before moving to failed_jobs

    def __init__(self, user_id: int):
        self.user_id = user_id

    async def handle(self) -> None:
        user = await User.get(id=self.user_id)
        await mailer.send(user.email, "Welcome!")

Dispatching

from forgeapi.queue import dispatch

await dispatch(SendWelcomeEmail(user_id=user.id))             # immediate
await dispatch(SendWelcomeEmail(user_id=user.id), delay=60)   # available after 60 s
await dispatch(SendWelcomeEmail(user_id=user.id), queue="high")  # override queue

Running the worker

CLI (recommended for production):

forgeapi queue:work        # start infinite worker loop (press Ctrl+C to stop)
forgeapi queue:run         # process all currently pending jobs once (use from cron)

In-process with Core (config-driven):

Add config/queue.py — Core(app) starts the queue worker as a background task on startup.

# config/queue.py
config = {
    "enabled": True,
    "queue": "default",   # queue name to consume
}

# main.py — worker starts automatically
from fastapi import FastAPI
from forgeapi import Core

app = FastAPI()
core = Core(app)

Manual lifespan (without Core):

import asyncio
from contextlib import asynccontextmanager
from forgeapi.queue import Queue

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

Failed jobs

After max_tries attempts (with exponential backoff), the job is moved to failed_jobs.

forgeapi queue:failed           # list failed jobs
forgeapi queue:retry <id>       # move a failed job back to the queue
forgeapi queue:flush            # delete all failed jobs

DB tables

Table	Description
queued_jobs	Pending/reserved jobs (queue, payload, attempts, reserved_at, available_at)
failed_jobs	Permanently failed jobs (queue, payload, exception, failed_at)

---

24. 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

Queue

forgeapi queue:work               # start infinite worker loop
forgeapi queue:run                # process all pending jobs once (use from cron)
forgeapi queue:failed             # list failed jobs
forgeapi queue:retry <id>         # move a failed job back to the queue
forgeapi queue:flush              # delete all failed jobs

Inspection

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

---

25. 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("COOKIE_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": "cookie", "secret": env("COOKIE_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/broadcast.py — activates BroadcastProvider (see §7)
config = {
    "enabled": True,
    "driver": "redis",
    "url": env("REDIS_URL", "redis://localhost:6379"),
    "namespace": "forge",
    "mode": "pubsub",       # "pubsub" | "stream"
    # "maxlen": None,       # stream mode: keep last N messages per key
    # "group": "backend",   # stream mode: consumer group name
    # "consumer": "worker-1",
}

# config/scheduler.py — activates SchedulerProvider (see §22)
config = {
    "enabled": True,
    "schedule": "schedule.py",   # path to your schedule.py relative to project root
}

# config/queue.py — activates QueueProvider (see §23)
config = {
    "enabled": True,
    "queue": "default",   # queue name the in-process worker will consume
}

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.

---

26. 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.

---

27. 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:

# Interactive prompt — choose scope on the fly
forgeapi mcp:install

# Per-project — adds to .mcp.json in the current directory
forgeapi mcp:install --project

# Global (user scope) — available in all projects
forgeapi mcp:install --global

# Check current status
forgeapi mcp:status

Or use the claude CLI directly:

# Per-project
claude mcp add forge-kits forgeapi-mcp

# Global (user scope)
claude mcp add --scope user forge-kits forgeapi-mcp

---

Global setup for Claude Code (manual)

Install once and use across all your projects. Create or edit ~/.claude/.mcp.json:

macOS / Linux — ~/.claude/.mcp.json:

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

Windows — C:\Users\<your-username>\.claude\.mcp.json:

{
  "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"
    }
  }
}

Tip: forgeapi mcp:install --project generates .mcp.json automatically with the correct path for your current venv.

---

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)	List project files and structure

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 (cookie/telegram), session ops
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	config/ directory format, all known sections
models	ModelMixin, @scope, Tortoise field types
cache	Cache facade, drivers, remember/pull/increment
storage	Storage facade, LocalDriver, S3Driver, ImageProcessor
scheduler	Scheduler, timing methods, lifespan integration
queue	Job, dispatch, Queue worker, CLI commands
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
broadcasting_pubsub	BroadcastManager pub/sub mode
broadcasting_stream	BroadcastManager stream mode (persistent)
rbac	Full RBAC — model, seeder, protected routes
pagination	QuerySet .paginate() with filters
guard	API key, active-user guards
cache	remember, pull, counters in a controller
scheduler	lifespan task worker with daily/weekly jobs
queue	Job definition, dispatch, worker, CLI

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.