Master FastAPI Production: Routing, DB Integration, Validation & Error Handling

After a basic introduction to FastAPI, this article focuses on practical aspects needed for a production environment, such as database integration, routing (blueprints), data validation, and error handling.

Blueprint

FastAPI does not have a built‑in Blueprint concept; instead, include_router is used to add routes, providing similar functionality to Flask blueprints.

import time
from typing import List
from starlette.templating import Jinja2Templates
from fastapi import Depends, FastAPI, HTTPException
from starlette.staticfiles import StaticFiles
from app import models
from app.database.database import SessionLocal, engine
from app.home import user, index

app = FastAPI()

app.mount("/static", StaticFiles(directory="app/static"), name="static")  # mount static files
templates = Jinja2Templates(directory="templates")  # template directory

app.include_router(index.userRouter)
app.include_router(user.userRouter, prefix="/user")

In the home directory, user.py and index.py define an APIRouter() instance (or a subclass) and use the prefix argument to specify sub‑route paths.

Supporting Multiple Request Methods

By inspecting request.method you can handle different HTTP verbs, and APIRouter.add_api_route allows specifying a list of methods for a single endpoint.

# userRouter example
@userRouter.post("/login/", response_model=schemas.UserOut)
async def login(*, request: Request, db: Session = Depends(get_db), username: str = Form(None), password: str = Form(None)):
    if request.method == "POST":
        db_user = db.query(models.User).filter(User.username == username).first()
        if not db_user:
            raise HTTPException(status_code=400, detail="用户不存在")
        print("验证通过 ！！！")
        return RedirectResponse('/index')
    return templates.TemplateResponse("user/login.html", {"request": request})

# add routes that accept both GET and POST
userRouter.add_api_route(methods=['GET', 'POST'], path="/login", endpoint=login)
userRouter.add_api_route(methods=['GET', 'POST'], path="/list", endpoint=userList)

Database

FastAPI commonly uses SQLAlchemy for ORM operations. The following snippet shows engine creation, sessionmaker setup, and a helper to obtain a database session.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:[email protected]:3306/blog"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Model definitions for User and Article illustrate typical fields and relationships.

class User(UserMixin, Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    email = Column(String(64))
    username = Column(String(64))
    role = Column(String(64))
    password_hash = Column(String(128))
    head_img = Column(String(128))
    create_time = Column(DateTime, default=datetime.now)

    def __repr__(self):
        return f'<User %r>' % self.username

class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True)
    title = Column(String(32))
    author = Column(String(32))
    img_url = Column(Text, nullable=False)
    content = Column(Text, nullable=False)
    tag = Column(String(64), nullable=True)
    uuid = Column(Text, default=uuid.uuid4())
    desc = Column(String(100), nullable=False)
    create_time = Column(DateTime, default=datetime.now)
    articleDetail = relationship('Article_Detail', backref='article')

CRUD examples for article details (add, delete, update, query) demonstrate typical database operations.

async def articleDetailAdd(*, request: Request, db: Session = Depends(get_db), d_content: str, uid: int):
    if request.method == "POST":
        addArticleDetail = Article_Detail(d_content=d_content, uid=uid)
        db.add(addArticleDetail)
        db.commit()
        db.refresh(addArticleDetail)
        print("添加成功 ！！！")
        return "添加成功"
    return "缺少参数"

async def articleDetailDel(*, request: Request, db: Session = Depends(get_db), aid: int):
    if request.method == "POST":
        db.query(Article_Detail).filter(Article_Detail.id == aid).delete()
        db.commit()
        print("删除成功 ！！！")
        return "删除成功"
    return "缺少参数"

async def articleDetailUpdate(*, request: Request, db: Session = Depends(get_db), aid: int, d_content: str):
    if request.method == "POST":
        articleInfo = db.query(Article_Detail).filter(Article_Detail.id == aid).first()
        if not articleInfo:
            raise HTTPException(status_code=400, detail="no no no !!")
        articleInfo.d_content = d_content
        db.commit()
        print("提交成功 ！！！")
        return "更新成功"
    return "缺少参数"

Data Validation

FastAPI uses Pydantic models to validate request bodies and control response fields via the response_model parameter.

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str = None

@app.post("/user/", response_model=UserOut)
async def create_user(*, user: UserIn):
    return user

The input model UserIn is used for the request body, while the output model UserOut restricts the fields returned to the client.

Exception Handling

Standard HTTPException can raise errors with specific status codes and messages. Custom exceptions and handlers provide more flexibility.

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

app = FastAPI()

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(status_code=418, content={"message": f"我家热得快炸了..."})

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

Additional handlers for StarletteHTTPException and RequestValidationError demonstrate how to return plain‑text responses with appropriate status codes.

from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return PlainTextResponse(str(exc), status_code=400)

Proper use of exception handling makes the codebase more robust, improves client experience, and simplifies maintenance.

Overall, the article shares practical FastAPI features for production use and encourages further exploration of topics such as security, middleware, and deployment.