当前位置：首页 > article >正文

Fastapi 学习使用

article 2025/6/3 6:05:57

Fastapi 学习使用

Fastapi 可以用来快速搭建 Web 应用来进行接口的搭建。

参考文章：https://blog.csdn.net/liudadaxuexi/article/details/141062582

参考文章：https://blog.csdn.net/jcgeneral/article/details/146505880

参考文章：https://blog.csdn.net/qq_21201267/article/details/121673693

参考文章：https://fastapi.tiangolo.com/zh/tutorial/testing/

参考文章：https://fastapi.tiangolo.com/zh/deployment/

1. 快速上手

# 安装三方包
pip install fastapi[all]
pip install uvicorn[standard]

简单的 Demo

import uvicorn
from fastapi import FastAPIapp = FastAPI(title="测试小Demo",description="文档描述",version="1.0",
)@app.get("/", tags=["demo"])
def index():return {"message": "Hello world"}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

2. request 的使用

FastAPI提供了Request对象，通过对HTTP请求报文的解析以更简便地处理来自客户端的请求信息。我们知道在HTTP协议中HTTP请求报文主要包括：

请求行（Request Line）
请求头（Headers）
空行（表示头部结束）
请求体（Body，可选）

示例：

POST /login HTTP/1.1
Host: example.com
User-Agent: curl/7.64.1
Accept: */*
Content-Type: application/json
Content-Length: 48{"username": "alice","password": "123456"
}

在FastAPI，用户可以通过Request对象获取它们对应的信息集合。

2.1 客户端信息集合

import uvicorn
from fastapi import FastAPI, Requestapp = FastAPI(title="测试小Demo",description="文档描述",version="1.0",
)@app.get("/", tags=["demo"])
async def index():return {"message": "Hello world"}@app.get("/test", tags=["test"])
async def client(request: Request):host = request.client.host  # 使用 request 获取客户端的信息;port = request.client.portprint(f"{host}:{port}")return {"client": request.client}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

后续不在复制完整的代码到笔记中来，只进行简单的代码进行复制；

@app.get("/test", tags=["test"])
async def client(request: Request):host = request.client.hostport = request.client.portprint(request.headers)print(request)return {"client": request.client}"""
# >>> 终端打印的数据如下所示；
Headers({'host': '127.0.0.1:8000', 'connection': 'keep-alive', 'cache-control': 'max-age=0', 'sec-ch-ua': '"Google Chrome";v="137", "Chromium";v="137", "Not/A)Brand";v="24"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', 'upgrade-insecure-requests': '1', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36', 'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', 'sec-fetch-site': 'none', 'sec-fetch-mode': 'navigate', 'sec-fetch-user': '?1', 'sec-fetch-dest': 'document', 'accept-encoding': 'gzip, deflate, br, zstd', 'accept-language': 'zh-CN,zh;q=0.9'})
<starlette.requests.Request object at 0x0000024ED531C370>
INFO:     127.0.0.1:4166 - "GET /test HTTP/1.1" 200 OK
"""

尝试获取更多的信息

@app.get("/test", tags=["test"])
async def client(request: Request):host = request.client.hostport = request.client.portprint(request.headers)print(request.url)  # 获取到 URL 的信息;print(request.cookies)  # 获取到 cookie 的信息的# return {"client": request.client}v = request.json()  # 返回的是一个 Request json 的对象print(v)return {"message": "hello world"}

其他的常见的形式

@app.get("/test", tags=["test"])
async def client(request: Request):request.body()  # 获取原始的请求体, 返回二进制的数据;request.json()  # 把请求体的解析为 json.return {"message": "hello world"}

获取表单的数据

import uvicorn
from fastapi import FastAPI, Request, Form
from watchfiles import awatchapp = FastAPI(title="测试小Demo",description="文档描述",version="1.0",
)@app.post("/req_form")
async def req_form(request: Request, name: str = Form(...), hobby: str = Form(...)):"""获取 form 表单的信息;:param request::param name::param hobby::return:"""v = {"form_name": name,"form_hobby": hobby,"form_params": await request.form()  # 获取到的 form 表单提交的信息,封装成为 form 的格式.}print(v)return {"message": "hello world"}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

获取文件的信息

@app.post("/req_file")
async def req_file(name: str = Form(...), resume_file: UploadFile = File(...)):res = {"name": name,"resume_file": resume_file,"resume_file_name": resume_file.filename,"resume_file_headers": resume_file.headers,"resume_file_size": resume_file.size,"resume_file_content_type": resume_file.content_type,"resume_file_content": await resume_file.read(),}return res

2.2 参数模型校验

在 Fastapi 中，请求参数模型（Request Parameter Models）主要是指通过 Pydantic 来定义并验证传入请求的数据。FastAPI 支持多种参数来源：路径参数、查询参数、请求体、表单数据、Header、Cookie、File 等，而模型通常用于请求体、查询参数和表单数据的结构化表示。

import uvicorn
from fastapi import FastAPI, Request, Form
from fastapi.params import Depends
from pydantic import BaseModel
from watchfiles import awatchapp = FastAPI(title="测试小Demo",description="文档描述",version="1.0",
)class QueryParam(BaseModel):""" 定义参数校验模型;"""q: str | None = Nonelimit: int = 10offset: int = 0def get_query_params(params: QueryParam = Depends()):print(params)return params@app.get("/query", tags=["query"])
async def query(params: QueryParam = Depends(get_query_params)):return {"params": params}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

接口上传的参数必须与模型限定的模型相同；

fastapi 会根据自己参数形式推断是 body 还是 query 形式的参数；

模型的嵌套使用

class Address(BaseModel):city: strstreet: strzipcode: strclass User(BaseModel):username: stremail: straddress: Address  # 嵌套模型@app.post("/users/")
def create_user(user: User):return {"username": user.username,"city": user.address.city,"zipcode": user.address.zipcode}

2.3 接收多个请求参数

Fastapi 可以接收多个模型的参数，使用 fastapi 中的 Body 类可以进行解析。

from fastapi import Bodyclass Item(BaseModel):name: strprice: floatclass User(BaseModel):username: stremail: str@app.post("/items/")
def create_item(item: Item = Body(...), user: User = Body(...)):return {"item": item, "user": user}

基础模型的参数也可以由 Body 类的进行解析

from fastapi import Body@app.post("/params/")
def receive_data(name: str = Body(...), age: int = Body(...)):return {"name": name, "age": age}

2.4 Filed 的使用

在 FastAPI 中，请求体字段主要用来描述请求体的结构、添加默认值、校验规则、描述信息、示例等。

from fastapi import FastAPI
from pydantic import BaseModel, Filedapp = FastAPI()class Item(BaseModel):name: str = Field(..., title="商品名称", min_length=1, max_length=100)description: str | None = Field(None, title="描述", max_length=300)price: float = Field(..., gt=0, description="商品价格，必须大于0")tax: float | None = Field(default=0.0, ge=0)@app.post("/items/")
def create_item(item: Item):return item

2.5 指定响应

在 FastAPI 中，响应模型是一个强大的功能，它允许开发者定义和限制返回客户端的数据结构，通过在路径操作装饰器中使用response_model参数，可以指定用于响应的模型，这有助于确保输出数据的质量和安全性。

响应模型可以是一个 Pydantic 模型，也可以是一个由 Pydantic 模型组成的列表，例如List[Item]。FastAPI 将使用这个模型来转换输出数据、进行数据校验、为响应添加JSON Schema，并在自动生成的文档系统中使用。

""" 指定响应模型;
"""
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: list[str] = []@app.post("/items/", response_model=Item)
async def create_item(item: Item):return item

无论路径操作函数返回的是一个字典、数据库对象还是其他模型，FastAPI都会使用response_model来执行字段约束和序列化。

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : learn_fast.py
@Time    : 2025/5/31 14:29
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 学习 FastAPI 的中的依赖注入的知识;
"""import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Request, APIRouter, Cookie
from pydantic import BaseModelapp = FastAPI()class ItemModel(BaseModel):title: strdescription: str@app.get("/index", response_model=ItemModel)
async def index():return {"title": "傻狗", "description": "傻狗还在加班.....", "code": 200}  # code 不在模型中,会直接被屏蔽掉if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

使用响应模型的一个关键优点是它会限制输出数据，只包含模型定义内的数据。这对于安全性至关重要，尤其是在处理敏感信息时。例如，你可能不希望在用户注册后将密码返回给客户端。在这种情况下，你可以创建一个包含密码的输入模型UserIn和一个不包含密码的输出模型UserOut：

from fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserIn(BaseModel):username: strpassword: stremail: EmailStrfull_name: str | None = Noneclass UserOut(BaseModel):username: stremail: EmailStrfull_name: str | None = None@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):return user

在这个例子中，即使路径操作函数返回包含密码的UserIn对象，FastAPI也会负责过滤掉未在UserOut输出模型中声明的所有数据。

在文档中查看响应模型

当你查看自动化文档时，你可以检查输入模型和输出模型是否都具有自己的JSON Schema，并且两种模型都将在交互式API文档中使用。你的响应模型可以具有默认值，但你可能希望从结果中忽略这些默认值。例如，当你在NoSQL数据库中保存了具有许多可选属性的模型，但你又不想发送充满默认值的很长的JSON响应。在这种情况下，你可以使用response_model_exclude_unset参数来仅返回显式设定的值。
```
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float = 10.5tags: list[str] = []@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):return items[item_id]
```
总结

使用路径操作装饰器的response_model参数来定义响应模型，特别是确保私有数据被过滤掉。使用response_model_exclude_unset来仅返回显式设定的值。这些实践有助于提高API的安全性和数据的准确性。

3. 中间件

在 FastAPI 中 Request 请求信息将依次经过 middleware、router 等组件进行处理，但request所包含的信息修改后在 scope 中是不可改变的，这意味着上一环节处理后不能直接通过修改相应的值从而来传递值。

**定义：**中间件是介于接收请求和发送响应之间的一个软件层，在 FastAPI 应用中。所有的请求首先经过一系列的中间件，然后才到达实际的业务逻辑处理函数；响应也会在返回客户之前经过这些中间件，因此中间件是处里全局任务的理想选择；

**常见的中间件：**跨域中间件、登录校验中间件、日志中间件等常见的类型。

3.1 跨域中间件

通过导入 FastApi 中的中间件，可以直接让后端满足跨域的请求。跨源资源共享（CORS）是一个安全特性，允许或限制网页上的资源可以被另一个域名的网页访问。在 API 开发中配置 CORS 是常见需求，尤其是当你的前端和后端部署在不同的域时。

常见的跨域方式：

配置文件中设置请求头
通过中间件进行配置
通过 Nginx 带来，进行请求的添加进行跨域的请求。、

# !/usr/bin/env python
# -*-coding:utf-8 -*-
import uvicorn
from fastapi import FastAPI, Request, Form, HTTPException
from fastapi.params import Depends, Query
from pydantic import BaseModel
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import defer
from fastapi.middleware.cors import CORSMiddleware
from watchfiles import awatchapp = FastAPI(title="测试小Demo",description="文档描述",version="1.0",
)# 使用中间件配置跨域的请求, 将 CORS 中间件添加到应用中。这一步是配置 CORS 支持的关键。
app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 可以是一个具体的域名列表或 ["*"]（允许所有域）。allow_credentials=True,  # 允许前端在请求中携带认证信息（如 Cookies）。这是一个布尔值。allow_methods=["GET", "POST", "PUT", "DELETE"],  # 指定允许的 HTTP 方法，如["GET", "POST", "PUT", "DELETE"]。allow_headers=["X-Custom-Header", "Content-Type"],  # 允许的 HTTP 请求头列表，这里允许 X-Custom-Header 和 Content-Type
)@app.get("/")
async def main():return {"message": "Hello World"}@app.post("/submit")
async def submit_form(data: dict):return {"received_data": data}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

配置中间的核心的步骤就是add_middleware()函数的配置；

# 使用中间件配置跨域的请求, 将 CORS 中间件添加到应用中。这一步是配置 CORS 支持的关键。
app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 可以是一个具体的域名列表或 ["*"]（允许所有域）。allow_credentials=True,  # 允许前端在请求中携带认证信息（如 Cookies）。这是一个布尔值。allow_methods=["GET", "POST", "PUT", "DELETE"],  # 指定允许的 HTTP 方法，如["GET", "POST", "PUT", "DELETE"]。allow_headers=["X-Custom-Header", "Content-Type"],  # 允许的 HTTP 请求头列表，这里允许 X-Custom-Header 和 Content-Type
)

3.2 自定义中间

中间件可以用来实例日志记录等功能，自定义中间件的时候可以通过使用第三方日志插件或者自定义的日志系统。

from starlette.middleware.base import BaseHTTPMiddlewareclass LoggingMiddleware(BaseHTTPMiddleware):"""自定义日志的中间件;1. 自定义中间件的时候, 需要继承 BaseHTTPMiddleware;"""pass

通过源码看到，我们必须要实现函数的dispatch()方法；

中间件类 LoggingMiddleware:
继承自 BaseHTTPMiddleware，重写 dispatch 方法来拦截请求和发送的响应。
使用 call_next(request) 异步调用接下来的请求处理，这将返回一个响应对象。
计算处理请求所需的时间，并构建一个包含请求方法、路径、响应状态和处理时间的日志消息。
使用 print 函数打印日志信息。在实际应用中，选择使用logging 模块或外部日志服务。

# !/usr/bin/env python
# -*-coding:utf-8 -*-import os
import time
from http.client import responsesimport uvicorn
from _pytest.nodes import Item
from fastapi import FastAPI, Request, Form, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from fastapi.responses import JSONResponse, PlainTextResponse
from starlette.responses import Responseclass LoggingMiddleware(BaseHTTPMiddleware):"""自定义日志的中间件;1. 自定义中间件的时候, 需要继承 BaseHTTPMiddleware;"""async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:"""重写父类的抽象方法, 添加上相关的日志处理逻辑;:param request::param call_next::return:"""start_time = time.time()  # 获取请求处理的信息;response = await call_next(request)  # 此处会调用下一个中间件;process_time = time.time() - start_time# 记录请求和响应的详细信息log_message = (f"请求方法: {request.method}\n"f"请求路径: {request.url.path}\n"f"响应状态: {response.status_code}\n"f"处理时间: {process_time:.4f} 秒")print(log_message)return responseapp = FastAPI(title="测试小Demo",description="文档描述",version="1.0",
)# 添加我们自定义的中间件;
app.add_middleware(LoggingMiddleware)# 使用 PlainTextResponse 返回简单文本响应
@app.get("/", response_class=PlainTextResponse)
async def read_root():# 可以直接返回字符串，FastAPI 会自动封装为 PlainTextResponsereturn "Hello, World!"# 使用 JSONResponse 创建一个更复杂的 JSON 响应
@app.post("/items/", response_class=JSONResponse)
async def create_item(item: dict):if "name" not in item:raise HTTPException(status_code=400, detail="Missing name filed in the item")response_data = {"item_name": item.get("name"), "item_id": 123}return JSONResponse(content=response_data, status_code=201)if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

上述的两个函数都会经历中间件类的处理；在正规项目的日志处理中通常会采取更加完备日志插件，比如 kafka 、RabbitMQ 等组件来收集日志的信息；

3.3 登录中间件

登录存在前后端分离和前后端不分离的模式，分离的时候使用的是 JWT 的校验模式，在不分离的部分使用的是 Session 和 Cookie 的形式进行的处理；

FastAPI 使用 SessionMiddleware 来管理用户会话，这对于需要追踪用户状态或者保持登录状态的应用尤为重要。

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from starlette.middleware.sessions import SessionMiddleware
import uvicorn
import osapp = FastAPI()# 设置 SessionMiddleware，secret_key 应该是一个长随机字符串
app.add_middleware(SessionMiddleware, secret_key="!se1cret2-ke3y-sh4ould-b5e-ve8ry-se8cure!"
)@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):# 访问会话中的数据count = request.session.get("count", 0)count += 1request.session["count"] = count  # 更新会话数据return f"<html><body><h1>Visit Count: {count}</h1></body></html>"@app.get("/reset")
async def reset_count(request: Request):request.session.pop("count", None)  # 重置会话数据return {"status": "session reset"}@app.get("/logout")
async def logout(request: Request):request.session.clear()  # 清空所有会话数据return {"status": "logged out"}if __name__ == "__main__":uvicorn.run(f"{os.path.basename(__file__).split('.')[0]}:app",host="127.0.0.1",port=8000,reload=True,)

上述的中间的核心功能是

from starlette.middleware.sessions import SessionMiddleware  # 导入框架中的 中间件# 添加中间件的时候，需要自定义 secret_key 的关键秘钥;
app.add_middleware(SessionMiddleware, secret_key="secretsfsadfasdfsadf@#@!%$$@!#%$!")

前后端分离的登录校验主要是通过 JWT 进行校验的登录，不在此处进行描述，在后续的部分进行描述；

补充：

中间件TrustedHostMiddleware的应用是用来提高应用的安全性，限制哪些主机名可以访问应用。

from fastapi import FastAPI
from starlette.middleware.trustedhost import TrustedHostMiddlewareapp = FastAPI()# 添加 TrustedHostMiddleware
app.add_middleware(TrustedHostMiddleware,allowed_hosts=["example.com", "www.example.com", "*.example.com"]  # 限制访问的主机名称;
)@app.get("/")
async def read_root():return {"message": "Hello, World!"}

TrustedHostMiddleware 用于确保接收到的请求来自预定义的一组受信任的主机名。它的主要作用是防止 HTTP Host 头攻击，这种攻击可能会误导应用生成错误的链接、重定向、或被认为是不安全的请求来源。
当一个请求被接收时，TrustedHostMiddleware 检查 HTTP 请求的 Host 头部是否在允许的主机名列表中。如果不在这个列表中，请求将被拒绝，并返回一个错误响应（通常是 400 Bad Request）。
适用于所有服务器端应用，特别是在面对公共互联网时，这有助于确保请求确实来自你预期的域。

4.依赖注入

FastAPI 提供了一种简化版的依赖注入系统，允许在 FastAPI 程序运行时将外部组件（依赖项、包括函数、类、对象）注入到程序内部的函数或者类中，而不是直接在函数内部创建管理他们。

区别于传统的依赖注入模式，FastAPI 的依赖注入采用的是函数式依赖注入，基于 Python 的类型提示和Depends。他没有传统的 IOC 容器，而是通过函数或类直接声明依赖，框架在运行时自动解析和注入。

4.1 注入方式

FastAPI 允许在路由函数、路由组、以及应用全局注入依赖项。当我们需要每个路由中可以根据 token 获取当前用户的信息的时候，可以采用如下的方式；

路由函数的注入形式

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : learn_fast.py
@Time    : 2025/5/31 14:29
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 学习 FastAPI 的中的依赖注入的知识;
"""
import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Requestapp = FastAPI()async def get_current_user(token: str = None):"""定义注入函数,实现用户的信息获取;Args:token:Returns:"""if token:return {"username": "Hello world"}else:raise HTTPException(status_code=401, detail="用户校验失败")@app.get("/user")
async def index(request: Request):# 当我们不使用注入但是想要校验的时候就只能在函数的内部调用函数q = request.query_params.get("token")user = await get_current_user(q)  # 手动调用依赖项;return {"hello": "world"}@app.get("/home")
async def home(user=Depends(get_current_user)):  # 此处使用是路由函数的注入形式;return {"message": "Hello world"}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

上述的代码采用了视图函数的注入方式，只有携带 token 参数才能够通过依赖项的校验；

路由路径注入

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : learn_fast.py
@Time    : 2025/5/31 14:29
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 学习 FastAPI 的中的依赖注入的知识;
"""
import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Requestapp = FastAPI()async def get_current_user(token: str = None):"""定义注入函数,实现用户的信息获取;Args:token:Returns:"""if token:return {"username": "Hello world"}else:raise HTTPException(status_code=401, detail="用户校验失败")async def check_login(token: str = None):"""使用依赖注入的形式进行注入的校验;Args:token:Returns:"""print("开始进行逻辑的校验......")if not token:raise HTTPException(status_code=401, detail="用户还没有登录")@app.get("/user")
async def index(request: Request):# 当我们不使用注入但是想要校验的时候就只能在函数的内部调用函数q = request.query_params.get("token")user = await get_current_user(q)  # 手动调用依赖项;return {"hello": "world"}@app.get("/home")
async def home(user=Depends(get_current_user)):  # 此处使用是路由函数的注入形式;return {"message": "Hello world"}@app.get("/index", dependencies=[Depends(check_login)])
async def index(request: Request):print("你好世界")return {"message": "Hello world"}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

我们在新的函数的路径中加上指定的依赖项的时候，就会通过校验的信息；

当我们携带 token 的时候就可以通过在路径中注入的函数校验；

路由组的注册

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : user_api.py
@Time    : 2025/5/31 15:19
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 在新的文件中编写指定部分的视图函数,完成视图函数的分发;
"""
from fastapi import APIRouter, Request, HTTPException, Dependsasync def check_login(token: str = None):"""使用依赖注入的形式进行注入的校验;Args:token:Returns:"""print("开始进行逻辑的校验......")if not token:raise HTTPException(status_code=401, detail="用户还没有登录")# 此处使用了路由组的注入, 这个表明此组路由的情况都会完成注册;
user_api = APIRouter(prefix="/users", tags=["users"], dependencies=[Depends(check_login)])@user_api.get("/index")
async def index(request: Request):print("你好世界")return {"message": "Hello world"}

全局注入

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : learn_fast.py
@Time    : 2025/5/31 14:29
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 学习 FastAPI 的中的依赖注入的知识;
"""
import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Request, APIRouterfrom learn_web.user_api import user_apiasync def get_current_user(token: str = None):"""定义注入函数,实现用户的信息获取;Args:token:Returns:"""if token:return {"username": "Hello world"}else:raise HTTPException(status_code=401, detail="用户校验失败")async def check_login(token: str = None):"""使用依赖注入的形式进行注入的校验;Args:token:Returns:"""print("开始进行逻辑的校验......")if not token:raise HTTPException(status_code=401, detail="用户还没有登录")app = FastAPI(dependencies=[Depends(get_current_user), Depends(check_login)])  # 此处进行全局的注册;
"""
dependencies=[Depends(get_current_user), Depends(check_login)]
列表中可以存在多个校验的方式;
"""# app.include_router(user_api)@app.get("/home")
def home():return {"message": "Hello world"}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

在使用全局依赖的时候我们会发现，它的效果与中间件非常相似；

""" 使用中间件来完成全局的校验;
"""
from fastapi import FastAPI
from dependence import *app = FastAPI()@app.middleware("http")  
async def check_login_middleware(request: Request, call_next):  token = request.query_params.get("token")  if not token:  raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="未登录")  else:  return await call_next(request)  @app.get("/")  
async def hello_world(req: Request):  user = await get_current_user(req.query_params.get("token"))  return {"user": user, "message": "Hello World!"}

在我们选择使用那种方式的时候，需要有一定的取舍原则;

**中间件：**会在所有请求和响应中执行相同的逻辑，更加适合全局的、横切面的任务，如日志、监控、错误的处理；
**依赖注入：**可以由针对地不同的路由分组、路由执行相同的逻辑，更适用于局部的、路由级别的任务，如数据库操作、业务逻辑封装、用户验证、以及一些变化可能性较高的功能逻辑；

4.2 依赖覆盖

在使用全局依赖项时，可以通过 FastAPI 实例的 dependency_overrides 来覆盖原来的依赖项设定，在上例中我们现在希望通过用户名和密码来检验用户时，可以这样做：

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : learn_fast.py
@Time    : 2025/5/31 14:29
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 学习 FastAPI 的中的依赖注入的知识;
"""
import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Request, APIRouterfrom learn_web.user_api import user_apiasync def get_current_user(token: str = None):"""定义注入函数,实现用户的信息获取;Args:token:Returns:"""print("我被覆盖了, 不会打印出来")if token:return {"username": "Hello world"}else:raise HTTPException(status_code=401, detail="用户校验失败")async def check_login(token: str = None):"""使用依赖注入的形式进行注入的校验;Args:token:Returns:"""print("开始进行逻辑的校验......")if not token:raise HTTPException(status_code=401, detail="用户还没有登录")app = FastAPI(dependencies=[Depends(get_current_user)])  # 此处进行全局的注册;app.dependency_overrides[get_current_user] = check_login  # 覆盖原来的依赖项# app.include_router(user_api)@app.get("/home")
def home():return {"message": "Hello world"}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

我们可以看到此时会直接进行覆盖，不在执行原来的全局注入；

4.3 参数化依赖

我们常见的用户逻辑不会只校验 token 还有很多的参数形式，以登录常用的用户名和密码为例；

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : learn_fast.py
@Time    : 2025/5/31 14:29
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 学习 FastAPI 的中的依赖注入的知识;
"""
import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Request, APIRouter
from sqlalchemy.orm import dependency
from watchfiles import awatchfrom learn_web.user_api import user_apiasync def get_user(user: str = None, password: str = None):"""用于通过用户和密码来获取用户的信息, 此处直接返回, 业务逻辑中应该加入对应的处理逻辑;Args:user:password:Returns:"""if user and password:return {"username": user, "password": password}def get_current_user(check_type: str = "token"):"""定义注入函数,实现用户的信息获取;Args:check_type:Returns:"""async def dependency(request: Request):"""使用闭包的形式, 对外部隐藏一下参数; 协程使用闭包的时候需要再前面加上 async 进行使用Args:request:Returns:"""print("这里执行了....")if check_type == "token":# 存在token 的时候token = request.query_params.get("token")await check_login(token)print({"token": token})return {"token": token}else:# 使用用户user = request.query_params.get("user")passwd = request.query_params.get("password")rsp = await get_user(user, passwd)if rsp:return rspelse:raise HTTPException(status_code=401, detail="校验失败。。。")return dependencyasync def check_login(token: str = None):"""使用依赖注入的形式进行注入的校验;Args:token:Returns:"""print("开始进行逻辑的校验......")if not token:raise HTTPException(status_code=401, detail="用户还没有登录")app = FastAPI()# app.include_router(user_api)@app.get("/home")
def home(user = Depends(get_current_user('user'))):  # 设置指定的是用户信息参数校验. 指定单个函数的校验;return {"message": "Hello world"}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

4.4 生成器依赖

yield生成器依赖特别适用于需要设置和清理操作的场景，如数据库会话管理:

from fastapi import Depends
from sqlalchemy.orm import Sessiondef get_db():db = SessionLocal()try:yield dbfinally:db.close()@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):user = db.query(User).filter(User.id == user_id).first()return user

但是数据库的操作在大型项目中不太会根据这种方式进行;

4.5 匿名函数的注入

匿名函数和常规函数的注入基本无差别;

from fastapi import FastAPI, Depends
from dependence import *app = FastAPI()  current_user = {"username": "faker"}  
@app.get("/")  
async def hello_world(user=Depends(lambda: current_user)):  # 使用简单函数的注入;return {"user": user, "message": "Hello World!"}

4.6 类的注入

官方的示例中给出的

from fastapi import Depends, FastAPIapp = FastAPI()fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]class CommonQueryParams:def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):self.q = qself.skip = skipself.limit = limit@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):response = {}if commons.q:response.update({"q": commons.q})items = fake_items_db[commons.skip : commons.skip + commons.limit]response.update({"items": items})return response

Python 类也是 可调用对象。因此，在 FastAPI 中，你可以使用一个 Python 类作为一个依赖项。实际上 FastAPI 检查的是它是一个 “可调用对象”（函数，类或其他任何类型）以及定义的参数。如果您在 FastAPI 中传递一个 “可调用对象” 作为依赖项，它将分析该 “可调用对象” 的参数，并以处理路径操作函数的参数的方式来处理它们。包括子依赖项。这也适用于完全没有参数的可调用对象。这与不带参数的路径操作函数一样。所以，我们可以将上面的依赖项 “可依赖对象” common_parameters 更改为类 CommonQueryParams

4.7 嵌套使用

多个依赖函数之间可以进行嵌套；

import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Request, APIRouter, Cookieapp = FastAPI()def query_extractor(q: str | None = None):"""基础依赖项Args:q:Returns:"""return q# 高级依赖项，依赖于基础依赖项
def query_or_cookie_extractor(q: str = Depends(query_extractor), last_query: str | None = Cookie(default=None)):if not q:return last_queryreturn q# 路由处理函数，依赖于高级依赖项
@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):return {"q_or_cookie": query_or_default}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

嵌套的时候可以使用类嵌套在函数内部过程类似；

4.8 函数原理探索

在 FastAPI 中，Depends 是一个核心函数，用于实现依赖注入，主要依赖 Python 的类型提示和函数调用动态解析;

依赖解析器

FastAPI 通过 Depends() 标记依赖，并在请求处理时识别它。使用 Python 的类型提示 和 inspect 模块 来分析依赖项的函数签名。

"""原理如下
"""
from inspect import signaturedef example_dependency(name: str):return f"Hello {name}"print(signature(example_dependency))
# 输出: (name: str)

执行依赖项

FastAPI 使用 依赖注入解析器 来执行依赖项。它会：判断依赖项是否是一个协程（async）或普通函数。提取参数值并传入依赖项。解析 yield 语句管理依赖项的生命周期。
```
import asyncioasync def async_dependency():await asyncio.sleep(1)return "Async Dependency Finished"async def main():result = await async_dependency()print(result)asyncio.run(main())
```
如果依赖项是异步的，FastAPI 会使用 await 进行处理。

依赖嵌套与上下文管理

FastAPI 支持依赖项中调用其他依赖项（依赖嵌套）。通过上下文管理（yield）处理资源清理。

from contextlib import asynccontextmanager@asynccontextmanager
async def db_session():print("Connecting to database...")yield {"db": "Connected"}print("Closing database connection...")async def get_db(session=Depends(db_session)):print(session)

使用 yield 提供依赖项，并在请求结束后执行清理任务。

FastAPI 的依赖注入解析过程
```
@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):user = db.query(User).filter(User.id == user_id).first()return user
```
- 扫描路由函数：检测 Depends() 标记的依赖项。
- 解析依赖关系：识别所有需要的依赖项，包括嵌套依赖。
- 执行依赖项：根据依赖项的函数类型（同步或异步）执行它。
- 注入结果：将执行结果传递给视图函数。
- 生命周期管理：如果依赖项使用 yield，则在响应发送后执行清理逻辑。

5. 结合数据库

通过使用 pydantic 和 sqlalchemy 进行使用。

5.1 数据库操作

# 安装组件
pip install sqlalchemy

sqlalchemy 与数据库的操作部分，与其他框架中的应用没有具体的区别；本次案例中不使用 MySQL 先使用 sqllite 进行简单测试

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : models.py
@Time    : 2025/5/31 17:47
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 
"""
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, create_engine
from sqlalchemy.orm import relationship, declarative_baseBase = declarative_base()# 定义 User 类
class User(Base):__tablename__ = 'users'  # 定义表名id = Column(Integer, primary_key=True, index=True)email = Column(String(255), unique=True, index=True)hashed_password = Column(String(255))is_active = Column(Boolean, default=True)items = relationship("Item", back_populates="owner")  # 关联 Item 表# 定义 Item 类
class Item(Base):__tablename__ = "items"id = Column(Integer, primary_key=True, index=True)title = Column(String(255), index=True)description = Column(String(255), index=True)owner_id = Column(Integer, ForeignKey('users.id'))owner = relationship("User", back_populates="items")  # 关联 User 表if __name__ == '__main__':# 创建数据库链接db = "fastapp"engine = create_engine(f'sqlite:///{db}.db')Base.metadata.drop_all(engine)Base.metadata.create_all(engine)

5.2 Pydantic 操作

通常会创建schemas.py文件中，在文件内部定义 Pydantic 模型；

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : schemas.py
@Time    : 2025/5/31 18:19
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 定义与数据库操作相对应的 pydantic 模型;
"""
from typing import List
from pydantic import BaseModelclass ItemBase(BaseModel):id: inttitle: strdescription: strclass Item(ItemBase):id: intowner_id: intclass Config:orm_mode = Trueclass UserBase(BaseModel):email: strclass UserCreate(UserBase):password: strclass User(BaseModel):id: int# title: str# description: stris_active: boolitems: List[Item] = []

5.3 数据库操作

这部分通常放在crud.py，在其中声明数据库的操作方法；

from sqlalchemy.orm import Sessionfrom . import models, schemasdef get_user(db: Session, user_id: int):"""根据id获取用户信息:param db: 数据库会话:param user_id: 用户id:return: 用户信息"""return db.query(models.User).filter(models.User.id == user_id).first()def get_user_by_email(db: Session, email: str):"""根据email获取用户信息:param db: 数据库会话:param email: 用户email:return: 用户信息"""return db.query(models.User).filter(models.User.email == email).first()def get_users(db: Session, skip: int = 0, limit: int = 100):"""获取特定数量的用户:param db: 数据库会话:param skip: 开始位置:param limit: 限制数量:return: 用户信息列表"""return db.query(models.User).offset(skip).limit(limit).all()def create_user(db: Session, user: schemas.UserCreate):"""创建用户:param db: 数据库会话:param user: 用户模型:return: 根据email和password登录的用户信息"""fake_hashed_password = user.password + "notreallyhashed"db_user = models.User(email=user.email, hashed_password=fake_hashed_password)db.add(db_user)      # 添加到会话db.commit()          # 提交到数据库db.refresh(db_user)  # 刷新数据库return db_userdef get_items(db: Session, skip: int = 0, limit: int = 100):"""获取指定数量的item:param db: 数据库会话:param skip: 开始位置:param limit: 限制数量:return: item列表"""return db.query(models.Item).offset(skip).limit(limit).all()def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):"""创建用户item:param db: 数据库会话:param item: Item对象:param user_id: 用户id:return: Item模型对象"""db_item = models.Item(**item.dict(), owner_id=user_id)db.add(db_item)db.commit()db.refresh(db_item)return db_item

这部分的代码通常需要自己封装的类；

5.4 结合使用

在视图函数中使用；

from typing import Listfrom fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Sessionfrom . import crud, models, schemas
from .database import SessionLocal, enginemodels.Base.metadata.create_all(bind=engine)app = FastAPI()# 依赖
def get_db():try:db = SessionLocal()yield dbfinally:db.close()@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):# 根据email查找用户db_user = crud.get_user_by_email(db, email=user.email)# 如果用户存在，提示该邮箱已经被注册if db_user:raise HTTPException(status_code=400, detail="Email already registered")# 返回创建的user对象return crud.create_user(db=db, user=user)@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):# 读取指定数量用户users = crud.get_users(db, skip=skip, limit=limit)return users@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):# 获取当前id的用户信息db_user = crud.get_user(db, user_id=user_id)# 如果没有信息，提示用户不存在if db_user is None:raise HTTPException(status_code=404, detail="User not found")return db_user@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):# 创建该用户的itemsreturn crud.create_user_item(db=db, item=item, user_id=user_id)@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):# 获取所有itemsitems = crud.get_items(db, skip=skip, limit=limit)return items

通常是数据库返回的对象，直接通过模型过滤掉不返回的信息；

6. 文档

FastAPI 开发的后端接口会直接根据自己定义好的或者开发好的接口形成 API 文档；

**个人观点：**虽然可以自动生成 API 文档，但是这并不意味这在系统开发之初的时候可以省略 API 文档，在设计阶段应该还是要有完善的 API 文档的设计；

默认地址：http:127.0.0.1:8000/docs

FastAPI是流行的 Python web 框架，适用于开发高吞吐量API和微服务（直接支持异步编程）。FastAPI的优势之一：通过提供高级抽象和自动数据模型转换，简化请求数据的处理（用户不需要手动处理原始请求数据），并能根据路由和 Pydantic 模型自动生成 OpenAPI 接口文档。

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : docs.py
@Time    : 2025/5/31 20:39
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 文档 app 测试;
"""
import uuid
import uvicorn
from typing import Any, Union, Optional
from typing_extensions import Literal
from fastapi import Body, FastAPI
from pydantic import (BaseModel,Field,UUID4
)app = FastAPI()class UserIn(BaseModel):channel: Literal[0, 1] = Field(0, title="渠道")username: str = Field(..., title="用户名")password: str = Field(..., title="用户密码", description="长度6-8位")email: str = Field(..., title="用户邮箱地址")full_name: str = Field(None, title="用户全名")request_id: Optional[UUID4]class UserOut(BaseModel):username: str = Field(..., title="用户名")email: str = Field(..., title="用户邮箱地址")full_name: str = Field(None, title="用户全名")request_id: Optional[UUID4]# FastAPI will take care of filtering out all the data that is not declared in the output model (using Pydantic).
# 因此，FastAPI将负责过滤掉输出模型中未声明的所有数据（使用Pydantic）。@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn = Body(examples={"example1": {"summary": "A short summary or description of the example","value": {# example data here"channel": 0,"username": "Foo","password": "33759","email": "chencare@163.com","full_name": "xiaotao"}}# 会将示例的数据显示在文档上。})
) -> UserOut:user.request_id = uuid.uuid4()print(user.request_id)return userif __name__ == '__main__':uvicorn.run(app=app, access_log=True, port=8000)

ReDoc vs. Swagger UI

ReDoc更美观，Swagger UI更注重交互（用户直接从界面中发送请求，查看响应，这对于测试和调试 API 非常有用。）

7. 静态文件

如果使用前后台不分离的开发方式，那么模板文件中使用的静态文件，比如css/js等文件的目录需要在后台进行配置，以便模板渲染是能正确读到这些静态文件。

使用 StaticFiles

# 文件目录结果;
│—main.py
│
├─static
│  │  bootstrap.css
│  │  bootstrap.js
│  │
│  └─fonts
│          glyphicons-halflings-regular.eot
│          glyphicons-halflings-regular.svg
│          glyphicons-halflings-regular.ttf
│          glyphicons-halflings-regular.woff
│          glyphicons-halflings-regular.woff2
│
└─templateshome.html

在main.py中直接挂载

import uvicorn
from fastapi.staticfiles import StaticFiles
from fastapi import FastAPI, Depends, HTTPException, Request, APIRouter, Cookie
from pydantic import BaseModelapp = FastAPI()class ItemModel(BaseModel):title: strdescription: strapp.mount("/static", StaticFiles(directory="static"), name="static")@app.get("/index", response_model=ItemModel)
async def index():return {"title": "傻狗", "description": "傻狗还在加班.....", "code": 200}if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)

通过 mount 将 StaticFiles 实例挂载到一个特定的路径上。其中 StaticFile 类中传递的directory参数是项目中静态文件的目录。

所有子应用的子路径将会挂载到实例上。所以任何以“/static”开头的路径资源都可以访问到。

# Request URL: http://127.0.0.1:8000/static/bootstrap.css
# 给一个名称，用于FastAPI内部调用，所以在home.html中可以使用如下url_for的调用方式引入静态文件：
# <link rel="stylesheet" href="{{ url_for('static',path='/bootstrap.css') }}">

8. 测试

# 安装
pip install httpx

它基于 HTTPX，而HTTPX又是基于Requests设计的，所以很相似且易懂。有了它，你可以直接与FastAPI一起使用 pytest。要使用TestClient通过传入 FastAPI 应用创建一个 TestClient。创建名字以test_开头函数，标准以pytest的标准相同；使用 httpx 那样使用 TestClient 对象。为你需要检查的地方用标准的Python表达式写个简单的 assert 语句（重申，标准的pytest）。

8.1 简单测试

# !/usr/bin/env python
# -*-coding:utf-8 -*-"""
@File    : test_app.py
@Time    : 2025/5/31 21:18
@Author  : zi qing bao jian
@Version : 1.0
@Desc    : 对 FastAPI 开发的接口进行单元测试;
"""from fastapi.testclient import TestClientfrom learn_fast import appclient = TestClient(app)def test_index():response = client.get("/index")assert response.status_code == 200assert response.json() == {}

8.2 分离测试

在实际应用中，你可能会把你的测试放在另一个文件里。您的FastAPI应用程序也可能由一些文件模块组成等等。

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py  # 此脚本中存放,单元测试的信息;

扩展测试文件的示例，在官网的文档上；

from fastapi.testclient import TestClientfrom .main import appclient = TestClient(app)def test_read_item():response = client.get("/items/foo", headers={"X-Token": "coneofsilence"})assert response.status_code == 200assert response.json() == {"id": "foo","title": "Foo","description": "There goes my hero",}def test_read_item_bad_token():response = client.get("/items/foo", headers={"X-Token": "hailhydra"})assert response.status_code == 400assert response.json() == {"detail": "Invalid X-Token header"}def test_read_nonexistent_item():response = client.get("/items/baz", headers={"X-Token": "coneofsilence"})assert response.status_code == 404assert response.json() == {"detail": "Item not found"}def test_create_item():response = client.post("/items/",headers={"X-Token": "coneofsilence"},json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},)assert response.status_code == 200assert response.json() == {"id": "foobar","title": "Foo Bar","description": "The Foo Barters",}def test_create_item_bad_token():response = client.post("/items/",headers={"X-Token": "hailhydra"},json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"},)assert response.status_code == 400assert response.json() == {"detail": "Invalid X-Token header"}def test_create_existing_item():response = client.post("/items/",headers={"X-Token": "coneofsilence"},json={"id": "foo","title": "The Foo ID Stealers","description": "There goes my stealer",},)assert response.status_code == 409assert response.json() == {"detail": "Item already exists"}

每当你需要客户端在请求中传递信息，但你不知道如何传递时，你可以通过搜索（谷歌）如何用 httpx做，或者是用 requests 做，毕竟 HTTPX 的设计是基于 Requests 的设计的。

传一个路径或查询参数，添加到URL上。
传一个JSON体，传一个Python对象(例如一个dict)到参数 json。
如果你需要发送 Form Data 而不是 JSON，使用 data 参数。
要发送 headers，传 dict 给 headers 参数。
对于 cookies，传 dict 给 cookies 参数。

注意： TestClient 接收可以被转化为JSON的数据，而不是Pydantic模型。

9. 部署

对于Web API来说，通常涉及将上传到云服务器中，搭配一个性能和稳定性都不错的服务器程序，以便你的用户可以高效地访问你的应用程序，而不会出现中断或其他问题。这与开发阶段形成鲜明对比，在开发阶段，你不断更改代码、破坏代码、修复代码, 来回停止和重启服务器等。

部署的时候确定版本的 FastAPI 就不要去改变它的版本信息，以免兼容出错；

9.1 https

从开发人员的视角，在了解 HTTPS 时需要记住以下几点：

要使用 HTTPS，服务器需要拥有由第三方生成的证书(certificate)。

这些证书实际上是从第三方获取的，而不是“生成”的。
证书有生命周期

它们会过期。

然后它们需要更新，再次从第三方获取。
连接的加密发生在 TCP 层。

这是 HTTP 协议下面的一层。

因此，证书和加密处理是在 HTTP之前完成的。
TCP 不知道域名。仅仅知道 IP 地址。

有关所请求的 特定域名 的信息位于 HTTP 数据中。
HTTPS 证书证明某个域名，但协议和加密发生在 TCP 层，在知道正在处理哪个域名之前。
默认情况下，这意味着你每个 IP 地址只能拥有一个 HTTPS 证书

无论你的服务器有多大，或者服务器上的每个应用程序有多小。不过，对此有一个解决方案。
TLS 协议（在 HTTP 之下的TCP 层处理加密的协议）有一个扩展，称为 SNI。

SNI 扩展允许一台服务器（具有单个 IP 地址）拥有多个 HTTPS 证书并提供多个 HTTPS 域名/应用程序。
为此，服务器上会有单独的一个组件（程序）侦听公共 IP 地址，这个组件必须拥有服务器中的所有 HTTPS 证书。
获得安全连接后，通信协议仍然是HTTP。
内容是加密过的，即使它们是通过 HTTP 协议发送的。

9.2 asgi 服务器

让我们深入了解一些细节。

FastAPI 使用了一种用于构建 Python Web 框架和服务器的标准，称为 ASGI。FastAPI 本质上是一个 ASGI Web 框架。

要在远程服务器上运行 FastAPI 应用（或任何其他 ASGI 应用），您需要一个 ASGI 服务器程序，例如 Uvicorn。它是 fastapi 命令默认使用的 ASGI 服务器。

除此之外，还有其他一些可选的 ASGI 服务器，例如：

Uvicorn：高性能 ASGI 服务器。
Hypercorn：与 HTTP/2 和 Trio 等兼容的 ASGI 服务器。
Daphne：为 Django Channels 构建的 ASGI 服务器。
Granian：基于 Rust 的 HTTP 服务器，专为 Python 应用设计。
NGINX Unit：NGINX Unit 是一个轻量级且灵活的 Web 应用运行时环境。

9.3 多工作模式

部署应用程序时，您可能希望进行一些进程复制，以利用多核 CPU 并能够处理更多请求。正如您在上一章有关部署概念中看到的，您可以使用多种策略。使用 --workers 命令行选项来启动多个工作进程：

# 配置环境;
pip install -r requirements.tx
fastapi run --workers 4 main.py
# 这里唯一的新选项是 --workers 告诉 Uvicorn 启动 4 个工作进程。

9.4 容器化部署

# 选择镜像
FROM python:3.9# 
WORKDIR /code# 
COPY ./requirements.txt /code/requirements.txt# 
RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt# 
COPY ./app /code/app# 
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

在使用 docker 的命令进行部署运行

docker build -t fast:v1.0 .
docker run # 启动即可

9.5 Nginx 代理

使用 Nginx 代理的时候，可以根据自己端口的映射，以及与前端的配置均在 Nginx 中间进行配置；具体项目的配置文件不同，不在此处进行赘述；

学习到此处，我只觉得

坚持住！

继续努力，终成大器！