1. 基础用法 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from fastapi import Depends, FastAPI, HTTPExceptionfrom fastapi.security import OAuth2PasswordBearerapp = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token" ) async def get_current_user (token: str = Depends(oauth2_scheme ) ): user = fake_decode_token(token) if not user: raise HTTPException(status_code=401 , detail="无效令牌" ) return user @app.get("/users/me" ) async def read_users_me (current_user: User = Depends(get_current_user ) ): return current_user
通过OAuth2PasswordBearer声明OAuth2密码流认证方案,自动从Authorization头提取Bearer Token
get_current_user依赖项实现JWT令牌解码和用户信息提取
路径操作函数通过Depends()注入已验证的用户对象
流程图:
graph TD
A[请求到达] --> B{Authorization头存在?}
B -->|是| C[提取Bearer Token]
B -->|否| D[返回401错误]
C --> E[JWT解码验证]
E -->|成功| F[获取用户数据库记录]
E -->|失败| G[返回401错误]
F -->|用户存在| H[返回用户对象]
F -->|用户不存在| G
2. 带类型的路径参数 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from pydantic import BaseModelclass User (BaseModel ): username: str email: str | None = None disabled: bool | None = None async def get_current_active_user (current_user: User = Depends(get_current_user ) ): if current_user.disabled: raise HTTPException(status_code=400 , detail="未激活用户" ) return current_user @app.get("/active-users/me" ) async def read_active_users_me (user: User = Depends(get_current_active_user ) ): return user
添加User模型类型注解实现:
请求解析 :自动转换JWT负载为用户对象
数据验证 :检查用户激活状态
错误处理 :自动生成400错误响应
支持多层依赖注入,get_current_active_user依赖get_current_user
3. 生产环境注意事项
密钥管理 :使用openssl rand -hex 32生成高强度JWT密钥
令牌时效 :设置合理的ACCESS_TOKEN_EXPIRE_MINUTES过期时间(建议30分钟)
密码哈希 :采用passlib库的Bcrypt算法处理用户密码
错误响应 :统一错误格式{"detail": "错误描述"}
日志审计 :记录用户认证失败事件
测试命令示例:
1 2 3 4 5 curl -X POST -d "username=admin&password=secret" http://localhost:8000/token curl -H "Authorization: Bearer YOUR_TOKEN" http://localhost:8000/users/me