在前面的教程中我们深入探讨了 FastAPI 的一系列高级特性这些特性使得 FastAPI 不仅是一个简单的 API 框架更是一个能够构建复杂、高性能、生产级应用的全能工具。以下是核心高级特性的简要回顾1. 性能优化异步数据库操作使用异步驱动如asyncpg配合 SQLAlchemy 异步扩展实现真正的非阻塞 I/O。多级缓存内存缓存cachetools与分布式缓存Redis相结合降低数据库负载。Gzip 压缩通过GZipMiddleware减少网络传输量。速率限制使用slowapi防止滥用。连接池优化合理配置数据库连接池参数平衡资源与性能。性能监控中间件记录请求耗时定位瓶颈。2. 高级安全实践多因素认证 (MFA)集成 TOTP如 Google Authenticator增强账户安全。API 密钥管理为第三方开发者提供可过期、带权限的 API 密钥。增强的密码哈希使用passlib配置多种哈希算法bcrypt、argon2支持密码升级。CSRF 保护对于非 API 的传统表单添加 CSRF 令牌验证。3. 异步任务处理BackgroundTasks轻量级后台任务适用于简单操作。Celery 集成分布式任务队列支持重试、定时、任务状态跟踪。APScheduler实现定时任务如定期清理过期数据。4. 测试驱动开发单元测试使用pytest和TestClient编写测试覆盖各种场景。依赖项覆盖通过app.dependency_overrides替换真实依赖如数据库。异步测试使用AsyncClient测试异步端点。测试覆盖率pytest-cov测量覆盖率。5. GraphQL 集成Strawberry将 FastAPI 与 GraphQL 结合提供灵活的数据查询。Query 与 Mutation定义 GraphQL 类型和操作与 REST 共存。6. 微服务架构服务发现使用 Consul 注册与发现服务实例。服务间通信httpx.AsyncClient实现异步 HTTP 调用。容错与重试结合tenacity等库增强健壮性。这些高级特性相互配合能够帮助你构建出高可用、可扩展、安全的现代 Web 应用。下面我们将通过一个完整的实战项目——“实时聊天 用户认证”——来演示如何将这些知识付诸实践。完整项目实时聊天 用户认证本项目实现了一个简单的聊天应用包含以下功能用户注册、登录JWT 认证受保护的用户信息接口WebSocket 聊天室支持多用户实时广播消息WebSocket 连接需携带 JWT 令牌进行身份验证项目结构chat_app/ ├── main.py # 应用入口 ├── database.py # 数据库配置 ├── models.py # SQLAlchemy 模型 ├── schemas.py # Pydantic 模型 ├── auth.py # 认证相关密码哈希、JWT ├── dependencies.py # 公共依赖项 ├── routers/ │ ├── auth.py # 认证路由 │ ├── users.py # 用户信息路由 │ └── chat.py # 聊天 WebSocket 路由 ├── requirements.txt # 依赖列表 └── README.md # 项目说明环境要求Python 3.8安装依赖pip install -r requirements.txt依赖清单 (requirements.txt)fastapi uvicorn[standard] sqlalchemy python-jose[cryptography] passlib[bcrypt] python-multipart代码实现1. database.py - 数据库配置fromsqlalchemyimportcreate_enginefromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmaker SQLALCHEMY_DATABASE_URLsqlite:///./chat_app.db# SQLite 需要设置 check_same_threadFalseenginecreate_engine(SQLALCHEMY_DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)Basedeclarative_base()2. models.py - SQLAlchemy 模型fromsqlalchemyimportColumn,Integer,Stringfrom.databaseimportBaseclassUser(Base):__tablename__usersidColumn(Integer,primary_keyTrue,indexTrue)usernameColumn(String,uniqueTrue,indexTrue,nullableFalse)hashed_passwordColumn(String,nullableFalse)3. schemas.py - Pydantic 模型frompydanticimportBaseModelclassUserBase(BaseModel):username:strclassUserCreate(UserBase):password:strclassUser(UserBase):id:intclassConfig:orm_modeTrueclassToken(BaseModel):access_token:strtoken_type:strclassTokenData(BaseModel):username:str|NoneNone4. auth.py - 认证工具fromdatetimeimportdatetime,timedeltafromjoseimportJWTError,jwtfrompasslib.contextimportCryptContext# 密钥应从环境变量读取此处仅作示例SECRET_KEY09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7ALGORITHMHS256ACCESS_TOKEN_EXPIRE_MINUTES30pwd_contextCryptContext(schemes[bcrypt],deprecatedauto)defverify_password(plain_password,hashed_password):returnpwd_context.verify(plain_password,hashed_password)defget_password_hash(password):returnpwd_context.hash(password)defcreate_access_token(data:dict,expires_delta:timedelta|NoneNone):to_encodedata.copy()ifexpires_delta:expiredatetime.utcnow()expires_deltaelse:expiredatetime.utcnow()timedelta(minutes15)to_encode.update({exp:expire})encoded_jwtjwt.encode(to_encode,SECRET_KEY,algorithmALGORITHM)returnencoded_jwtdefdecode_token(token:str):try:payloadjwt.decode(token,SECRET_KEY,algorithms[ALGORITHM])username:strpayload.get(sub)ifusernameisNone:returnNonereturnusernameexceptJWTError:returnNone5. dependencies.py - 公共依赖项fromfastapiimportDepends,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearerfromsqlalchemy.ormimportSessionfromjoseimportJWTErrorfrom.importdatabase,models,auth,schemas oauth2_schemeOAuth2PasswordBearer(tokenUrl/auth/token)defget_db():dbdatabase.SessionLocal()try:yielddbfinally:db.close()defget_current_user(token:strDepends(oauth2_scheme),db:SessionDepends(get_db))-models.User:usernameauth.decode_token(token)ifusernameisNone:raiseHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailInvalid authentication credentials,headers{WWW-Authenticate:Bearer},)userdb.query(models.User).filter(models.User.usernameusername).first()ifuserisNone:raiseHTTPException(status_code404,detailUser not found)returnuser6. routers/auth.py - 认证路由fromfastapiimportAPIRouter,Depends,HTTPException,statusfromfastapi.securityimportOAuth2PasswordRequestFormfromsqlalchemy.ormimportSessionfrom..importschemas,models,auth,dependencies routerAPIRouter(prefix/auth,tags[authentication])router.post(/register,response_modelschemas.User)defregister(user:schemas.UserCreate,db:SessionDepends(dependencies.get_db)):db_userdb.query(models.User).filter(models.User.usernameuser.username).first()ifdb_user:raiseHTTPException(status_code400,detailUsername already registered)hashedauth.get_password_hash(user.password)new_usermodels.User(usernameuser.username,hashed_passwordhashed)db.add(new_user)db.commit()db.refresh(new_user)returnnew_userrouter.post(/token,response_modelschemas.Token)deflogin(form_data:OAuth2PasswordRequestFormDepends(),db:SessionDepends(dependencies.get_db)):userdb.query(models.User).filter(models.User.usernameform_data.username).first()ifnotuserornotauth.verify_password(form_data.password,user.hashed_password):raiseHTTPException(status_code400,detailIncorrect username or password)access_tokenauth.create_access_token(data{sub:user.username})return{access_token:access_token,token_type:bearer}7. routers/users.py - 用户信息路由fromfastapiimportAPIRouter,Dependsfrom..importschemas,dependencies routerAPIRouter(prefix/users,tags[users])router.get(/me,response_modelschemas.User)defread_users_me(current_userDepends(dependencies.get_current_user)):returncurrent_user8. routers/chat.py - 聊天 WebSocket 路由fromfastapiimportAPIRouter,WebSocket,WebSocketDisconnectfromtypingimportDictfrom..authimportdecode_token routerAPIRouter(prefix/chat,tags[chat])classConnectionManager:def__init__(self):self.active_connections:Dict[str,WebSocket]{}# username - websocketasyncdefconnect(self,websocket:WebSocket,username:str):awaitwebsocket.accept()self.active_connections[username]websocketdefdisconnect(self,username:str):self.active_connections.pop(username,None)asyncdefsend_personal(self,message:str,username:str):ifusernameinself.active_connections:awaitself.active_connections[username].send_text(message)asyncdefbroadcast(self,message:str,exclude_username:strNone):forusername,conninself.active_connections.items():ifusername!exclude_username:awaitconn.send_text(message)managerConnectionManager()router.websocket(/ws)asyncdefwebsocket_endpoint(websocket:WebSocket):# 从查询参数获取 tokentokenwebsocket.query_params.get(token)ifnottoken:awaitwebsocket.close(code1008)returnusernamedecode_token(token)ifnotusername:awaitwebsocket.close(code1008)returnawaitmanager.connect(websocket,username)try:whileTrue:dataawaitwebsocket.receive_text()# 广播消息给所有其他用户awaitmanager.broadcast(f{username}:{data},exclude_usernameusername)exceptWebSocketDisconnect:manager.disconnect(username)awaitmanager.broadcast(fUser{username}left the chat)9. main.py - 应用入口fromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewarefrom.databaseimportenginefrom.modelsimportBasefrom.routersimportauth,users,chat# 创建数据库表Base.metadata.create_all(bindengine)appFastAPI(titleChat App with JWT WebSocket)# 配置 CORSapp.add_middleware(CORSMiddleware,allow_origins[*],# 生产环境应限制allow_credentialsTrue,allow_methods[*],allow_headers[*],)# 包含路由app.include_router(auth.router)app.include_router(users.router)app.include_router(chat.router)app.get(/)defroot():return{message:Welcome to Chat App API}10. README.md - 项目说明# Chat App with FastAPI 一个简单的实时聊天应用包含用户认证和 WebSocket 通信。 ## 功能 - 用户注册 - 用户登录获取 JWT 令牌 - 获取当前用户信息 - WebSocket 聊天室需令牌验证 ## 快速开始 1. 克隆仓库 2. 安装依赖pip install -r requirements.txt 3. 运行服务uvicorn chat_app.main:app --reload 4. 访问 API 文档http://localhost:8000/docs 5. 使用 WebSocket 客户端连接 ws://localhost:8000/chat/ws?tokenyour_jwt_token 进行聊天 ## API 端点 - POST /auth/register - 注册新用户 - POST /auth/token - 登录获取令牌 - GET /users/me - 获取当前用户信息需要认证 - WebSocket /chat/ws - 聊天室连接需提供 token 查询参数运行与测试启动服务uvicorn chat_app.main:app--reload注册用户使用/auth/register端点POST JSON{username:alice,password:secret}登录获取令牌POST/auth/token表单数据username: alice password: secret返回类似{access_token:eyJhbGc...,token_type:bearer}WebSocket 连接使用 WebSocket 客户端如浏览器 JavaScript、Postman、wscat连接到ws://localhost:8000/chat/ws?tokeneyJhbGc...连接成功后可以发送消息所有其他在线用户将收到广播。测试受保护路由在请求头中添加Authorization: Bearer token访问/users/me。注意事项生产环境中请勿将密钥硬编码应使用环境变量。数据库使用 SQLite仅适用于开发测试。WebSocket 连接目前不存储聊天记录仅实时转发。通过这个完整项目你可以看到 FastAPI 的高级特性如何在实际中落地JWT 认证保护 HTTP 和 WebSocket 端点SQLAlchemy 处理数据持久化依赖项简化代码复用WebSocket 实现实时通信。希望这个示例能帮助你进一步掌握 FastAPI构建出更强大的应用
Python之FastAPI 高级特性总结与完整项目实战
在前面的教程中我们深入探讨了 FastAPI 的一系列高级特性这些特性使得 FastAPI 不仅是一个简单的 API 框架更是一个能够构建复杂、高性能、生产级应用的全能工具。以下是核心高级特性的简要回顾1. 性能优化异步数据库操作使用异步驱动如asyncpg配合 SQLAlchemy 异步扩展实现真正的非阻塞 I/O。多级缓存内存缓存cachetools与分布式缓存Redis相结合降低数据库负载。Gzip 压缩通过GZipMiddleware减少网络传输量。速率限制使用slowapi防止滥用。连接池优化合理配置数据库连接池参数平衡资源与性能。性能监控中间件记录请求耗时定位瓶颈。2. 高级安全实践多因素认证 (MFA)集成 TOTP如 Google Authenticator增强账户安全。API 密钥管理为第三方开发者提供可过期、带权限的 API 密钥。增强的密码哈希使用passlib配置多种哈希算法bcrypt、argon2支持密码升级。CSRF 保护对于非 API 的传统表单添加 CSRF 令牌验证。3. 异步任务处理BackgroundTasks轻量级后台任务适用于简单操作。Celery 集成分布式任务队列支持重试、定时、任务状态跟踪。APScheduler实现定时任务如定期清理过期数据。4. 测试驱动开发单元测试使用pytest和TestClient编写测试覆盖各种场景。依赖项覆盖通过app.dependency_overrides替换真实依赖如数据库。异步测试使用AsyncClient测试异步端点。测试覆盖率pytest-cov测量覆盖率。5. GraphQL 集成Strawberry将 FastAPI 与 GraphQL 结合提供灵活的数据查询。Query 与 Mutation定义 GraphQL 类型和操作与 REST 共存。6. 微服务架构服务发现使用 Consul 注册与发现服务实例。服务间通信httpx.AsyncClient实现异步 HTTP 调用。容错与重试结合tenacity等库增强健壮性。这些高级特性相互配合能够帮助你构建出高可用、可扩展、安全的现代 Web 应用。下面我们将通过一个完整的实战项目——“实时聊天 用户认证”——来演示如何将这些知识付诸实践。完整项目实时聊天 用户认证本项目实现了一个简单的聊天应用包含以下功能用户注册、登录JWT 认证受保护的用户信息接口WebSocket 聊天室支持多用户实时广播消息WebSocket 连接需携带 JWT 令牌进行身份验证项目结构chat_app/ ├── main.py # 应用入口 ├── database.py # 数据库配置 ├── models.py # SQLAlchemy 模型 ├── schemas.py # Pydantic 模型 ├── auth.py # 认证相关密码哈希、JWT ├── dependencies.py # 公共依赖项 ├── routers/ │ ├── auth.py # 认证路由 │ ├── users.py # 用户信息路由 │ └── chat.py # 聊天 WebSocket 路由 ├── requirements.txt # 依赖列表 └── README.md # 项目说明环境要求Python 3.8安装依赖pip install -r requirements.txt依赖清单 (requirements.txt)fastapi uvicorn[standard] sqlalchemy python-jose[cryptography] passlib[bcrypt] python-multipart代码实现1. database.py - 数据库配置fromsqlalchemyimportcreate_enginefromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmaker SQLALCHEMY_DATABASE_URLsqlite:///./chat_app.db# SQLite 需要设置 check_same_threadFalseenginecreate_engine(SQLALCHEMY_DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)Basedeclarative_base()2. models.py - SQLAlchemy 模型fromsqlalchemyimportColumn,Integer,Stringfrom.databaseimportBaseclassUser(Base):__tablename__usersidColumn(Integer,primary_keyTrue,indexTrue)usernameColumn(String,uniqueTrue,indexTrue,nullableFalse)hashed_passwordColumn(String,nullableFalse)3. schemas.py - Pydantic 模型frompydanticimportBaseModelclassUserBase(BaseModel):username:strclassUserCreate(UserBase):password:strclassUser(UserBase):id:intclassConfig:orm_modeTrueclassToken(BaseModel):access_token:strtoken_type:strclassTokenData(BaseModel):username:str|NoneNone4. auth.py - 认证工具fromdatetimeimportdatetime,timedeltafromjoseimportJWTError,jwtfrompasslib.contextimportCryptContext# 密钥应从环境变量读取此处仅作示例SECRET_KEY09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7ALGORITHMHS256ACCESS_TOKEN_EXPIRE_MINUTES30pwd_contextCryptContext(schemes[bcrypt],deprecatedauto)defverify_password(plain_password,hashed_password):returnpwd_context.verify(plain_password,hashed_password)defget_password_hash(password):returnpwd_context.hash(password)defcreate_access_token(data:dict,expires_delta:timedelta|NoneNone):to_encodedata.copy()ifexpires_delta:expiredatetime.utcnow()expires_deltaelse:expiredatetime.utcnow()timedelta(minutes15)to_encode.update({exp:expire})encoded_jwtjwt.encode(to_encode,SECRET_KEY,algorithmALGORITHM)returnencoded_jwtdefdecode_token(token:str):try:payloadjwt.decode(token,SECRET_KEY,algorithms[ALGORITHM])username:strpayload.get(sub)ifusernameisNone:returnNonereturnusernameexceptJWTError:returnNone5. dependencies.py - 公共依赖项fromfastapiimportDepends,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearerfromsqlalchemy.ormimportSessionfromjoseimportJWTErrorfrom.importdatabase,models,auth,schemas oauth2_schemeOAuth2PasswordBearer(tokenUrl/auth/token)defget_db():dbdatabase.SessionLocal()try:yielddbfinally:db.close()defget_current_user(token:strDepends(oauth2_scheme),db:SessionDepends(get_db))-models.User:usernameauth.decode_token(token)ifusernameisNone:raiseHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailInvalid authentication credentials,headers{WWW-Authenticate:Bearer},)userdb.query(models.User).filter(models.User.usernameusername).first()ifuserisNone:raiseHTTPException(status_code404,detailUser not found)returnuser6. routers/auth.py - 认证路由fromfastapiimportAPIRouter,Depends,HTTPException,statusfromfastapi.securityimportOAuth2PasswordRequestFormfromsqlalchemy.ormimportSessionfrom..importschemas,models,auth,dependencies routerAPIRouter(prefix/auth,tags[authentication])router.post(/register,response_modelschemas.User)defregister(user:schemas.UserCreate,db:SessionDepends(dependencies.get_db)):db_userdb.query(models.User).filter(models.User.usernameuser.username).first()ifdb_user:raiseHTTPException(status_code400,detailUsername already registered)hashedauth.get_password_hash(user.password)new_usermodels.User(usernameuser.username,hashed_passwordhashed)db.add(new_user)db.commit()db.refresh(new_user)returnnew_userrouter.post(/token,response_modelschemas.Token)deflogin(form_data:OAuth2PasswordRequestFormDepends(),db:SessionDepends(dependencies.get_db)):userdb.query(models.User).filter(models.User.usernameform_data.username).first()ifnotuserornotauth.verify_password(form_data.password,user.hashed_password):raiseHTTPException(status_code400,detailIncorrect username or password)access_tokenauth.create_access_token(data{sub:user.username})return{access_token:access_token,token_type:bearer}7. routers/users.py - 用户信息路由fromfastapiimportAPIRouter,Dependsfrom..importschemas,dependencies routerAPIRouter(prefix/users,tags[users])router.get(/me,response_modelschemas.User)defread_users_me(current_userDepends(dependencies.get_current_user)):returncurrent_user8. routers/chat.py - 聊天 WebSocket 路由fromfastapiimportAPIRouter,WebSocket,WebSocketDisconnectfromtypingimportDictfrom..authimportdecode_token routerAPIRouter(prefix/chat,tags[chat])classConnectionManager:def__init__(self):self.active_connections:Dict[str,WebSocket]{}# username - websocketasyncdefconnect(self,websocket:WebSocket,username:str):awaitwebsocket.accept()self.active_connections[username]websocketdefdisconnect(self,username:str):self.active_connections.pop(username,None)asyncdefsend_personal(self,message:str,username:str):ifusernameinself.active_connections:awaitself.active_connections[username].send_text(message)asyncdefbroadcast(self,message:str,exclude_username:strNone):forusername,conninself.active_connections.items():ifusername!exclude_username:awaitconn.send_text(message)managerConnectionManager()router.websocket(/ws)asyncdefwebsocket_endpoint(websocket:WebSocket):# 从查询参数获取 tokentokenwebsocket.query_params.get(token)ifnottoken:awaitwebsocket.close(code1008)returnusernamedecode_token(token)ifnotusername:awaitwebsocket.close(code1008)returnawaitmanager.connect(websocket,username)try:whileTrue:dataawaitwebsocket.receive_text()# 广播消息给所有其他用户awaitmanager.broadcast(f{username}:{data},exclude_usernameusername)exceptWebSocketDisconnect:manager.disconnect(username)awaitmanager.broadcast(fUser{username}left the chat)9. main.py - 应用入口fromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewarefrom.databaseimportenginefrom.modelsimportBasefrom.routersimportauth,users,chat# 创建数据库表Base.metadata.create_all(bindengine)appFastAPI(titleChat App with JWT WebSocket)# 配置 CORSapp.add_middleware(CORSMiddleware,allow_origins[*],# 生产环境应限制allow_credentialsTrue,allow_methods[*],allow_headers[*],)# 包含路由app.include_router(auth.router)app.include_router(users.router)app.include_router(chat.router)app.get(/)defroot():return{message:Welcome to Chat App API}10. README.md - 项目说明# Chat App with FastAPI 一个简单的实时聊天应用包含用户认证和 WebSocket 通信。 ## 功能 - 用户注册 - 用户登录获取 JWT 令牌 - 获取当前用户信息 - WebSocket 聊天室需令牌验证 ## 快速开始 1. 克隆仓库 2. 安装依赖pip install -r requirements.txt 3. 运行服务uvicorn chat_app.main:app --reload 4. 访问 API 文档http://localhost:8000/docs 5. 使用 WebSocket 客户端连接 ws://localhost:8000/chat/ws?tokenyour_jwt_token 进行聊天 ## API 端点 - POST /auth/register - 注册新用户 - POST /auth/token - 登录获取令牌 - GET /users/me - 获取当前用户信息需要认证 - WebSocket /chat/ws - 聊天室连接需提供 token 查询参数运行与测试启动服务uvicorn chat_app.main:app--reload注册用户使用/auth/register端点POST JSON{username:alice,password:secret}登录获取令牌POST/auth/token表单数据username: alice password: secret返回类似{access_token:eyJhbGc...,token_type:bearer}WebSocket 连接使用 WebSocket 客户端如浏览器 JavaScript、Postman、wscat连接到ws://localhost:8000/chat/ws?tokeneyJhbGc...连接成功后可以发送消息所有其他在线用户将收到广播。测试受保护路由在请求头中添加Authorization: Bearer token访问/users/me。注意事项生产环境中请勿将密钥硬编码应使用环境变量。数据库使用 SQLite仅适用于开发测试。WebSocket 连接目前不存储聊天记录仅实时转发。通过这个完整项目你可以看到 FastAPI 的高级特性如何在实际中落地JWT 认证保护 HTTP 和 WebSocket 端点SQLAlchemy 处理数据持久化依赖项简化代码复用WebSocket 实现实时通信。希望这个示例能帮助你进一步掌握 FastAPI构建出更强大的应用
