VB.net 2010 视频教程 VB.net 2010 视频教程 python基础视频教程
SQL Server 2008 视频教程 c#入门经典教程 Visual Basic从门到精通视频教程
当前位置:
首页 > Python基础教程 >
  • FastAPI路由与请求处理全解:手把手打造用户管理系统

以下为基于FastAPI构建用户管理系统的技术全解,包含路由设计、请求处理及进阶优化方案:


一、环境搭建与初始化

安装核心库 
pip install fastapi[all] sqlalchemy databases[postgresql] passlib python-jose 

项目结构

├── app 
│   ├── models        # 数据模型层 
│   ├── routes        # 路由控制器 
│   ├── services      # 业务逻辑 
│   ├── dependencies  # 依赖注入 
│   └── database.py   # 数据库连接 

二、数据库模型设计(SQLAlchemy示例)

models/user.py 
from sqlalchemy import Column, Integer, String, Boolean 
from app.database import Base 
 
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True)
    email = Column(String(120), unique=True)
    hashed_password = Column(String(300))
    is_active = Column(Boolean, default=True)

三、Pydantic模型设计

schemas/user.py 
from pydantic import BaseModel, EmailStr 
 
class UserCreate(BaseModel):
    username: str 
    email: EmailStr 
    password: str 
 
class UserResponse(BaseModel):
    id: int 
    username: str 
    email: str 
    is_active: bool 
 
    class Config:
        orm_mode = True  # 支持ORM对象转换 

四、核心路由设计

routes/users.py 
from fastapi import APIRouter, Depends, HTTPException 
from sqlalchemy.orm import Session 
 
router = APIRouter(prefix="/users", tags=["用户管理"])
 
@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # 检查用户名重复 
    if db.query(User).filter(User.username == user.username).first():
        raise HTTPException(status_code=409, detail="用户名已存在")
    
    # 密码哈希处理 
    hashed_password = get_password_hash(user.password)
    
    # 创建用户对象 
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password 
    )
    
    # 提交事务 
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user 

五、请求处理深度解析

  1. 参数类型处理
  • 路径参数:@router.get("/{user_id}")
  • 查询参数:def get_users(page: int = 1, limit: int = 10)
  • 请求体:user: UserCreate
  • 表单数据:from fastapi import Form
  1. 异常处理机制
自定义异常处理器 
from fastapi import FastAPI 
from fastapi.exceptions import RequestValidationError 
 
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body}
    )

六、进阶优化方案

  1. 分页查询实现
分页响应模型 
class PaginatedResponse(BaseModel):
    total: int 
    items: List[UserResponse]
 
@router.get("/", response_model=PaginatedResponse)
async def list_users(
    db: Session = Depends(get_db),
    page: int = 1,
    limit: int = 10 
):
    query = db.query(User)
    total = query.count()
    items = query.offset((page-1)*limit).limit(limit).all()
    return {"total": total, "items": items}
  1. 安全认证集成
JWT认证示例 
from fastapi.security import OAuth2PasswordBearer 
 
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=401,
        detail="无效凭证"
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception 
    except JWTError:
        raise credentials_exception 
    
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise credentials_exception 
    return user 

七、性能优化策略

  1. 异步数据库:使用databases库实现异步查询
  2. 缓存机制:集成Redis缓存高频请求
  3. 请求验证:通过Pydantic模型前置过滤无效请求
  4. 依赖复用:将数据库会话等公共依赖进行模块化封装

八、测试方案

使用TestClient进行接口测试 
from fastapi.testclient import TestClient 
 
def test_create_user():
    with TestClient(app) as client:
        response = client.post("/users/", json={
            "username": "testuser",
            "email": "test@example.com",
            "password": "secret"
        })
        assert response.status_code == 201 
        assert "id" in response.json()

总结建议

  1. 采用领域驱动设计划分业务模块
  2. 使用APIRouter实现模块化路由管理
  3. 通过依赖注入解耦业务逻辑
  4. 结合Swagger UI实现API文档自动化
  5. 部署时建议配合UVicorn和Gunicorn实现生产级服务

(完整代码库可参考:https://github.com/example/fastapi-user-system)

最后,如果你对python语言还有任何疑问或者需要进一步的帮助,请访问https://www.xin3721.com 本站原创,转载请注明出处:https://www.xin3721.com


相关教程