当前位置:
首页 > 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
五、请求处理深度解析
- 参数类型处理
-
路径参数:
@router.get("/{user_id}")
-
查询参数:
def get_users(page: int = 1, limit: int = 10)
-
请求体:
user: UserCreate
-
表单数据:
from fastapi import Form
- 异常处理机制
自定义异常处理器
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}
)
六、进阶优化方案
- 分页查询实现
分页响应模型
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}
- 安全认证集成
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
七、性能优化策略
-
异步数据库:使用
databases
库实现异步查询 - 缓存机制:集成Redis缓存高频请求
- 请求验证:通过Pydantic模型前置过滤无效请求
- 依赖复用:将数据库会话等公共依赖进行模块化封装
八、测试方案
使用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()
总结建议
- 采用领域驱动设计划分业务模块
- 使用APIRouter实现模块化路由管理
- 通过依赖注入解耦业务逻辑
- 结合Swagger UI实现API文档自动化
- 部署时建议配合UVicorn和Gunicorn实现生产级服务
(完整代码库可参考:https://github.com/example/fastapi-user-system)
最后,如果你对python语言还有任何疑问或者需要进一步的帮助,请访问https://www.xin3721.com 本站原创,转载请注明出处:https://www.xin3721.com
栏目列表
最新更新
求1000阶乘的结果末尾有多少个0
详解MyBatis延迟加载是如何实现的
IDEA 控制台中文乱码4种解决方案
SpringBoot中版本兼容性处理的实现示例
Spring的IOC解决程序耦合的实现
详解Spring多数据源如何切换
Java报错:UnsupportedOperationException in Col
使用Spring Batch实现批处理任务的详细教程
java中怎么将多个音频文件拼接合成一个
SpringBoot整合ES多个精确值查询 terms功能实
SQL Server 中的数据类型隐式转换问题
SQL Server中T-SQL 数据类型转换详解
sqlserver 数据类型转换小实验
SQL Server数据类型转换方法
SQL Server 2017无法连接到服务器的问题解决
SQLServer地址搜索性能优化
Sql Server查询性能优化之不可小觑的书签查
SQL Server数据库的高性能优化经验总结
SQL SERVER性能优化综述(很好的总结,不要错
开启SQLSERVER数据库缓存依赖优化网站性能
uniapp/H5 获取手机桌面壁纸 (静态壁纸)
[前端] DNS解析与优化
为什么在js中需要添加addEventListener()?
JS模块化系统
js通过Object.defineProperty() 定义和控制对象
这是目前我见过最好的跨域解决方案!
减少回流与重绘
减少回流与重绘
如何使用KrpanoToolJS在浏览器切图
performance.now() 与 Date.now() 对比