FastAPI 中间件详解：实现高性能 Web 应用的完整指南和实际案例

在现代 Web 开发中，FastAPI 已成为开发者构建高性能 API 的首选框架之一。其引人注目的特性之一就是中间件机制。中间件在请求处理管道中插入额外的逻辑，能够显著提高应用的可扩展性和可维护性。今天，我们将深入探讨 FastAPI 的中间件系统，包括其工作原理、使用方法，以及在实际项目中的一些应用示例。

FastAPI 中间件的核心机制

中间件的基本工作流程

FastAPI 中的中间件是通过 Starlette 框架实现的。每个中间件是一个 异步函数，接受请求对象并返回响应对象。中间件的设计允许你在请求到达路由之前进行预处理，或者在响应发送给客户端之前进行后处理。

中间件的执行顺序是它们被添加到应用程序中的顺序。每个请求经过所有中间件后，才会达到最终的路由处理器。这种设计提供了极大的灵活性，允许开发者轻松添加或删除功能，而不影响应用的主要逻辑。

以下是一个简单的中间件示例，用于测量请求处理时间，并将其添加到响应头中：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import timeapp = FastAPI()@app.middleware("http")
async def add_process_time_header(request: Request, call_next):start_time = time.time()response = await call_next(request)process_time = time.time() - start_timeresponse.headers["X-Process-Time"] = str(process_time)return response@app.get("/")
async def read_root():return {"message": "Hello World"}

这个中间件在处理请求前记录开始时间，调用 call_next 继续请求处理，并在请求完成后计算请求处理时间，将其添加到响应头中。

深入解析中间件的实现

Starlette 和 BaseHTTPMiddleware

FastAPI 的中间件功能是由 Starlette 提供的。Starlette 是一个轻量级的 ASGI 框架，专注于高性能和可扩展性。BaseHTTPMiddleware 是 Starlette 提供的一个基类，用于实现 HTTP 中间件。

class BaseHTTPMiddleware:async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:raise NotImplementedError("You need to override the dispatch method.")

在这个抽象基类中，dispatch 方法是核心方法，所有自定义的中间件都需要实现这个方法。通过 dispatch 方法，你可以在请求被传递到下一个中间件或最终的路由处理器之前，插入特定的逻辑。

FastAPI 的中间件注册

在 FastAPI 中，中间件通过 add_middleware 方法进行注册，并存储在 self.middleware_stack 中，形成一个链式的中间件结构。

class FastAPI:def __init__(self, ...):self.middleware_stack = self.build_middleware_stack()def build_middleware_stack(self) -> ASGIApp:...for middleware in reversed(self.user_middleware):app = middleware(app)...return appdef add_middleware(self, middleware_class: Type[BaseHTTPMiddleware], **options: Any) -> None:middleware = middleware_class(app=self, **options)self.user_middleware.append(middleware)

在这个结构中，每次请求都会依次经过所有注册的中间件，最后抵达路由处理器。这种设计不仅简化了中间件的添加和删除，也使得应用架构更加模块化和可维护。

实际应用中的中间件案例

案例 1：企业级身份验证系统

在大型应用中，身份验证是一个关键组件。我们可以通过中间件实现复杂的身份验证逻辑，如多因素认证（MFA）。以下是一个使用 JWT 进行身份验证的中间件示例：

from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from pydantic import BaseModel
from aiomysql.sa import create_engine
from sqlalchemy import MetaData, Table, Column, Integer, String, DateTime
from aiologger.loggers.json import JsonLogger
from aiologger.handlers.files import AsyncFileHandler
import asyncio
import datetimeapp = FastAPI()
metadata = MetaData()users = Table('users', metadata,Column('id', Integer, primary_key=True),Column('username', String(50)),Column('password_hash', String(128)),Column('mfa_secret', String(128))
)auth_logs = Table('auth_logs', metadata,Column('id', Integer, primary_key=True),Column('username', String(50)),Column('timestamp', DateTime),Column('status', String(10))
)SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")logger = JsonLogger.with_default_handlers()
file_handler = AsyncFileHandler('/var/log/auth.log')
logger.addHandler(file_handler)engine = create_engine(user='user',db='auth_db',host='localhost',password='password',port=3306,loop=asyncio.get_event_loop(),echo=True
)class User(BaseModel):username: strpassword_hash: strmfa_secret: strclass TokenData(BaseModel):username: str = Noneasync def get_current_user(token: str = Depends(oauth2_scheme)):credentials_exception = HTTPException(status_code=401,detail="Could not validate credentials",headers={"WWW-Authenticate": "Bearer"},)try:payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])username: str = payload.get("sub")if username is None:raise credentials_exceptiontoken_data = TokenData(username=username)except JWTError:raise credentials_exceptionasync with engine.acquire() as conn:result = await conn.execute(users.select().where(users.c.username == token_data.username))user_dict = await result.fetchone()if user_dict is None:raise credentials_exceptionreturn user_dict@app.middleware("http")
async def log_auth_request(request: Request, call_next):username = request.headers.get("Authorization", "").split(" ")[1] if "Authorization" in request.headers else "unknown"start_time = datetime.datetime.utcnow()response = await call_next(request)process_time = (datetime.datetime.utcnow() - start_time).total_seconds()async with engine.acquire() as conn:await conn.execute(auth_logs.insert().values(username=username,timestamp=datetime.datetime.utcnow(),status="success" if response.status_code == 200 else "failure"))await logger.info({"username": username,"method": request.method,"path": request.url.path,"status": response.status_code,"process_time": process_time})return response@app.post("/token")
async def login_for_access_token(username: str, password: str):async with engine.acquire() as conn:result = await conn.execute(users.select().where(users.c.username == username))user_dict = await result.fetchone()if user_dict is None or user_dict.password_hash != password:raise HTTPException(status_code=401, detail="Incorrect username or password")access_token_expires = datetime.timedelta(minutes=30)access_token = jwt.encode({"sub": username}, SECRET_KEY, algorithm=ALGORITHM)return {"access_token": access_token, "token_type": "bearer"}@app.get("/protected")
async def protected_route(user: dict = Depends(get_current_user)):return {"message": f"Welcome, {user['username']}!"}

在这个示例中，我们使用 OAuth2PasswordBearer 来处理身份验证请求，并使用 jwt 库来解析和验证 JWT。aiomysql 用于连接 MySQL 数据库，aiologger 用于记录日志。这种设计不仅实现了身份验证，还提供了详细的日志记录功能，以便审计和监控。

案例 2：请求限流实现

请求限流是保护服务器免受流量冲击和滥用的重要措施。我们可以使用中间件来限制每秒的请求数量，如下所示：

from fastapi import FastAPI, Request, HTTPException
from aiolimiter import AsyncLimiter
import asyncioapp = FastAPI()
limiter = AsyncLimiter(max_rate=10, time_period=1)@app.middleware("http")
async def limit_requests(request: Request, call_next):try:await limiter.acquire()except asyncio.TimeoutError:raise HTTPException(status_code=429, detail="Too Many Requests")response = await call_next(request)return response@app.get("/")
async def read_root():return {"message": "Hello World"}

使用 aiolimiter 库，我们可以轻松实现每秒最多 10 个请求的限流策略。当请求超过限制时，返回 HTTP 429 错误。这种方法非常适合用在需要保护的 API 服务中，以防止恶意流量攻击。

中间件的性能影响与优化

虽然中间件非常强大，但它们也会对应用的性能产生影响。每个请求都会经过所有中间件，因此需要谨慎设计，确保它们不会引入不必要的延迟。特别是那些涉及到 I/O 操作的中间件，更应该优化以减少性能开销。

以下是一些优化中间件性能的建议：

  • 异步处理: 确保所有 I/O 操作都是异步的，以避免阻塞事件循环。
  • 减少逻辑复杂度: 在中间件中保持逻辑简单，避免复杂的计算和数据处理。
  • 缓存结果: 如果中间件中有重复计算的逻辑，可以考虑缓存结果，以减少重复计算的开销。
  • 合理排序: 根据中间件的重要性和执行时间来合理排序，确保最重要的中间件先执行。

动态配置和管理中间件

FastAPI 允许你在运行时动态添加或删除中间件。这对于需要根据环境或条件来调整中间件的应用非常有用。你可以通过 app.middleware 方法来添加中间件，也可以通过 app.user_middleware 属性来查看和管理已注册的中间件。

例如，你可以根据环境变量来决定是否启用某个中间件：

import os
from fastapi import FastAPIapp = FastAPI()if os.getenv("ENABLE_AUTH", "false").lower() == "true":app.middleware("http")(authenticate_request)

这种动态管理的能力使得应用在不同的环境中能够灵活地适应需求变化。

总结

FastAPI 的中间件系统是其强大功能的一部分，通过合理使用中间件，我们可以构建功能丰富、性能优异的 Web 应用。无论是身份验证、请求限流、CORS 控制还是日志记录，中间件都能提供有效的解决方案。希望通过本文的深入探讨，你能更好地理解和应用 FastAPI 中间件，为你的项目增光添彩。

如果你有任何问题或建议，欢迎随时与我交流！在这里插入代码片

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/474326.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

MySQL索引的底层实现原理是什么？

MySQL索引的底层实现原理是什么？

MySQL索引的底层实现主要基于B树数据结构。B树是一种平衡多路查找树，具有以下特点： 1、树的所有叶子节点都位于同一层： 这确保了从根节点到每个叶子节点的路径长度相同，保证了查询效率的一致性。 2、节点中的数据按键值大小有序…
阅读更多...
手搓神经网络（MLP）解决MNIST手写数字识别问题 | 数学推导+代码实现 | 仅用numpy，tensor和torch基本计算 | 含正反向传播数学推导

手搓神经网络（MLP）解决MNIST手写数字识别问题 | 数学推导+代码实现 | 仅用numpy，tensor和torch基本计算 | 含正反向传播数学推导

手写数字识别（神经网络入门） 文章目录 手写数字识别（神经网络入门）实验概述实验过程数据准备模型实现线性变换层前向传播反向传播更新参数整体实现 激活函数层（ReLU）前向传播反向传播整体实现 Softmax层&am…
阅读更多...
在MATLAB中导入TXT文件的若干方法

在MATLAB中导入TXT文件的若干方法

这是一篇关于如何在MATLAB中导入TXT文件的文章，包括示例代码和详细说明 文章目录 在MATLAB中导入TXT文件1. 使用readtable函数导入TXT文件示例代码说明 2. 使用load函数导入TXT文件示例代码说明 3. 使用importdata函数导入TXT文件示例代码说明 4. 自定义导入选项示例…
阅读更多...
ks 小程序sig3

ks 小程序sig3

前言 搞了app版的快手之后 （被风控麻了） 于是试下vx小程序版的 抓包调试 小程序抓包问题 网上很多教程， github也有开源的工具代码 自行搜索 因为我们需要调试代码，所以就用了下开源的工具 （可以用chrome的F12功能&a…
阅读更多...
解决Spring Boot整合Redis时的连接问题

解决Spring Boot整合Redis时的连接问题

前言 在使用Spring Boot整合Redis的过程中，经常会遇到连接问题，尤其是当Redis服务部署在远程服务器上时。 问题描述 当你尝试连接到Redis服务器时，可能会遇到以下错误： org.springframework.data.redis.connection.PoolExcept…
阅读更多...
算法--“汽车加油”问题.

算法--“汽车加油”问题.

def greedy():n 100 # 汽车满油后可行驶的最大距离d [50, 80, 39, 60, 40, 32] # 加油站的距离k len(d) # 加油站的数量# 检查是否有加油站距离超过汽车的最大行驶距离for dist in d:if dist > n:print(no solution)returnnum 0 # 加油次数current_position 0 # 当…
阅读更多...
道陟科技EMB产品开发进展与标准设计的建议｜2024电动汽车智能底盘大会

道陟科技EMB产品开发进展与标准设计的建议｜2024电动汽车智能底盘大会

11月12日，2024电动汽车智能底盘大会在重庆开幕。会议由中国汽车工程学会主办，电动汽车产业技术创新战略联盟、中国汽车工程学会智能底盘分会、智能绿色车辆与交通全国重点实验室承办。本届大会围绕电动汽车智能底盘相关技术发展与融合，满足高…
阅读更多...
sqli—labs靶场 5-8关 （每日4关练习）持续更新！！！

sqli—labs靶场 5-8关 （每日4关练习）持续更新！！！

Less-5 上来先进行查看是否有注入点，判断闭合方式，查询数据列数，用union联合注入查看回显位，发现到这一步的时候，和前四道题不太一样了，竟然没有回显位？？？ 我们看一下源…
阅读更多...
【qt】控件3

【qt】控件3

1.setToolTip和setToolTipDuration setToolTip这个函数用来设置提醒内容 setToolTipDuration这个函数用来设置提醒时间 Widget::Widget(QWidget *parent): QWidget(parent), ui(new Ui::Widget) {ui->setupUi(this);ui->help->setToolTip("按下这个按键就可以提…
阅读更多...
STM32 使用 STM32CubeMX HAL库实现低功耗模式

STM32 使用 STM32CubeMX HAL库实现低功耗模式

STM32 使用 HAL 库的低功耗模式测试使用 ...... 矜辰所致前言 上次画了一个 STM32L010F4 最小系统的板子，也做了一些基本测试，但是最重要的低功耗一直拖到现在，以前在使用 STM32L151 的时候用标准库做过低功耗的项目，现在都使…
阅读更多...
js实现导航栏鼠标移入时，下划线跟随鼠标滑动

js实现导航栏鼠标移入时，下划线跟随鼠标滑动

话不多说&#xff0c;上代码&#xff1a; html代码&#xff1a; <div class"nav clearfix"><div class"bottomLine"></div><ul class"clearfix"><li class"nav__item"><a href"./index.html&…
阅读更多...
React教程第二节之虚拟DOM与Diffing算法理解

React教程第二节之虚拟DOM与Diffing算法理解

1、什么是虚拟DOM 虚拟DOM 是javascript的一个对象&#xff0c;是内存中的一种数据结构&#xff0c;以树的形式存储UI的状态&#xff0c;树中的每个节点都代表着真实的DOM&#xff0c;用来描述我们希望在页面看到的 HTML结构&#xff1b; 现在的MVVM 框架&#xff0c;大多使用…
阅读更多...
多线程4：线程池、并发、并行、综合案例-抢红包游戏

多线程4：线程池、并发、并行、综合案例-抢红包游戏

欢迎来到“雪碧聊技术”CSDN博客&#xff01; 在这里&#xff0c;您将踏入一个专注于Java开发技术的知识殿堂。无论您是Java编程的初学者&#xff0c;还是具有一定经验的开发者&#xff0c;相信我的博客都能为您提供宝贵的学习资源和实用技巧。作为您的技术向导&#xff0c;我将…
阅读更多...
【WPF】Prism库学习（一）

【WPF】Prism库学习（一）

Prism介绍 1. Prism框架概述&#xff1a; Prism是一个用于构建松耦合、可维护和可测试的XAML应用程序的框架。它支持WPF、.NET MAUI、Uno Platform和Xamarin Forms等多个平台。对于每个平台&#xff0c;Prism都有单独的发布版本&#xff0c;并且它们在不同的时间线上独立开发。…
阅读更多...
通过华为鲲鹏认证发行上市的集成平台产品推荐

通过华为鲲鹏认证发行上市的集成平台产品推荐

华为鲲鹏认证是技术实力与品质的权威象征&#xff0c;代表着产品达到了高标准的要求。从技术层面看&#xff0c;认证确保产品与华为鲲鹏架构深度融合&#xff0c;能充分释放鲲鹏芯片的高性能、低功耗优势&#xff0c;为集成平台的高效运行提供强大动力。在安全方面&#xff0c;…
阅读更多...
STM32 ADC --- 任意单通道采样

STM32 ADC --- 任意单通道采样

STM32 ADC — 单通道采样 文章目录 STM32 ADC --- 单通道采样cubeMX配置代码修改&#xff1a;应用 使用cubeMX生成HAL工程 需求&#xff1a;有多个通道需要进行ADC采样&#xff0c;实现每次采样只采样一个通道&#xff0c;且可以随时采样不同通道的功能。 cubeMX配置 这里我们…
阅读更多...
web——upload-labs——第十二关——%00截断

web——upload-labs——第十二关——%00截断

查看源码 分析源码我们可以知道&#xff0c;这里是基于白名单过滤&#xff0c;只允许上传jpg,png,gif&#xff0c;但是这里注意第八行&#xff0c;上传路径是可以控制的&#xff0c;所以可以利用%00截断&#xff0c;来达到上传木马的目的。这里要注意一下&#xff0c;%00截断想…
阅读更多...
STM32 创建一个工程文件（寄存器、标准库）

STM32 创建一个工程文件（寄存器、标准库）

首先到官网下载对应型号的固件包&#xff1a; 像我的STM32F103C8T6的就下载这个&#xff1a; 依次打开&#xff1a; .\STM32F10x_StdPeriph_Lib_V3.5.0\STM32F10x_StdPeriph_Lib_V3.5.0\Libraries\CMSIS\CM3\DeviceSupport\ST\STM32F10x\startup\arm 可以看到&#xff1a; 这…
阅读更多...
Spark：大数据处理的强大引擎

Spark：大数据处理的强大引擎

一、Spark 简介 Apache Spark 是一个专为大规模数据处理而设计的快速、通用、可扩展的大数据分析计算引擎。它诞生于 2009 年&#xff0c;由美国加州伯克利大学的 AMP 实验室开发&#xff0c;2013 年被纳入 Apache 开源项目&#xff0c;并迅速成为顶级项目。 Spark 被认为是 …
阅读更多...
鸿蒙HarmonyOS 地图定位到当前位置 site查询等操作

鸿蒙HarmonyOS 地图定位到当前位置 site查询等操作

应用服务Map使用 地图定位 地点查询及导航 周边查询 点位标记定义等 地图定位 前提地图已经能正常显示&#xff0c;若不能显示请大家参考之前的那篇如何显示地图的博文 地图相关的api 位置效果图&#xff1a; module.json5配置权限 "requestPermissions": [{&…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023