0%

FastAPI自学教程(37) - 获取当前用户信息

1. 基础用法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer

app = 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 BaseModel

class 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. 生产环境注意事项

  1. 密钥管理:使用openssl rand -hex 32生成高强度JWT密钥
  2. 令牌时效:设置合理的ACCESS_TOKEN_EXPIRE_MINUTES过期时间(建议30分钟)
  3. 密码哈希:采用passlib库的Bcrypt算法处理用户密码
  4. 错误响应:统一错误格式{"detail": "错误描述"}
  5. 日志审计:记录用户认证失败事件

测试命令示例:

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