FastAPI

- 摘要
- 概述
- 快速开始
- 基础应用
- - 路由注册和端点绑定
  - 路由端点传参与校验
  - 请求和响应报文
  - 后台异步任务执行
  - 异常与错误
  - 中间件
  - 数据库操作
  - 应用启动和关闭回调
  - 多应用挂载
  - 自定义配置swagger ui
  - 应用配置信息读取
- 继续学习与最佳实践
- - 安全认证机制*
  - 依赖注入
  - Pydantic
  - Pytest单元测试
  - Linux部署应用程序*
  - 实战常见问题

摘要

本篇博客参考《FastAPI Web开发入门、进阶与实战》对其前半部分的内容进行总结，以便加深理解和记忆。后半部分的内容等日后有时间再继续学习更新。

概述

FastAPI

Python Web服务器网关接口
- WSGI（Python Web Server Gateway Interface）：指定了web服务器和Python web应用或web框架之间的标准接口，以提高web应用在一系列web服务器间的移植性。PEP 333 – Python Web Server Gateway Interface v1.0 | peps.python.org
- ASGI：和WSGI一样，都是为Python语言定义的 Web服务器和Web应用程序或框架之间的一种简单而通用的接口。ASGI是WSGI的一种扩展的实现，并且提供异步特性和WebSocket等的支持。同时ASGI也是兼容WSGI的，在某种程度上可以理解为ASGI是WSGI的超集，所以ASGI可以支持同步和异步同时运行。在Python 3.5增加async/await特性之后，基于asyncio异步协程的应用编程变得更加方便。ASGI协议规范就是用于asyncio框架中底层服务器/应用程序的接口。
- 最新的HTTP支持异步长连接，而传统的WSGI应用支持单次同步调用，即仅在接收一个请求后返回响应，从而无法支持HTTP长轮询或WebSocket连接
OpenAPI规范（OAS）：一个定义标准的与具体编程语言无关的 RESTful API的规范
框架对比（选择框架的最终目的是提高开发效率，实现业务需求。框架对于应用开发本身只是一个辅助实现业务逻辑的工具）
- Bottle：比较小众，最简单、快速和轻量级的WSGI微型Web框架。整个框架只有一个文件模块。框架本身除了Python标准库之外，不产生其他第三方的依赖项。局限性：插件生态比较少，很多功能需要自己实现扩展
- Flask：轻量级的Web应用框架，基于Werkzeug WSGI工具箱和 Jinja2模板引擎而开发出来的。对于一些功能插件保留了弹性扩展，而且持续更新维护
- Django：大而全的框架，部分模块也无法进行定制
- Sanic：和FastAPI框架一样，是一个异步框架。它是首批基于asyncio的极端快速Python框架之一。它允许使用Python 3.5中添加的async/await语法，这使得用户的代码不阻塞，速度更快。它不仅是一个框架，也是一个服务器，可以随时为用户编写的Web应用程序提供部署服务
- Starlette：一个轻量级的 ASGI 框架，用于构建异步 web 应用
- FastAPI：集众框架之长，诞生于2018年12月，在测试领域开始流行，同时支持同步和异步特性，在单线程模式下也可以支持更多的任务并发处理
FastAPI特性
- 支持ASGI（Asynchronous Server Gateway Interface）协议的 Web应用框架，也就是说，它同时兼容ASGI和WSGI的应用
- 天然支持异步协程处理，能快速处理更多的HTTP请求
- 使用了Pydantic类型提示的特性，可以更加高效、快速地进行接口数据类型校验及模型响应等处理，它还可以自动对响应数据进行格式化和序列化处理
- 提供依赖注入系统的实现，它可以让用户更高效地进行代码复用
- 它支持WebSocket、GraphQL等
- 支持异步后台任务，可以方便地对耗时的任务进行异步处理
- 支持服务进程启动和关闭事件回调监听，可以方便地进行一些插件的扩展初始化
- 支持跨域请求CORS、压缩Gzip请求、静态文件、流式响应
- 支持自定义相关中间件来处理请求及响应
- 支持开箱即用OpenAPI（以前被称为Swagger）和JSON Schema，可以自动生成交互式文档
- 使用uvloop模块，让原生标准的asyncio内置的事件循环更快

快速开始

导入

pip install fastapi

常用依赖库

email.validator：主要用于邮件格式校验处理
requests：使用单例测试TestClient或请求第三方接口时需要使用该依赖库
aiofiles：主要用于异步处理文件读写操作
jinja2：主要供用户渲染静态文件模板时使用，当项目要使用后端渲染模板时安装该依赖库
Python-multipart：当需要获取From表单数据时，只有通过这个库才可以提取表单的数据并进行解析
itsdangerous：用于在SessionMiddleware中间件中生成Session临时身份令牌
graphene：需要GraphQLApp支持时安装这个依赖库
orjson：主要用于JSON序列化和反序列化，如使用FastAPI提供的ORJSONR-esponse响应体处理时，则需要安装这个依赖库
ujson：主要用于JSON序列化和反序列化，如需要使用FastAPI提供的UJSONResponse响应体时就需要安装该依赖库
uvicorn：主要用于运行和加载服务应用程序Web服务

项目构建

# app.y
# 定义当前源文件的编码方式
# -*- coding: utf-8 -*-
import os.path
import pathlib
import uvicorn
from fastapi import FastAPI, Request
from fastapi.routing import APIRoute# 3.预设路由列表
async def fastapi_about():return JSONResponse({"data": "about"})routes = [APIRoute(path="/about", endpoint=fastapi_about, methods=["GET", "POST"])]# 4.全局异常/错误捕获
async def exception_not_found(request, exc):return JSONResponse({"code": exc.status_code,"error": "404 NOT FOUND",}, status_code=exc.status_code)exception_handlers = {404: exception_not_found,
}"""一.实例化FastAPI类得到应用程序实例对象，代表当前进程的一个实例
"""
app = FastAPI(# 1.交互式文档参数描述title="学习FastAPI框架", description="有关FastAPI框架文档的介绍和描述", version="0.0.1",openapi_prefix='',  # 访问openapi_json.json文件路径的前缀swagger_ui_oauth2_redirect_url="/docs/oauth2-redirect",  # 使用OAuth2进行身份认证时授权URL重定向地址swagger_ui_init_oauth=None,  # 自定义OAuth认证信息配置docs_url="/docs",  # swagger ui的请求地址，填为None则可关闭redoc_url="/redoc",  # redoc ui的请求地址，填None则可关闭terms_of_service="http://www.lincat.work",  # 团队服务条款的URLdeprecated=None,  # 是否标注所有API为废弃标识# 联系人contact={"name": "李一帆","url": "http://www.lincat.work","email": "liyifan999@nenu.edu.cn"},# 配置公开API的许可信息license_info={"name": "版权信息说明License v1.0","url": "http://www.lincat.work"},# 默认接口的分组列表信息，可定义相关API分组Tag名称openapi_tags=[{"name": "接口分组","descrition": "接口分组信息说明"}, ],# 服务请求地址相关的参数说明servers=[{"url": "/","description": "本地调试环境"}, {"url": "http://www.lincat.work","description": "线上测试环境"}, {"url": "http://www.lincat.work","description": "线上生产环境"}],# 2.关闭OPEN API：生产环境中开启此文档访问会带来安全隐患，使用身份认证或IP白名单来限制# openapi_url=None,# 3.设置路由参数列表routes=routes,# 4.全局异常捕获exception_handlers=exception_handlers,# 5.开启DeBug模式，在接口函数内设一个“1988/0”的错误"ZeroDivision-Error“，访问某接口地址时，页面会# 显示出详细的错误堆栈异常信息。注意线上生产环境应避免开启该功能，另外使用DeBug模式会使全局异常处理失效debug=True)"""二.挂载静态文件
"""
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, JSONResponsetemplates = Jinja2Templates(directory=f"{pathlib.Path.cwd()}/template/")
staticfiles = StaticFiles(directory=f"{pathlib.Path.cwd()}/static/")
app.mount("/static", staticfiles, name='static')"""三.写Restful接口
"""@app.get("/", tags=['hello'], response_class=HTMLResponse)
async def get_response(request: Request):return templates.TemplateResponse("index.html", {"request": request})@app.get("/hello", tags=['hello'])
def app_hello():"""tags参数表示该API归属于某个分组标签下，进行API分组归档@app.get可以修饰def定义的同步函数，也可以修饰async def定义的协程函数同步函数会运行于外部的线程池中，协程函数会运行于异步事件循环中。同步函数是基于多线程并发模式处理的，协程函数是基于单线程内的异步并发模式处理的"""# return {"data": {"Hello FastAPI"}}return JSONResponse({"data": {"Hello FastAPI"}})"""四.应用启动
"""
if __name__ == '__main__':"""1.使用命令方式启动：uvicorn app:app --reload2.导入uvicorn模块（ASGI服务容器），使用代码方式启动reload参数用于热启动，代码修改程序会自动重启服务进程，提高编码效率，部署时需注释掉将host设置为：0.0.0.0即可允许局域网内提供服务app参数有3种默认类型ASGIApplication的实例化对象可调用对象(app)字符串：指出“模块名+app对象名”         """app_module_name = os.path.basename(__file__).replace(".py", "")uvicorn.run(app=f'{app_module_name}:app', host='127.0.0.1', port=8000, reload=True)

不建议在异步操作中使用同步函数，因为在一个线程中执行同步函数必定会引起阻塞，如一般不在异步协程函数中使用time.sleep()而是使用await asyncio.sleep()

基础应用

路由注册和端点绑定

路由注册信息

# app.routes保存了所有路由的注册信息
print(app.routes)

路由注册

路由装饰器

def get("""与API可视化文档显示有关的字段描述"""# tags:设置API文档中接口所属组别的标签名，可以将其理解为分组名称，支持设定多个所属分组。# summary:设置API文档中该路由接口的名称，默认值为当前被装饰的函数(又称端点函数或视图函数)的名称# description:设置API文档中对该路由功能的详细描述# response_description:设置API文档中对该路由响应报文信息结果的描述。# deprecated:设置API文档中是否将该路由标记为废弃接口# operation id:自定义设置路径操作中使用的OpenAPI的operation_id名称。# name:API文档中该路由接口的名称。其功能和summary类似，但是name主要供用户反向查询使用。两者同时存在时会优先显示summary# openapi_extra:用于自定义或扩展API文档中对应的openapi_extra字段的功能。# include in schema:表示该路由接口相关信息是否在API文档中显示。"""与响应报文处理有关的字段描述"""# path:定义路由访问的URL地址。# response_model:定义函数处理结果中返回的JSON的模型类，这里会把输出数据转换为对应的response_model中声明的数据模型。# status_code:定义响应报文状态码。# 设置响应报文使用的Response类，默认返回response class:JSONResponse .# responses:设定不同响应报文状态码下不同的响应模型# response_model include:设置响应模型的JSON信息中包含哪些字段，参数格式为集合{字段名，字段名，…}。# response_model_exclude:设定响应模型的JSON信息中需要过滤哪些字段。# response model exclude unset:设定不返回响应模型的JSON信息中没有值的字段。# response model exclude defaults:设定不返回响应模型的JSON信息中有默认值的字段。# response modelexclude none:设定不返回响应模型的JSON信息中值为None的字段。"""其他字段信息"""# dependencies:配置当前路径装饰器的依赖项列表	
)

多重URL地址绑定函数（多个URL映射到一个端点）

# index视图函数同时被多个装饰器修饰
@app.get('/',response_class=JSONResponse)
@app.get('/index',response_class=JSONResponse)
@app.post('/index',response_class=JSONResponse)
@app.get('/app/hello',tags=['index'])
def index():return {"data":"hello"}

静态路由和动态路由（参数动态化）

# 动态路由
@app.get('/usr/{userId}')
async def login(userId:str):return {"data":"dynamic"}
# 静态路由
@app.get('/usr/userId')
async def login(userId:str):return {"data":"static"}

谁先注册先映射到谁

app.api_route装饰器

@app.api_route(path='/index', methods=["GET","POST"])
async def index():return {"data":"index"}

APIRouter

注意APIRouter与APIRoute是不同的，前者主要用于定义路由组（一个路由组的根路由），可以在大型项目中针对不同的业务模块分级分组。APIRoute则表示具体的路由节点

router_user = APIRouter(prefix='/user',tags=['用户模块'])
router_pay = APIRouter(prefix='/pay',tags=['支付模块'])"""1.装饰器方式"""
# 1.1单一方法
@router_usr.get("/user/login")
def user_login():return {"data":"OK"}# 1.2同时配置多个方法
@router_pay.api_route("/pay/buy", methods=['GET','POST'])
def user_pay():return {"data":"OK"}"""2.函数调用方式"""
def user_login():return {"data":"OK"}def user_pay():return {"data":"OK"}# 2.1直接添加
router_user.add_api_route("/user/login", endpoint=user_login,methods=['GET'])
# 2.2定义APIRoute实例
user_pay.add_api_route(APIRoute(path="/pay/buy", endpoint=user_pay,methods=['GET','POST']))# 3.添加路由分组（必不可少）
app.include_router(router_user)
app.include_router(router_pay)

路由端点传参与校验

参数校验库

库	介绍
WTForms	支持多个Web框架的Form组件，主要用于对用户请求数据进行验证
valideer	轻量级、可扩展的数据验证和适配库
validators
cerberus	用于Python的轻量级和可扩展的数据验证库
colander	用于对XML、JSON、HTML以及其他同样简单的序列化数据进行校验和反序列化的库
isonschema	用来标记和校验JSON数据，可在自动化测试中验证JSON的整体结构和字段类型
schematics	用于将类型组合到结构中并验证它们，然后根据简单的描述转换数据的形状
voluptuous	主要用于验证以JSON、YAML等形式传入Python的数据

FastAPI基于Pydantic的参数校验与路由端点传参

路径操作参数：路由中的参数；路径函数操作：视图函数参数

路径参数是URL的关键组成部分，如果缺少对路径参数值的传递，则无法构成完整的URL请求，所以任何路径参数都应该声明为必选项

在路径参数中，无默认参数值的参数应该放到有默认值的参数前

【GET]

# 1.Path参数
# 1.1 带/的关键子路径：若传入的路径参数带/，如文件类型的路径，则URL会识别出多重路径，因此需要进行修饰
@app.get("/uls/{file_path:path}")
async def callback_file_path(file_path:str):return {'data':file_path}
# 1.2 Path参数校验
@app.get("/pay/{user_id}/artical/{artical_id}")
async def read_user_artical(user_id: int=Path(# default默认值,...表示必传...,# API开放交换文档描述信息title="用户ID",description="用户信息ID",# 参数校验# ge:参数值 ≥ ，gt:＞，lt：＜，le：≤ge = 1000),artical_id: str=Path(default="index",tittle="文章ID",description="文章ID"，min_length = 1max_length = 50)):return {"data":{"user_id":user_id,"artical_id":artical_id}}# 2.Query：用于GET请求中，非路径参数的查询参数（？后）,其参数与Path基本相同
@app.get("/query")
async def callback(# bool类型会被FastAPI进行参数转换：如true/false,1/0,on/off会被转换为True/Falseisbool:bool=False,user_id: Optional[int] = None,user_name: str = Query(None,min_length=1),# 列表传参：http://ip:port/query?q=test1&q=test2q: List[str] = Query(["test1","test2"])):pass

【POST】

FastAPI会将参数识别为JSON格式字符串，并自动将字段转换为对应的数据类型；自动进行参数规则的校验，校验失败会返回错误，指出错误的位置和信息；为模型生成JSON Schema定义，并显示在API交互文档中

1）引入Pydantic模型来声明请求体并绑定此处给出了分文件的示例

# record_scout.py
from pydantic import BaseModel
class RecordScout(BaseModel):"""侦察决策"""pilot_id: str = None  # 飞行员idpilot_name: str = None  # 飞行员姓名id: str = None  # 仿真飞行idrecon_route: str = None  # 侦查航路，具体航路点信息以及航路更新时间（具体存储的时候直接存储解析出数据的json串）recon: str = None  # 侦查方案（雷达+光电或者光电）detection_area: str = None  # 侦查区域分配（目前无该数据）target_detection: str = None  # 目标探测情况# api_record_scout.py
from fastapi import APIRouter
from models.dbpool import pgdbpool
from models.record_scout import RecordScoutrouter = APIRouter(prefix="/record_scouting", tags=['侦查决策记录表'])
@router.post("/add/", tags=['侦查决策记录表'])
def add_record_scouting(record_scout: RecordScout):""" 增加一条侦查决策记录"""try:pgdb = pgdbpool.get_conn()cursor = pgdb.cursor()cursor.execute(f"""INSERT INTO record_scouting (pilot_id,pilot_name,id,recon_route,recon,detection_area,target_detection)VALUES (%s,%s,%s,%s,%s,%s,%s)""",(record_scout.pilot_id, record_scout.pilot_name, record_scout.id, record_scout.recon_route,record_scout.recon, record_scout.detection_area, record_scout.target_detection))pgdb.commit()result = {"code": 200, "data": record_scout, "msg": "success"}except Exception as e:result = {"code": -1, "data": [], "msg": str(e)}return result# main.py
app.include_router(api_record_scouting.router)

2）Body类绑定请求体参数（属性与Path、Query相似）

# 1.单值
@app.post("/action/body")
def callbackbody(token:str = Body(...),user_id:int = Body(...,gt=10),artical_id:str = Body(default=None)
):pass
{"token":"string","user_id":0,"artical_id":"string"
}# 2.embed参数
class Item(BaseModel):user_id: int = Body(...,gt=10)# Field字段和Path、Query等一样，但它只能用于类内字段的校验和定义toekn: str = Field(...,description="")
@app.post("/action/")
def read_item(item:Item = Body(default=None,# embed:False表示Item不会成为请求体的一部分，True表示会成为请求体的一部分embed=False))
False:{"user_id":0,"token":"string"}
True:{"item":{"user_id":0,"token":"string"}}# 3.多个Request Body参数
class ItemUser(BaseModel):pass
class User(BaseModel):pass
@app.put("/item/")
async def update_item(item: ItemUser,user:User):pass
{"item":{},"user":{}
}# 4.多个模型和单个Request Body
class ItemUser(BaseModel):pass
class User(BaseModel):pass
@app.put("/item/")
async def update_item(item: ItemUser,user:User,importance:int=Body(...)):pass
{"item":{},"user":{},"importance":0
}# 5.模型嵌套声明
class User(BaseModel):pass
class ItemUser(BaseModel): # 嵌套类user: User# 嵌套列表users: List[User]# 嵌套集合，传输时会转换为列表传输tags: Set[int]
@app.put("/item/")
async def update_item(item: ItemUser,importance:int=Body(...)):pass
{"item":{"user":{}"users":["user":{},]"tags":[]},"importance":0
}# 6.任意dict字典构成请求体
@app.post("/demo/dict/")
async def update_item(item:Dict[str,str],user:Dict[str,Dict[str,str]])pass
{"item":{"additionalProp1":"string","additionalProp2":"string","additionalProp3":"string",}"user":{"additionalProp1":{"additionalProp1":"string","additionalProp2":"string","additionalProp3":"string",},"additionalProp2":{"additionalProp1":"string","additionalProp2":"string","additionalProp3":"string",},"additionalProp3":{"additionalProp1":"string","additionalProp2":"string","additionalProp3":"string",},}
}

3）Form数据和文件处理

"""
1.表单数据：表单数据默认使用POST传输，表单传输过程使用特殊编码与JSON不同，设置的Content-Type为：application/x-www-form-urlencoded
"""
# 1.1安第三方库
pip install python-multipart
# 1.2接收表单数据，Form是根据Body扩展而来
@app.post("/demo/login")
async def login(username:str=Form(...,title="用户名"),password:str=Form(...)):return:{"data":{"username":username,"password":password}}"""2.文件上传：文件类型数据的Content-Type为：multipart/form-data"""
# 2.1File bytes上传,File类是基于Form扩展的
# 2.1.1同步读
@app.post("/sync_file")
def sync_file(file:bytes=File(...)):with open('./data.bat','wb') as f:f.write(file)return {'file_size':len(file)}
# 2.1.2异步读取
pip install aiofiles
import aiofiles
@app.post("/async_file")
def async_file(file:bytes=File(...)):async with aiofile.open('./data.bat','wb') as f:await f.write(file)return {'file_size':len(file)}# 2.2 UploadFile接收文件上传
# File对象通过字节流读取文件，缺乏文件元数据信息，如文件名称、文件格式类型等，若想获取文件类型则需从请求头中截取
# UploadFile更加高级，当读取文件大小超过内存时会保存在磁盘中，因而可以读取大文件，包含文件元数据信息，包含对文件处理的异步接口；但只能异步处理文件
@app.post("/async_file")
def async_file(file:UploadFile=File(...)):content = await file.read()with open('./data.bat','wb') as f:await f.write(content)return {'file_name':file.filename,'content-type':file.content_type}# 2.3多文件上传
@app.post("/async_file")
def async_file(file:List[UploadFile]=File(...)):pass
@app.post("/async_file")
def async_file(file:List[bytes]=File(...)):pass

4）读Header和Cookie

# Header
@app.get('/header/')
async def read(# HTTP默认headers参数，convert_underscores=True用于将user-agent转为user_agent，以便成为合法变量user_agent: Optional[str] = Header(None,convert_underscores=True),# 重名请求头参数：以列表形式读x_token:List[str] = Header(None))
# Cookie：Cookie和Session机制理解
# 服务端给客户端分发Cookie
@app.get("/set_cookie")
def setCookie(response:Response):response.setCookie(key="lyf", value="liyifan999@nenu.edu.cn")return {"data":"OK"}
# 服务器端读客户端携带的Cookie
@app.get("/get_cookie")
async def getCookie(lyf:Optional[str]=Cookie(None)):return {"data":lyf}

请求和响应报文

请求报文

构成
- 请求行：请求方法、URL、协议/版本
- 请求报文头（键值对）
  - User-Agent：发出请求的代理用户（浏览器）
  - Accept：客户端接收的响应主题类型（text/HTML）
  - Host：服务器的主机名（ip）和端口号
  - Cookie：指定与请求关联的Cookie
  - Referer：指定链接到当前页面的上一个页面URL
  - Authorization：用于身份验证的凭据
  - Cache-Control：指定缓存控制选项，控制响应是否可以缓存及缓存方式
- 空行：标识请求头结束
- 请求主体正文：主要是Body数据
Request对象

# 由于FastAPI基于Starlette扩展而来，它实际直接使用了Starlette的Requst，因此也可以直接导入Starlette的Request
from fastapi import Request@app.get("/get_request")
async def get_request("""Request类的属性app:当前请求的上下文应用url:当前请求的完整URL对象_url:当前请求完整的URL对象components:表示请求URL包括哪些部分，协议、请求地址、路径、查询参数等path:URL的路径信息port:URL的端口号query:请求URL提交的字符串形式的查询参数scheme:请求URL使用的协议HTTP还是HTTPSis_secure:请求URL是否启用HTTPS安全校验base_url:请求的服务URL地址method:请求方法client:当前请求客户端的信息，host和portCookies:请求报文中提交的Cookies值字典信息headers:请求报文中所有的请求头信息path_params:当前请求提交的路径参数字典信息query_params:请求的查询参数session:Sessionstate:请求的状态值，通常作为上下文传递scope:请求范围"""request: Request):# 注意以下三个是协程对象，只能在协程函数中才能正常读取解析form_data = await request.form()body_data = await request.body()# json方法能读到值的前提是body有具体的值json_data = await request.json() if body_data else Nonereturn {'url': request.url,'base_url': request.base_url,'client_host': request.client.host,'query_params': request.query_params,'form_data': form_data,'body_data': body_data,'json_data': json_data}

响应报文

构成
- 状态行：协议/版本、状态码、状态信息（HTTP/1.1 200OK）
- 响应头：键值对
  - Content-Type：响应主体的媒体类型
  - Accept：客户端接收的响应主体类型
  - Content-Length：响应主体长度
  - Cookie
  - Server：提供服务的Web服务器软件
  - Date：日期和时间
  - Cache-Control：缓存控制选项
  - Set-Cookie：设置Cookie
  - Location：指定重定向URL
- 空行：表示响应头的结束
- 响应正文主体：响应内容信息
HTTP状态码分类
- 1xx：信息性状态码，表示正在处理请求
  - 100 Continue：服务器已收到请求头，且客户端应继续发送请求主体
  - 101 Switch Protocols：客户端请求升级协议，如WebSocket
- 2xx：成功状态码，表示服务器已成功收到请求并成功处理
  - 200 OK
  - 201 Created：服务器创建了一个新资源
  - 204 No Content：请求成功，但响应不包含任何数据
- 3xx：重定向状态码，表示必须采取进一步的行动才能完成请求
  - 301：Moved Permanently：资源永久移动到新位置（URL路由改变）
  - 302：Found：资源暂时移动到新位置（URL不变）
  - 304：Not Modified：客户端缓存仍有效
- 4xx：客户端错误状态码，客户端提交参数错误或语法错误，服务器无法完成请求
  - 400 Bad Request：请求无效或无法被服务器理解
  - 401 Unauthorized：请求需要身份验证
  - 403 Forbidden：服务器拒绝请求
  - 404 Not Found：请求的资源不存在
- 5xx：服务器错误状态码
  - 500 Internal Server Error：服务器遇意外情况无法完成请求
  - 503 Service Unavailable：服务器无法处理请求，因为它过载或正在进行维护
指定HTTP状态码

# 1.
@app.get("/",status_code=500)
async def set_http_code():pass
# 2.
from fastapi import status
@app.get("/",status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
async def set_http_code():pass
# 3.
from fastapi import status
@app.get("/")
async def set_http_code(response:Response):response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
# 4.
from fastapi import status
from fastapi.response import JSONResponse
@app.get("/")
async def set_http_code():return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)

response_model定义响应报文内容：直接输出Pydantic模型返回。优点：可以进行数据验证、完善API文档、可以做返回模板

from pydantic import BaseModel
class UserIn(BaseModel):username: strpassword: str
class UserOut(BaseModel):username: str
# 基于此种模型输出，可以自动进行数据的类型转换（实际是对应字段的填充）忽略UserIn的password不报错
@app.post("/user",response_model=UserOout# response_model_exclude:指定返回数据中应排除的字段# response_model_include:指定响应模型中返回数据应包含的字段# response_model_exclude_unset:指定模型中返回数据应排除返回值为空的字段# response_model_exclude_defaults:指定模型中返回数据应排除返回值带有默认值的字段# response_model_exclude_None:指定模型中返回数据应排除返回值为None的字段)
async def create_user(user:UserIn):return user

Response类及其子类

# HTMLResponse
from fastapi.response import HTMLResponse
"""使用html模板的示例见快速开始，也可直接返回html静态文件，但无法动态传入参数"""
@app.get("/",response_class=HTMLResponse)
async def index():html = """<!DOCUMENTTYPE HTML><html><head><tittle>Hello</tittle></head><body></body><html>"""return HTMLResponse(content=html,status_code=200)# JSONResponse:FastAPI默认会将字典按JSONResponse封装返回
from fastapi.response import JSONResponse
@app.get("/",response_class=JSONResponse)
async def index():return JSONResponse(status_code=404,content={"ok"})# PlainTextResponse
from fastapi.response import PlainTextResponse
@app.get("/",response_class=PlainTextResponse)
async def index():return PlainTextResponse(status_code=404,content={"ok"})# RedirectResponse重定向
from fastapi.response import HTMLResponse,RedirectResponse
@app.get("/",response_class=HTMLResponse)
async def index():# 外部地址重定向redirect  = RedirectResponse("http://www.lincat.work",status_code=301)# 内部地址重定向redirect  = RedirectResponse("/index",statu_code=302)return redirect# StreamingResponse 多用于流媒体
from fastapi.response import StreamingResponse
@app.get("/stream_video")
def stream_video():return StreamingResponse(read_in_chunks(),media_type="multipart/x-mixed-replace;boundry=frame")# FileResponse 文件下载
@app.post("download_file")
async def download_file():return FileResponse(path="./data.bat", filename='data.bat')# 自定义Response类型（xml）
@app.get_xml("/")
def get_xml():data = """<node><to>lyf</to></node>"""return Response(content=data, media_type="application/xml")

后台异步任务执行

后台异步任务用于处理耗时操作，可以是同步任务也可以是异步任务

def send_mail(n):time.sleep(n)
@app.post("/index")
async def index(tasks:BackgroundTasks):tasks.add(send_mail,10)return {"data":"index"}

异常与错误

错误：表示应用程序存在较为严重的问题，可能会导致应用崩溃或无法正常运行

异常：表示应用程序在运行中出现了可预测和可恢复的问题，这些问题可以被捕获和处理，应用可以继续运行

在FastAPI框架中，错误和异常都是Exception的子类，如HTTPException、RequestValidationError

# 自定义异常
class CustomException(Exception):def __init(self, message:str):self.message = message
# 全局异常捕获处理
@app.exception_handler(CustomException):
async def custom_exception_handler(request: Request, exc: CustomExcetion):return JSONResponse(content={"message":exc.message})# 异常抛出
@app.get("/custom_exception")
async def read_unicorn(name: str='lyf'):if name == "lyf":raise CustomException(message="抛出自定义异常")return {"name":name}

中间件抛出异常

中间件抛出异常的情况下，是无法通过全局异常类拦截自定义异常的，因为FastAPI框架在底层中对任何类型的中间件抛出的异常都统归一于顶层的ServerErrorMiddleware中间件进行捕获，并在ServerErrotMiddleware中捕获的的异常以Exception的方式抛出。

为了捕获中间件抛出的异常，要在全局异常捕获修饰函数中增加对Exception的捕获，再根据异常类分类处理

@app.exception_handler(Exception)
async def custom_exception_handler(request:Request,exc:Exception):if isinstance(exc,CustomException):print("触发全局自定义CustomException")return JSONResponse(content={"message":exc.message})

中间件

FastAPI的中间件是在处理HTTP请求和响应之前或之后添加的组件，允许开发者对HTTP请求进行重写、过滤、修改和添加信息，以及对HTTP响应进行修改或处理。中间件具有轻量、低级别、可拔插等特点，可以在全局范围内对客户端的请求进行拦截。FastAPI中的中间件只是一个简单类，遵循ASGI规范。其应用场景有：请求跨域处理、API限流限速、对接口进行监控、日志请求记录、IP白名单限制处理、请求权限校验、请求缓存校验、认证和授权、压缩响应内容等

# 1.HTTP请求中间件：注意request和call_next必须加入，call_next表示下一个符合ASGI协议的APP对象（RequestResonseEndpoint）
@app.middleware("http")
async def middleware1(request: Request, call_next):return response# 2.跨域中间件
# 解决跨域的其他处理方法
# 2.1 使用代理机制（同源服务器下的后端进行代理请求以获取非同源服务下的资源数据，之后返回给同源服务器）
# 2.2 使用jsonp方式，仅支持GET请求
# 2.3 使用CORS方式，支持的请求方式更多，浏览器兼容性更大
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 设置允许的源域名allow_credentials=True,  # 设置是否允许携带cookieallow_methods=["*"],  # 设置允许的请求方法allow_headers=["*"],  # 设置允许的请求头
)# 3.HTTPSRedirectMiddleware 强制所有请求使用https或wss协议
# 4.TrustedHostMiddleware 强制要求Header中的host选项必须来自某个指定的host才允许访问对应的地址# 5.自定义中间件
from starlette.middleware.base import BaseHttpMiddleware
class CustomMiddleware(BaseHTTPMiddleware):async def dispatch(self, request:Request,call_next):passreturn resposne
app.add_middleware(CustomMiddleware)# 6.基于中间件获取响应报文
class CustomMiddle:aysnc def __call__(self,request: Request,call_next: RequestResponseEndpoint,*args,**kwargs):response = await call_next(request)return response
# 注意添加方式
app.middleware('http')(CustomMiddle())

数据库操作

ORM框架：SQLAchemy、SQLModel

数据库操作的同步和异步

Redis的异步操作工具：aioredis

应用启动和关闭回调

应用的启动和关闭回调可以注册多个回调事件（启动和关闭时），事件可以是同步的也可以是异步的。在基于startup和shutdown事件回调处理业务逻辑时不建议添加中间件注册处理

# 启动回调：多用于设置和读取应用配置参数、通过对数据库初始化对象存储到app上下文中、初始化第三方插件库
@app.on_event("startup")
async def start_up_event():pass
# 关闭回调：windows系统中，要出发关闭回调，则需要以命令行启动应用，且以ctrl+c关闭应用才能触发
@app.on_event*("shutdown")
def shut_down():pass

多应用挂载

项目庞大需要拆分时，除了APIRouter可以对模块进行肘，也可以通过主从应用挂载来解决这一问题

挂载子FastAPI应用

# 主应用的交互文档为：http://127.0.0.1:8000/docs
app = FastAPI(tittle="主应用")
# 子应用的交互文档为：http://127.0.0.1:8000/subapp/docs
subapp = FastAPI(tittle="子应用")
app.mount(path='/subapp',app=subapp,name='subapp')

挂载其他WSGI应用

通过WSGI Middleware，WSGI应用（Flask、Django）也可以通过FastAPI挂载和部署启动

from fastapi.middleware.wsgi import WSGIMiddleware
app = FastAPI(tittle="主应用")
flasK_app = Flask(__name__)
app.mount(path='/flaskapp',app=WSGIMiddleware(flask_app),name='flask_app')

自定义配置swagger ui

问题：在使用API可视化交互文档时，有时网络是正常的，但依然无法正常加载可视化API文档页面，这是因为内置swagger ui的静态文件是从第三方的CDN服务商上加载的，CDN服务问题会导致无法正常加载可视化API文档页面
解决办法：将swagger ui静态文件，预先下载到本地，再自定义配置swagger ui，使其从本地加载
docs渲染HTML内容的位置为：...\fastapi\openapi\docs.py中get_swagger_ui_html
下载swagger-ui-bundle.js、swagger-ui.css、favicon.png到本地，放到static/swagger文件夹下
自定义API交互接口

# 挂载
staticfiles = StaticFiles(directory=f"{pathlib.Path.cwd()}/static/")
app.mount("/static", staticfiles, name='static')@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():return get_swagger_ui_html(openapi_url=app.openapi_url,title=f'{app.title}-Swagger UI',swagger_js_url="/static/swagger/swagger-ui-bundle.js",swagger_css_url="/static/swagger/swagger-ui.css",swagger_favicon_url="/static/swagger/favicon.png"
)

redoc模式也可按上述步骤进行

应用配置信息读取

应用程序通常在启动前读取相关的配置参数，大部分配置参数通常不会硬编码到项目中，而是通过外部文件或环境变量中读取。在大型微服务应用中，还可能把参数写入线上的配置中心进行统一管理，通过配置中心就可以做到不重新打包、发布版本就变更参数

基于配置文件的读取

# *.ini（windows）
[fastapi]
debug = True
tittle = "FastAPI"
[redis]
ip = 127.0.0.1
port = 6379
password = 123456

import configparser
config = configparser.ConfigParser()
config.read('conf.ini', encoding='utf-8')
app = FastAPI(debug=bool(config.get('fastapi_config','debug'))tittle=config.get('fastapi_config','tittle')
)

基于Pydantic和.env环境变量读取配置

通过环境变量读取配置参数是FastAPI官方推荐的方式，通过Pydantic可以直接解析出对应的配置参数项

# 读取环境变量依赖包
pip install python-doten# 定义配置文件.env内容
DEBUG=true
TITTLE="FastAPI"
DESCRIPTION="FastAPI文档明细描述"
VERSION="v.1.0.0"

使用读取环境变量的方式，系统会自动解析当前环境是否存在对应的值，若不存在则使用默认值，若配置类中没定义默认值，则触发异常校验

# 定义配置类
from pydantic import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):debug: bool = Falsetittle: strdescription: strversion: strclass Config:env_file = ".env"env_file_encoding = 'utf-8'@validator(# 表示需要校验哪个字段fields="version", # 表示自定义的校验规则是否在标准验证器之前调用，否则在其后调用pre=pre,# each_item:表示对于集合类型，是否对其单个元素校验# check_fileds:表示是否进行字段是否存在的校验# always:表示字段缺失时是否进行校验# allow_reuse:表示存在另一个验证器引用修饰函数，是否跟踪并引发错误抛出)def version_len_check(cls, v:str) -> Optional[str]:if v and len(v) == 0:return Nonereturn v# 指定读取.env文件的方式
# settings = Settings(_env_file='.env', _env_file_encoding='utf-8')
# print(settings.debug)# 对于非单例模式的实例对象，可以通过添加缓存的方式来避免多次初始化，提高整体性能
@lru_cache()def get_settings():# 实例化Settings对象，完成.env文件解析return Settings()app = FastAPI(debug=get_settings().debug)