通义千问2.5-7B自动化测试生成：CI/CD集成部署案例

FastAPI 实现用户资源CRUD的完整指南结合前文RESTful API设计规范FastAPI实现用户资源CRUD需遵循“资源为中心、HTTP方法语义化、统一接口”原则同时利用FastAPI原生特性依赖注入、数据校验、自动文档提升开发效率。

以下是分步实现方案包含核心代码与逻辑说明。

前置准备

环境依赖安装需安装核心依赖兼顾数据校验、密码安全、服务运行等需求pipinstallfastapi uvicorn pydantic python-jose[cryptography]passlib[bcrypt]依赖说明fastapi核心Web框架支持RESTful、依赖注入、自动文档uvicornASGI服务器用于运行FastAPI应用pydantic数据校验与模型定义确保请求/响应格式合规python-joseJWT认证实现适配RESTful无状态特性passlib密码哈希存储避免明文泄露。

项目基础结构极简结构适用于单体示例复杂项目可拆分路由、模型、依赖等模块project/ ├── main.py # 核心入口模型、接口、依赖、服务启动 └── requirements.txt # 依赖清单

核心实现步骤

初始化应用与配置创建FastAPI实例配置跨域、认证密钥等基础参数贴合RESTful跨平台、安全需求fromfastapiimportFastAPI,Depends,HTTPExceptionfromfastapi.middleware.corsimportCORSMiddlewarefrompydanticimportBaseModelfrompasslib.contextimportCryptContextfromjoseimportJWTError,jwtfromdatetimeimportdatetime,timedeltafromtypingimportOptional,List# 应用初始化appFastAPI(title用户资源CRUD API,description基于FastAPI实现的RESTful用户CRUD接口,version

1.

0.

# 跨域配置支持前端跨域访问生产环境指定具体域名app.add_middleware(CORSMiddleware,allow_origins[*],allow_credentialsTrue,allow_methods[*],allow_headers[*],)# 安全配置SECRET_KEYyour-secret-key# 生产环境用openssl rand -hex 32生成ALGORITHMHS256ACCESS_TOKEN_EXPIRE_MINUTES30# 密码哈希上下文pwd_contextCryptContext(schemes[bcrypt],deprecatedauto)# 模拟数据库生产环境替换为PostgreSQL/MySQLSQLAlchemyfake_users_db{}# 结构{username: {id, username, email, hashed_password, ...}}

数据模型定义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_modeTrue# 支持从字典/ORM对象转换# JWT令牌模型classToken(BaseModel):access_token:strtoken_type:str

核心工具函数认证、密码处理适配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]:用户认证校验用户名密码userget_user(db,username)ifnotuserornotverify_password(password,user[hashed_password]):returnNonereturnuserdefcreate_access_token(data:dict,expires_delta:Optional[timedelta]None)-str:生成JWT访问令牌to_encodedata.copy()expiredatetime.utcnow()(expires_deltaortimedelta(minutes

)to_encode.update({exp:expire})returnjwt.encode(to_encode,SECRET_KEY,algorithmALGORITHM)

依赖项权限控制、认证校验利用FastAPI依赖注入实现接口权限管控贴合RESTful“统一接口”中的认证需求fromfastapi.securityimportOAuth2PasswordBearer oauth2_schemeOAuth2PasswordBearer(tokenUrltoken)# 指定令牌获取接口asyncdefget_current_user(token:strDepends(oauth2_scheme))-UserResponse:获取当前登录用户依赖项所有需认证接口复用credentials_exceptionHTTPException(status_code401,# 符合RESTful 401未认证状态码detail无法验证凭据,headers{WWW-Authenticate:Bearer},)try:payloadjwt.decode(token,SECRET_KEY,algorithms[ALGORITHM])username:strpayload.get(sub)ifusernameisNone:raisecredentials_exceptionexceptJWTError:raisecredentials_exception userget_user(fake_users_db,username)ifuserisNone:raisecredentials_exceptionreturnUserResponse(**user)asyncdefget_current_active_user(current_user:UserResponseDepends(get_current_user))-UserResponse:获取当前活跃用户排除禁用账号ifcurrent_user.disabled:raiseHTTPException(status_code400,detail用户已被禁用)# 400客户端错误returncurrent_user

实现用户CRUD接口RESTful规范严格遵循RESTful语义用HTTP方法表示操作URI标识资源正确使用状态码。

1获取令牌认证接口fromfastapi.securityimportOAuth2PasswordRequestFormapp.post(/token,response_modelToken,tags[认证])asyncdeflogin_for_access_token(form_data:OAuth2PasswordRequestFormDepends()):用户登录获取JWT令牌POST非幂等符合创建令牌语义userauthenticate_user(fake_users_db,form_data.username,form_data.password)ifnotuser:raiseHTTPException(status_code401,detail用户名或密码错误,headers{WWW-Authenticate:Bearer},)access_token_expirestimedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES)access_tokencreate_access_token(data{sub:user[username]},expires_deltaaccess_token_expires)return{access_token:access_token,token_type:bearer}2查询用户GETapp.get(/users/me/,response_modelUserResponse,tags[用户管理])asyncdefread_users_me(current_user:UserResponseDepends(get_current_active_user)):获取当前登录用户信息需认证GET幂等、可缓存returncurrent_userapp.get(/users/,response_modelList[UserResponse],tags[用户管理])asyncdefread_users(skip:int0,limit:int10,current_user:UserResponseDepends(get_current_active_user)):获取用户列表支持分页GET幂等user_listlist(fake_users_db.values())return[UserResponse(**user)foruserinuser_list[skip:skiplimit]]app.get(/users/{username},response_modelUserResponse,tags[用户管理])asyncdefread_user(username:str,current_user:UserResponseDepends(get_current_active_user)):获取指定用户名的用户URI标识单个资源404资源不存在userget_user(fake_users_db,username)ifnotuser:raiseHTTPException(status_code404,detail用户不存在)# 符合RESTful 404状态码returnUserResponse(**user)3创建用户POSTapp.post(/users/,response_modelUserResponse,status_code201,tags[用户管理])asyncdefcreate_user(user:UserCreate):创建新用户POST非幂等201创建成功# 校验用户名是否已存在409冲突ifget_user(fake_users_db,user.username):raiseHTTPException(status_code409,detail用户名已存在)# RESTful 409冲突状态码# 生成用户数据ID自增密码哈希存储user_idlen(fake_users_db)1hashed_passwordget_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/PATCHapp.put(/users/{username},response_modelUserResponse,tags[用户管理])asyncdefupdate_user(username:str,user_update:UserUpdate,current_user:UserResponseDepends(get_current_active_user)):全量更新用户PUT幂等需提供完整字段userget_user(fake_users_db,username)ifnotuser:raiseHTTPException(status_code404,detail用户不存在)# 仅允许更新自己或管理员更新他人简化权限ifcurrent_user.username!username:raiseHTTPException(status_code403,detail无权限更新该用户)# 403权限不足# 全量更新缺省字段置为默认值updated_useruser.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_modelUserResponse,tags[用户管理])asyncdefpatch_user(username:str,user_update:UserUpdate,current_user:UserResponseDepends(get_current_active_user)):增量更新用户PATCH非幂等仅传需修改字段userget_user(fake_users_db,username)ifnotuser:raiseHTTPException(status_code404,detail用户不存在)ifcurrent_user.username!username:raiseHTTPException(status_code403,detail无权限更新该用户)# 增量更新仅修改非None字段updated_useruser.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删除用户DELETEapp.delete(/users/{username},status_code204,tags[用户管理])asyncdefdelete_user(username:str,current_user:UserResponseDepends(get_current_active_user)):删除用户DELETE幂等204无响应体userget_user(fake_users_db,username)ifnotuser:raiseHTTPException(status_code404,detail用户不存在)ifcurrent_user.username!username:raiseHTTPException(status_code403,detail无权限删除该用户)delfake_users_db[username]returnNone# 204状态码无需响应体

服务启动入口if__name____main__:importuvicorn# 启动服务host

0.

0.

0允许外部访问reloadTrue开发环境热重载uvicorn.run(main:app,host

0.

0.

0,port8000,reloadTrue)

接口测试与文档

自动文档访问FastAPI原生生成符合OpenAPI规范的文档无需额外配置Swagger UI推荐支持在线调试http://localhost:8000/docsReDoc简洁文档http://localhost:8000/redoc

测试流程启动服务后访问Swagger UI先通过/token接口获取令牌点击页面顶部“Authorize”输入Bearer {token}完成认证依次测试/users/创建/查询、/users/{username}查询/更新/删除接口验证状态码与响应格式。

生产环境优化替换模拟数据库用SQLAlchemy/Prisma连接PostgreSQL/MySQL实现数据持久化增强权限控制实现RBAC模型区分管理员/普通用户权限如管理员可查询所有用户接口限流集成slowapi对/token、/users/等接口限流返回429状态码安全加固启用HTTPS定期轮换SECRET_KEY增加密码复杂度校验日志与监控添加接口调用日志集成Prometheus/Grafana监控接口性能。

关键RESTful规范贴合点URI设计用复数/users表示资源集合/users/{username}表示单个资源无动词HTTP方法GET查、POST增、PUT/PATCH改、DELETE删语义精准状态码200成功、201创建、204删除成功、401未认证、403无权限、404资源不存在、409冲突符合RESTful语义无状态基于JWT认证服务器不存储会话支持水平扩展自描述消息通过Pydantic模型定义请求/响应格式HTTP头传递认证信息。

91网站-91网站应用