Loading...
Loading...
Guides the agent through scaffolding and building FastAPI applications, including project structure, API routes, request/response models, path and query parameters, dependency injection, middleware, error handling, and boilerplate generation. Triggered when the user asks to "scaffold a FastAPI project", "create a FastAPI app", "add an API endpoint", "create a router", "add middleware", "implement dependency injection", "handle errors", "set up CORS", "create background tasks", "implement WebSocket", "structure a FastAPI project", "generate boilerplate", or "add authentication".
npx skill4agent add ingpdw/pdw-python-dev-tool app-scaffolding/docs/redocproject/
├── app/
│ ├── __init__.py
│ ├── main.py # Application factory and lifespan
│ ├── config.py # Settings via pydantic-settings
│ ├── dependencies.py # Shared dependencies
│ ├── models.py # SQLAlchemy / ORM models
│ ├── schemas.py # Pydantic request/response schemas
│ ├── routers/
│ │ ├── __init__.py # Router registration
│ │ ├── users.py
│ │ ├── items.py
│ │ └── health.py
│ ├── services/ # Business logic layer
│ ├── middleware/ # Custom middleware
│ └── exceptions.py # Custom exception handlers
├── tests/
│ ├── conftest.py
│ ├── test_users.py
│ └── test_items.py
├── alembic/ # Database migrations
├── pyproject.toml
└── .envassets/app-template/on_eventfrom contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup: initialize resources
app.state.db_pool = await create_db_pool()
app.state.http_client = httpx.AsyncClient()
yield
# Shutdown: release resources
await app.state.http_client.aclose()
await app.state.db_pool.close()
def create_app() -> FastAPI:
app = FastAPI(
title="My API",
version="0.1.0",
lifespan=lifespan,
)
app.include_router(users_router)
app.include_router(items_router)
return app
app = create_app()@app.get("/users/{user_id}")
async def get_user(user_id: int) -> User:
...@app.get("/items")
async def list_items(skip: int = 0, limit: int = 20, q: str | None = None):
...from pydantic import BaseModel
class ItemCreate(BaseModel):
name: str
price: float
description: str | None = None
@app.post("/items", status_code=201)
async def create_item(item: ItemCreate) -> Item:
...@app.put("/items/{item_id}")
async def update_item(
item_id: int, # path
item: ItemUpdate, # body
q: str | None = None, # query
x_token: Annotated[str, Header()], # header
):
...@app.post("/users", response_model=UserOut, status_code=201)
async def create_user(user: UserCreate):
...
@app.get("/report", response_class=HTMLResponse)
async def get_report():
return "<html>...</html>"response_model_exclude_unset=Trueresponse_model_exclude={"password"}# app/routers/users.py
from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/")
async def list_users():
...
@router.get("/{user_id}")
async def get_user(user_id: int):
...from app.routers import users, items
app.include_router(users.router)
app.include_router(items.router, prefix="/api/v1")router = APIRouter(
prefix="/admin",
tags=["admin"],
dependencies=[Depends(require_admin)],
)Depends()from fastapi import Depends
from typing import Annotated
async def get_db():
async with async_session() as session:
yield session
DbSession = Annotated[AsyncSession, Depends(get_db)]
async def get_current_user(db: DbSession, token: str = Depends(oauth2_scheme)) -> User:
...
CurrentUser = Annotated[User, Depends(get_current_user)]
@app.get("/me")
async def read_me(user: CurrentUser):
return userreferences/dependency-injection.mdclass UserBase(BaseModel):
email: EmailStr
name: str
class UserCreate(UserBase):
password: str
class UserOut(UserBase):
id: int
created_at: datetime
model_config = ConfigDict(from_attributes=True)@app.on_event("startup")@app.on_event("shutdown")yieldapp.state@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = await aioredis.from_url("redis://localhost")
yield
await app.state.redis.close()from fastapi import BackgroundTasks
def send_notification(email: str, message: str):
# blocking or async work
...
@app.post("/orders")
async def create_order(order: OrderCreate, background_tasks: BackgroundTasks):
result = await process_order(order)
background_tasks.add_task(send_notification, order.email, "Order confirmed")
return resultfrom fastapi import HTTPException
@app.get("/items/{item_id}")
async def get_item(item_id: int):
item = await fetch_item(item_id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return itemfrom fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
class ItemNotFoundError(Exception):
def __init__(self, item_id: int):
self.item_id = item_id
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request, exc):
return JSONResponse(
status_code=404,
content={"detail": f"Item {exc.item_id} not found"},
)
@app.exception_handler(RequestValidationError)
async def validation_handler(request, exc):
return JSONResponse(
status_code=422,
content={"detail": exc.errors(), "body": exc.body},
)from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)allow_origins=["*"]allow_credentials=Truereferences/middleware.mdTestClienthttpx.AsyncClientfrom fastapi.testclient import TestClient
def test_read_items():
app.dependency_overrides[get_db] = lambda: mock_db
client = TestClient(app)
response = client.get("/items")
assert response.status_code == 200
app.dependency_overrides.clear()httpximport pytest
from httpx import ASGITransport, AsyncClient
@pytest.mark.anyio
async def test_create_item():
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.post("/items", json={"name": "Widget", "price": 9.99})
assert response.status_code == 201references/routing-patterns.mdreferences/middleware.mdreferences/dependency-injection.mdassets/app-template/