FastAPI 实现用户资源CRUD的完整指南

FastAPI 实现用户资源CRUD的完整指南

结合前文RESTful API设计规范，FastAPI实现用户资源CRUD需遵循“资源为中心、HTTP方法语义化、统一接口”原则，同时利用FastAPI原生特性（依赖注入、数据校验、自动文档）提升开发效率。以下是分步实现方案，包含核心代码与逻辑说明。

一、前置准备

1. 环境依赖安装

需安装核心依赖，兼顾数据校验、密码安全、服务运行等需求：

pipinstallfastapi uvicorn pydantic python-jose[cryptography]passlib[bcrypt]

依赖说明：

• fastapi：核心Web框架，支持RESTful、依赖注入、自动文档；

• uvicorn：ASGI服务器，用于运行FastAPI应用；

• pydantic：数据校验与模型定义，确保请求/响应格式合规；

• python-jose：JWT认证实现，适配RESTful无状态特性；

• passlib：密码哈希存储，避免明文泄露。

2. 项目基础结构

极简结构适用于单体示例，复杂项目可拆分路由、模型、依赖等模块：

project/ ├── main.py # 核心入口（模型、接口、依赖、服务启动） └── requirements.txt # 依赖清单

二、核心实现步骤

1. 初始化应用与配置

创建FastAPI实例，配置跨域、认证密钥等基础参数，贴合RESTful跨平台、安全需求：

fromfastapiimportFastAPI,Depends,HTTPExceptionfromfastapi.middleware.corsimportCORSMiddlewarefrompydanticimportBaseModelfrompasslib.contextimportCryptContextfromjoseimportJWTError,jwtfromdatetimeimportdatetime,timedeltafromtypingimportOptional,List# 应用初始化app=FastAPI(title="用户资源CRUD API",description="基于FastAPI实现的RESTful用户CRUD接口",version="1.0.0")# 跨域配置（支持前端跨域访问，生产环境指定具体域名）app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)# 安全配置SECRET_KEY="your-secret-key"# 生产环境用openssl rand -hex 32生成ALGORITHM="HS256"ACCESS_TOKEN_EXPIRE_MINUTES=30# 密码哈希上下文pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto")# 模拟数据库（生产环境替换为PostgreSQL/MySQL+SQLAlchemy）fake_users_db={}# 结构：{username: {id, username, email, hashed_password, ...}}

2. 数据模型定义（Pydantic）

遵循RESTful“自描述消息”原则，用Pydantic定义请求/响应模型，实现自动数据校验：

classUserBase(BaseModel):"""基础用户模型，共享字段"""username:str# 用户名（唯一标识资源的核心字段）email:str# 邮箱full_name:Optional[str]=None# 可选全名disabled:Optional[bool]=False# 账号状态classUserCreate(UserBase):"""创建用户请求模型（新增密码字段）"""password:str# 密码（仅创建时传递，响应中隐藏）classUserUpdate(BaseModel):"""更新用户请求模型（支持增量更新）"""email:Optional[str]=Nonefull_name:Optional[str]=Nonedisabled:Optional[bool]=NoneclassUserResponse(UserBase):"""用户响应模型（包含ID，隐藏密码）"""id:int# 资源唯一IDclassConfig:orm_mode=True# 支持从字典/ORM对象转换# JWT令牌模型classToken(BaseModel):access_token:strtoken_type:str

3. 核心工具函数（认证、密码处理）

适配RESTful无状态特性，实现密码哈希、JWT生成/校验、用户查询等通用逻辑：

defverify_password(plain_password:str,hashed_password:str)->bool:"""验证密码正确性"""returnpwd_context.verify(plain_password,hashed_password)defget_password_hash(password:str)->str:"""生成密码哈希"""returnpwd_context.hash(password)defget_user(db:dict,username:str)->Optional[dict]:"""从数据库查询用户"""returndb.get(username)defauthenticate_user(db:dict,username:str,password:str)->Optional[dict]:"""用户认证（校验用户名+密码）"""user=get_user(db,username)ifnotuserornotverify_password(password,user["hashed_password"]):returnNonereturnuserdefcreate_access_token(data:dict,expires_delta:Optional[timedelta]=None)->str:"""生成JWT访问令牌"""to_encode=data.copy()expire=datetime.utcnow()+(expires_deltaortimedelta(minutes=15))to_encode.update({"exp":expire})returnjwt.encode(to_encode,SECRET_KEY,algorithm=ALGORITHM)

4. 依赖项（权限控制、认证校验）

利用FastAPI依赖注入，实现接口权限管控，贴合RESTful“统一接口”中的认证需求：

fromfastapi.securityimportOAuth2PasswordBearer oauth2_scheme=OAuth2PasswordBearer(tokenUrl="token")# 指定令牌获取接口asyncdefget_current_user(token:str=Depends(oauth2_scheme))->UserResponse:"""获取当前登录用户（依赖项，所有需认证接口复用）"""credentials_exception=HTTPException(status_code=401,# 符合RESTful 401未认证状态码detail="无法验证凭据",headers={"WWW-Authenticate":"Bearer"},)try:payload=jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM])username:str=payload.get("sub")ifusernameisNone:raisecredentials_exceptionexceptJWTError:raisecredentials_exception user=get_user(fake_users_db,username)ifuserisNone:raisecredentials_exceptionreturnUserResponse(**user)asyncdefget_current_active_user(current_user:UserResponse=Depends(get_current_user))->UserResponse:"""获取当前活跃用户（排除禁用账号）"""ifcurrent_user.disabled:raiseHTTPException(status_code=400,detail="用户已被禁用")# 400客户端错误returncurrent_user

5. 实现用户CRUD接口（RESTful规范）

严格遵循RESTful语义：用HTTP方法表示操作，URI标识资源，正确使用状态码。

（1）获取令牌（认证接口）

fromfastapi.securityimportOAuth2PasswordRequestForm@app.post("/token",response_model=Token,tags=["认证"])asyncdeflogin_for_access_token(form_data:OAuth2PasswordRequestForm=Depends()):"""用户登录获取JWT令牌（POST非幂等，符合创建令牌语义）"""user=authenticate_user(fake_users_db,form_data.username,form_data.password)ifnotuser:raiseHTTPException(status_code=401,detail="用户名或密码错误",headers={"WWW-Authenticate":"Bearer"},)access_token_expires=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)access_token=create_access_token(data={"sub":user["username"]},expires_delta=access_token_expires)return{"access_token":access_token,"token_type":"bearer"}

（2）查询用户（GET）

@app.get("/users/me/",response_model=UserResponse,tags=["用户管理"])asyncdefread_users_me(current_user:UserResponse=Depends(get_current_active_user)):"""获取当前登录用户信息（需认证，GET幂等、可缓存）"""returncurrent_user@app.get("/users/",response_model=List[UserResponse],tags=["用户管理"])asyncdefread_users(skip:int=0,limit:int=10,current_user:UserResponse=Depends(get_current_active_user)):"""获取用户列表（支持分页，GET幂等）"""user_list=list(fake_users_db.values())return[UserResponse(**user)foruserinuser_list[skip:skip+limit]]@app.get("/users/{username}",response_model=UserResponse,tags=["用户管理"])asyncdefread_user(username:str,current_user:UserResponse=Depends(get_current_active_user)):"""获取指定用户名的用户（URI标识单个资源，404资源不存在）"""user=get_user(fake_users_db,username)ifnotuser:raiseHTTPException(status_code=404,detail="用户不存在")# 符合RESTful 404状态码returnUserResponse(**user)

（3）创建用户（POST）

@app.post("/users/",response_model=UserResponse,status_code=201,tags=["用户管理"])asyncdefcreate_user(user:UserCreate):"""创建新用户（POST非幂等，201创建成功）"""# 校验用户名是否已存在（409冲突）ifget_user(fake_users_db,user.username):raiseHTTPException(status_code=409,detail="用户名已存在")# RESTful 409冲突状态码# 生成用户数据（ID自增，密码哈希存储）user_id=len(fake_users_db)+1hashed_password=get_password_hash(user.password)new_user={"id":user_id,"username":user.username,"email":user.email,"full_name":user.full_name,"disabled":user.disabled,"hashed_password":hashed_password}fake_users_db[user.username]=new_userreturnUserResponse(**new_user)

（4）更新用户（PUT/PATCH）

@app.put("/users/{username}",response_model=UserResponse,tags=["用户管理"])asyncdefupdate_user(username:str,user_update:UserUpdate,current_user:UserResponse=Depends(get_current_active_user)):"""全量更新用户（PUT幂等，需提供完整字段）"""user=get_user(fake_users_db,username)ifnotuser:raiseHTTPException(status_code=404,detail="用户不存在")# 仅允许更新自己或管理员更新他人（简化权限）ifcurrent_user.username!=username:raiseHTTPException(status_code=403,detail="无权限更新该用户")# 403权限不足# 全量更新（缺省字段置为默认值）updated_user=user.copy()updated_user["email"]=user_update.emailoruser["email"]updated_user["full_name"]=user_update.full_nameoruser["full_name"]updated_user["disabled"]=user_update.disabledifuser_update.disabledisnotNoneelseuser["disabled"]fake_users_db[username]=updated_userreturnUserResponse(**updated_user)@app.patch("/users/{username}",response_model=UserResponse,tags=["用户管理"])asyncdefpatch_user(username:str,user_update:UserUpdate,current_user:UserResponse=Depends(get_current_active_user)):"""增量更新用户（PATCH非幂等，仅传需修改字段）"""user=get_user(fake_users_db,username)ifnotuser:raiseHTTPException(status_code=404,detail="用户不存在")ifcurrent_user.username!=username:raiseHTTPException(status_code=403,detail="无权限更新该用户")# 增量更新（仅修改非None字段）updated_user=user.copy()ifuser_update.email:updated_user["email"]=user_update.emailifuser_update.full_nameisnotNone:updated_user["full_name"]=user_update.full_nameifuser_update.disabledisnotNone:updated_user["disabled"]=user_update.disabled fake_users_db[username]=updated_userreturnUserResponse(**updated_user)

（5）删除用户（DELETE）

@app.delete("/users/{username}",status_code=204,tags=["用户管理"])asyncdefdelete_user(username:str,current_user:UserResponse=Depends(get_current_active_user)):"""删除用户（DELETE幂等，204无响应体）"""user=get_user(fake_users_db,username)ifnotuser:raiseHTTPException(status_code=404,detail="用户不存在")ifcurrent_user.username!=username:raiseHTTPException(status_code=403,detail="无权限删除该用户")delfake_users_db[username]returnNone# 204状态码无需响应体

6. 服务启动入口

if__name__=="__main__":importuvicorn# 启动服务：host=0.0.0.0允许外部访问，reload=True开发环境热重载uvicorn.run("main:app",host="0.0.0.0",port=8000,reload=True)

三、接口测试与文档

1. 自动文档访问

FastAPI原生生成符合OpenAPI规范的文档，无需额外配置：

• Swagger UI（推荐，支持在线调试）：http://localhost:8000/docs

• ReDoc（简洁文档）：http://localhost:8000/redoc

2. 测试流程

1. 启动服务后，访问Swagger UI，先通过/token接口获取令牌；

2. 点击页面顶部“Authorize”，输入Bearer {token}完成认证；

3. 依次测试/users/（创建/查询）、/users/{username}（查询/更新/删除）接口，验证状态码与响应格式。

四、生产环境优化

1. 替换模拟数据库：用SQLAlchemy/Prisma连接PostgreSQL/MySQL，实现数据持久化；

2. 增强权限控制：实现RBAC模型，区分管理员/普通用户权限（如管理员可查询所有用户）；

3. 接口限流：集成slowapi，对/token、/users/等接口限流，返回429状态码；

4. 安全加固：启用HTTPS，定期轮换SECRET_KEY，增加密码复杂度校验；

5. 日志与监控：添加接口调用日志，集成Prometheus/Grafana监控接口性能。

五、关键RESTful规范贴合点

• URI设计：用复数/users表示资源集合，/users/{username}表示单个资源，无动词；

• HTTP方法：GET（查）、POST（增）、PUT/PATCH（改）、DELETE（删），语义精准；

• 状态码：200（成功）、201（创建）、204（删除成功）、401（未认证）、403（无权限）、404（资源不存在）、409（冲突），符合RESTful语义；

• 无状态：基于JWT认证，服务器不存储会话，支持水平扩展；

• 自描述消息：通过Pydantic模型定义请求/响应格式，HTTP头传递认证信息。