FastAPI

在 FastAPI 中，Request 对象和 FastAPI 应用实例 (app) 是核心概念，它们在应用状态管理和依赖注入中扮演着关键角色。 本文将介绍它们的关系、设计理念，以及如何利用 app.state 实现单例模式。 FastAPI 对象 FastAPI 对象是整个应用的核心实例： from fastapi import FastAPI app = FastAPI(title="示例应用") 核心职责 路由管理：通过 @app.get()、@app.post() 等装饰器定义 URL 到视图函数的映射。 中间件和事件管理：可注册中间件处理请求/响应，支持 startup 与 shutdown 事件。 应用状态管理：提供 app.state，可存放全局单例对象、数据库连接池、配置等。 异常处理与依赖注入：管理异常处理器，并协助依赖注入机制。 单例模式存储 这里要使用 app 的 State 对象存储单例，app 中定义如下 # app self.state: Annotated[ State, Doc( """ A state object for the application. This is the same object for the entire application, it doesn't change from request to request. You normally wouldn't use this in FastAPI, for most of the cases you would instead use FastAPI dependencies. This is simply inherited from Starlette. Read more about it in the [Starlette docs for Applications](https://www.starlette.dev/applications/#storing-state-on-the-app-instance). """ ), ] = State() State 源码如下，简单看就是一个字典 ...

Lifespan Events 生命周期事件 通过生命周期事件可以定义在应用开启之前需要执行的代码，这意味着这些代码会在开始接收外部请求之前被执行一次。 同样地，也可以定义应用在关闭的时候定义需要执行的代码，在尽力处理完所有请求后，该代码会被执行一次。 这对于设置需要在整个 app 的请求间共享的资源时非常有用，或者是需要进行清理工作的时候。 例如，一个数据库连接池，或者加载一个共享的机器学习模型。 Use Case 使用示例 下面通过一个例子说明如何使用。 假如你有一个机器学习模型，并且需要让其处理请求，由于请求都共享同一个模型，因此不是一个请求对应一个模型，或一个用户一个模型。 假设模型加载需要一定的时间，因为要从磁盘中读取大量的数据，因此不能每个请求都去加载一次。 你可以在顶层的模块文件中定义加载，但这意味着当进行简单的自动化测试的时候，也会加载该模型，这样就会很慢。 这就是需要解决的问题，需要在请求响应之前加载模型，也不是在代码被加载的时候加载模型。 Lifespan 生命周期 可以通过在 FastAPI app 中使用 lifespan 参数来定义启动和关闭逻辑，以及一个 “context manager” (上下文管理器)。 通过下面这种方法创建一个含 yield 的 function from contextlib import asynccontextmanager from fastapi impor FastAPI def fake_answer_to_everything_ml_model(x: float): return x * 42 ml_models = {} @asynccontextmanager async def lifespan(app: FastAPI): # Load the ML model ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model yield # Clean up the ML models and release the resources ml_models.clear() app = FastAPI(lifespan=lifespan) @app.get("/predict") async def predict(x: float): result = ml_models["answer_to_everything"](x) return {"result": result} 这里在生成器 yield 之前将模拟的昂贵函数放入机器学习字典中。 这段代码将在应用程序接收请求之前执行，即启动阶段。 ...

Background Tasks 后台任务 你可以定义一个在返回响应之后运行的后台任务。 这对请求之后执行一些操作十分有用，客户端无需一直等待操作任务完成再接收响应。 这包含一些例子： 执行操作后发送电子邮件 由于连接邮件服务器并发送邮件一般会比较“慢”（几秒钟），你可以立刻返回响应并在后台发送邮件请求。 处理数据 例如，你收到了一个文件需要缓慢处理，你可以返回一个 “Accepted” 响应 (HTTP 202) 并在后台处理文件。 Using Background Tasks 使用后台任务 首先要导入 BackgroundTasks 并在执行函数中定义一个路径参数，使用 BackgroundTasks 类型声明。 from fastapi import BackgroundTasks, FastAPI app = FastAPI() def write_notification(email: str, message=""): with open("log.txt", mode="w") as email_file: content = f"notification for {email}: {message}" email_file.write(content) @app.post("/send-notification/{email}") async def send_notification(email: str, backgroud_tasks: Background(Tasks): # Add parameter here background_tasks.add_task(write_notification, email, message="some notification") # add backgroud task here return {"message": "Notification sent in the background"} Create a task function 创建任务函数 创建一个函数放到后台运行，只是一个接收参数的基本函数，可以是 async def 或者就普通的 def 函数，FastAPI 会正确的处理它。 在这个例子中的任务函数将会编写文件，并且写入操作不使用 async 或 await 故使用 def 定义了一个基本的函数。 ...

Middleware 你可以添加中间件到 FastAPI 应用中。 “中间件” 是一个函数，它在每个请求被特定路径操作之前对其进行处理，同时在每个响应返回之前也对其进行处理。 在到达应用程序之前处理请求 可以在请求中做一些事情，或运行任何需要的代码 将处理后的请求传递给应用程序 之后处理应用程序返回的响应 可以对响应做一些事情，或运行任何需要的代码 然后返回响应 Create a Middleware 创建一个中间件 想要创建一个中间件，你可以在函数上面使用装饰器 @app.middleware("http")，该函数接受： request 请求 一个函数 call_next 并将会接收 request 作为一个参数 该函数会将 request 传递给对应的路径操作 然后返回对应路由操作生成的 response 你可以修改或者直接返回 response import time from fastapi import FastAPI, Request app = FastAPI() @app.middleware("http") async def add_process_time_header(request: Request, call_next): ... TIP 自定义专属 headers 可以使用 X-prefix 来添加。 但如果你有一个自定义的 header 并想要客户端能够看到这些信息，你需要使用 Starlette’s CORS docs 中的参数参数 expose_headers 将其加入你的 CORS 设置里 (CORS (Corss-Origin Resource Sharing))。 Before and after the response 在响应前后 你也可以在 request 前后运行代码，也可以在 response 前后运行代码。 ...

What are microservices ? 什么是微服务? 微服务可以有多种不同的定义方式, 具体取决于希望强调微服务架构的哪个方面, 不同作者会给出略有不同但相关的定义 Sam Newman, 微服务领域最有影响力的作者之一, 给出了一个极简的定义: “Microservices are small, autonomous services that work together.” 这个定义强调了这样一个事实: 微服务是彼此独立运行的应用程序, 但它们可以协作完成任务. 该定义还强调微服务是 “small (小的)”, 这里的 small 并不是指微服务代码量的大小, 而是指微服务具有狭窄且定义清晰的职责范围, 符合单一职责原则(Single Responsibility Principle) —— 即“只做一件事，并把它做好”. James Lewis 和 Martin Fowler 撰写的一篇开创性文章提供了一个更详细的定义, 他们将微服务定义为一种架构风格(architectural style) “an approach to developing a single application as a suite of small services, each running in its own process and communicating with lightweight mechanisms, often an HTTP resource API” ...

本篇文章介绍 FastAPI 的返回类型 response model 可以在返回函数的类型注解中声明该接口的响应数据类型 类型注解的用法和输入数据参数一样, 可以使用: Pydantic 模型 list 列表 dict 字典 scalar 标量值 (int, bool …) @app.post("/items/") async def create_item(item: Item) -> Item: ... @app.get("/items/") async def read_items() -> list[Item]: ... FastAPI 会使用返回类型完成一下事情: 验证返回类型 如果返回的数据无效, 说明业务代码有问题, FastAPI 会返回服务器错误, 而不是把数据发给客户端 在 OpenAPI 中为响应添加 JSON Schema 用于自动生成接口文档, 自动生成客户端代码 最重要的是 它会限制并过滤出数据, 只保留返回类型中定义的字段 response_model Parameter 有时候可能需要返回的数据和类型注解不完全一致, 例如: 可能想返回字典或数据库对象, 但声明的响应类型为 Pydantic 模型 这样 Pydantic 会做数据文档、验证等工作, 即使返回的是字典或 ORM 对象 如果直接用返回类型注解, 编辑器会提示类型不匹配的错误 这种情况下, 可以用路径装饰器的 response_model 参数来声明响应类型, 而不是用返回类型注解 class Item(BaseModel): name: str description: str | None = None price: float tax: float | None = None tags: list[str] = [] @app.post("/items/", response_model=Item) async def create_item(item: Item) -> Any: return item @app.get("/items/", response_model=list[Item]) async def read_items() -> Any: return [ {"name": "Portal Gun", "price": 42.0}, {"name": "Plumbus", "price": 32.0}, ] 注意: ...

这篇文章介绍 Fastapi 的 Cookie 和 Header 参数 Cookie Parameters 通过定义 Query 和 Path 参数一样定义 Cookie 参数 from typing Annotated from fastapi import Cookie, FastAPI app = FastAPI() @app.get("/items/") async def read_items(ads_id: Annotated[str | None, Cookie()] = None): return {"ads_id": ads_id} Cookie Parameters Models 如果有一组相关的 cookies, 可以使用 Pydantic model 来声明. 这样可以在多个部分复用这个模型, 同时还能一次性为所有参数声明验证规则和元数据. 下面使用 Pydantic 模型定义 Cookies, 然后将参数声明为 Cookie from typing import Annotated from fastapi import FastAPI, Cookie from pydantic import BaseModel app = FastAPI() class Cookie(BaseModel): session_id: str fatebook_tracker: str | None = None googall_tracker: str | None = None @app.get("/items/") async def read_items(cookies: Annotated[Cookies, Cookie()]): return cookies Forbid Extra Cookies 禁止额外的Cookie 在某些场景下(虽然并不常见), 可能希望限制 API 只能接收特定的 Cookie. 这样, API 就可以"自己"管理 Cookie 同意策略了. from typing import Annotated from fastapi import FastAPI, Cookie from pydantic import BaseModel app = FastAPI() class Cookies(BaseModel): model_config = {"extra": "forbid"} # forbid extra cookies session_id: str fatebook_tracker: str | None = None googall_tracker: str | None = None @app.get("/items/") async def read_items(cookies: Annotated[Cookies, Cookie()]): return cookies 这样, 如果客户端发送额外的 cookies, 则会收到一个错误响应. 例如, 客户端发送了 santa_tracker 这个额外 Cookie ...

本篇文章介绍 FastAPI Request Body 的进阶用法 Body - Multiple Parameters 首先, 可以将Path, Query 和 request body 参数声明自由的写在一起 对于 request body 参数可以是可选的, 并且可设置为默认的 None from typing import Annotated from fastapi import FastAPI, Path from pydantic import BaseModel app = FastAPI() class Item(BaseModel): name: str description: str | None = None price: float tax: float | None = None @app.put("/items/{item_id}") async def update_item( item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)], # Path q: str | None = None, # Query item: Item | None = None, # body ): results = {"item_id": item_id} if q: results.update({"q": q}) if item: results.update({"item": item}) return results Multiple body parameters 多参数请求体 在上面例子中, FastAPI 期望一个包含 Item 属性的 JSON body, 例如 { "name": "Foo", "description": "The pretender", "price": 42.0, "tax": 3.2 } 但也可以声明多个body parameters, 例如 item 和 user ...

这篇文章介绍 FastAPI 中的参数验证功能 Query Parameters and String Validations FastAPI 允许为参数声明额外的信息和验证规则 from fastapi import FastAPI app = FastAPI() @app.get("/items/") async def read_items(q: str | None = None): results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]} if q: results.update({"q": q}) return results q 是类型为 str | None 的查询参数, 这意味着它可以是字符串, 也可以是 None. 其默认值是 None, 因此 FastAPI 会识别它为“可选参数” FastAPI 通过 = None 的默认值知道该参数是非必填的 使用 str | None 还能帮助编辑器提供更好的类型提示和错误检测 Additional validation 额外验证 即使 q 是可选的, 但仍然可以设置条件: 如果提供了 q, 则长度不能超过50个字符 使用 Query 和 Annotated 来实现 from typing import Annotated from fastapi import FastAPI, Query app = FastAPI() @app.get("/items/") async def read_items(q: Annotated[str | None, Query(max_length=50)] = None): results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]} if q: results.update({"q": q}) return results 使用 Annotated 包装后, 就可以传递额外的元数据(Query(max_length=5)), 用于校验或者文档 ...

FastAPI 是一个现代、快速（高性能）的 Python Web 框架, 它自动处理参数的解析、验证和文档生成 本文将介绍 FastAPI 中三类最常用的参数: 路径参数 (Path Parameters)、查询参数 (Query Parameters) 和 请求体(Request Body) 的用法与原理 1. Path Parameters 路径参数 路径参数是 URL 路径中的动态部分, 使用 {} 包裹表示 from fastapi import FastAPI app = FastAPI() @app.get("/items/{item_id}") async def read_item(item_id: str): return {"item_id": item_id} 访问 /items/foo 返回: {"item_id": "foo"} Data conversion & validation 类型声明与自动转换 可以为路径参数声明类型, FastAPI 会自动解析并验证: @app.get("/items/{item_id}") async def read_item(item_id: int): return {"item_id": item_id} 访问 /items/3, item_id 会被转换为 int 类型 Routing orders 路由匹配顺序 路径匹配按声明顺序执行, 例如 @app.get("/users/me") async def read_user_me(): return {"user_id": "current_user"} @app.get("/users/{user_id}") async def read_user(user_id: str): return {"user_id": user_id} 必须先声明 /users/me, 否则会被 /users/{user_id} 捕获 ...