腾讯云网站建设教程,百度收录规则,什么网站免费可以做app,百度员工收入工资表一#xff0c;什么是中间件
中间件是一种软件组件#xff0c;它在请求到达应用程序处理程序之前和/或响应发送回客户端之前执行操作。 请求从客户端发出。
请求首先经过Middleware 1。
然后经过Middleware 2。
请求到达FastAPI路由处理器。
响应从路由处理器返回。
响应经过…一什么是中间件
中间件是一种软件组件它在请求到达应用程序处理程序之前和/或响应发送回客户端之前执行操作。 请求从客户端发出。
请求首先经过Middleware 1。
然后经过Middleware 2。
请求到达FastAPI路由处理器。
响应从路由处理器返回。
响应经过Middleware 2。
最后经过Middleware 1。
响应返回给客户端。import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponseapp FastAPI()app.middleware(http)
async def simple_middleware(request: Request, call_next):# 在请求处理之前执行的代码print(fReceived request: {request.method} {request.url})# 调用下一个中间件或路由处理器response await call_next(request)# 在响应返回之前执行的代码如果响应是JSONResponse我们在响应头中添加一个自定义字段。if isinstance(response, JSONResponse):response.headers[X-Processed-By] SimpleMiddlewareprint(fProcessed response: {response.status_code})return responseapp.get(/)
async def root():return {message: Hello World}if __main__ __name__:uvicorn.run(app, host127.0.0.1, port8088)在FastAPI中中间件在很多场景下都非常有用比如
请求日志记录认证和授权响应修改性能监控跨域资源共享CORS处理
二常用的中间件
一CORSMiddleware
1同源策略与跨域资源共享
同源策略是一个重要的安全概念由网页浏览器强制执行。它限制了一个源origin如果两个 URL 的协议、端口和主机都相同的话则这两个 URL 是同源的中加载的文档或脚本如何与来自另一个源的资源进行交互。
定义如果两个URL的协议、域名和端口都相同则它们被认为是同源的。目的防止恶意网站读取另一个网站的敏感数据。限制脚本只能访问来自同一源的数据不能直接访问不同源的资源。
例如
https://example.com/page1 可以访问 https://example.com/page2https://example.com 不能直接访问 https://api.example.com 或 http://example.com不同协议
CORS是一种机制它使用额外的 HTTP 头来告诉浏览器让运行在一个源domain上的 Web 应用被准许访问来自不同源服务器上的指定的资源。
目的允许服务器声明哪些源可以访问它们的资源从而放宽同源策略的限制。工作原理服务器在响应中包含特定的 HTTP 头告诉浏览器允许跨域请求。
关键的 CORS 头部
Access-Control-Allow-Origin: 指定允许访问资源的源。Access-Control-Allow-Methods: 指定允许的 HTTP 方法。Access-Control-Allow-Headers: 指定允许的请求头。 同源策略和CORS看似矛盾但实际上它们共同构成了web安全和功能性之间的平衡。为什么在有同源策略的情况下还需要CORS? Web的演变 - 早期Web最初的Web主要由静态页面组成不同源之间的交互很少。- 同源策略的引入随着Web变得更加动态同源策略被引入以防止潜在的跨站点脚本攻击XSS和数据窃取。- Web 2.0时代随着AJAX的兴起Web应用变得更加动态和交互式。- 现代Web现在的Web充满了单页应用SPAs、微服务架构和复杂的API驱动的应用。同源策略的局限性 - 虽然同源策略提供了重要的安全保护但它也限制了合法的跨域请求。- 在现代Web应用中前端和后端经常部署在不同的域上或者一个应用需要访问多个不同域的API。CORS的必要性 - 业务需求公司可能需要在多个子域或完全不同的域之间共享资源。- API经济许多公司提供API服务这些API需要被不同域的客户端访问。- 微服务架构不同的服务可能部署在不同的域上但需要相互通信。- 开发和测试开发环境和生产环境可能使用不同的域。CORS如何平衡安全和功能 - 控制访问CORS允许服务器明确指定哪些域可以访问其资源。- 细粒度控制可以控制允许的HTTP方法、头部等。- 预检请求对于非简单请求CORS使用预检请求机制增加了一层安全检查。- 保持同源策略CORS并没有废除同源策略而是提供了一种受控的方式来放宽限制。CORS的优势 - 安全性虽然允许跨域请求但CORS通过明确的服务器配置来维护安全性。- 灵活性开发者可以构建更复杂、分布式的应用架构。- 标准化CORS提供了一个标准化的方法来处理跨域请求取代了之前的一些不安全或复杂的变通方法如JSONP。实际应用场景 - 前后端分离前端可能托管在CDN上而后端API在不同的域。- 第三方服务集成如嵌入地图、支付服务或社交媒体小工具。- 多环境部署开发、测试和生产环境可能使用不同的域。2使用CORSMiddleware
以前后端分离项目为例。
在前端本身不需要特别的 CORS 配置因为 CORS 主要是由服务器端控制的。但是你需要确保你的 API 请求使用了正确的 URL例如
import axios from axios;const api axios.create({baseURL: http://localhost:8000, // 你的 FastAPI 服务器地址
});// 使用方式
api.get(/some-endpoint).then(response {console.log(response.data);
});在 fastapi 中需要配置 CORS 中间件来允许跨源请求
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp FastAPI()# 配置 CORS
app.add_middleware(CORSMiddleware,allow_origins[http://localhost:8080], # 允许的源这里假设你的前端应用运行在 8080 端口allow_credentialsTrue,allow_methods[*], # 允许所有方法allow_headers[*], # 允许所有头
)# 你的路由和其他代码...app.get(/)
async def root():return {message: Hello World}if __name__ __main__:import uvicornuvicorn.run(app, host0.0.0.0, port8000)二GZipMiddleware
1HTTP 响应压缩
自动压缩 HTTP 响应是一个优化技术旨在减少传输的数据量从而加快网页加载速度并减少带宽使用。这种方式尤其适用于文本内容如 HTML、CSS、JavaScript 和 JSON。
1确保 Web 服务器如 Nginx 或 Apache配置了 Gzip 或 Brotli 压缩。服务器会在发送响应前自动压缩响应数据。 在 Nginx 中启用 Gzip 压缩 http {gzip on;gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xmlrss text/javascript;gzip_proxied any;gzip_min_length 1000;
}在 Apache 中启用 Gzip 压缩: IfModule mod_deflate.cAddOutputFilterByType DEFLATE text/html text/plain text/xml text/css application/json application/javascript text/javascript
/IfModule2客户端支持现代浏览器默认支持 Gzip 和 Brotli 压缩。在请求头中浏览器会发送 Accept-Encoding 字段指明支持的压缩算法
Accept-Encoding: gzip, deflate, br3服务器响应服务器检查 Accept-Encoding 字段并使用适当的压缩算法对响应内容进行压缩同时在响应头中添加 Content-Encoding 字段指明使用的压缩算法
HTTP/1.1 200 OK
Content-Type: text/html
Content-Encoding: gzip
Content-Length: 512compressed content4客户端对数据进行解压以恢复原始内容。
2使用 GZipMiddleware
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponseapp FastAPI()# 添加 GZip 中间件在响应大小超过 100000 字节时才进行压缩
app.add_middleware(GZipMiddleware, minimum_size100000)app.get(/item/)
async def test():file_path 02 部署前准备--配置Settings.py.mp4# # 获取文件大小# import os# file_size os.path.getsize(file_path)return FileResponse(file_path,media_typevideo/mp4,filenamefile_path,# headers{Content-Length: str(file_size)})if __main__ __name__:uvicorn.run(app, host127.0.0.1, port8088)三TrustedHostMiddleware
1HTTP Host Header攻击
HTTP Host Header攻击是一种利用 HTTP 请求中 Host 头部的安全漏洞。这种攻击可能导致各种安全问题包括但不限于网站重定向、缓存污染和密码重置漏洞。 HTTP Host header attacks 如何识别和利用HTTP Host头的漏洞 2使用 TrustedHostMiddleware
Trusted Host Middleware 是一种安全机制用于限制应用程序只接受来自特定主机或域名的请求。这是一个重要的安全特性。
工作原理 - 检查请求头中间件检查每个incoming请求的 Host 头。 - 比对允许列表将 Host 头与预先配置的允许主机列表进行比对。 - 处理结果如果 Host 头匹配允许列表中的一项请求被允许通过否则请求被拒绝。
from fastapi import FastAPI
from starlette.middleware.trustedhost import TrustedHostMiddlewareapp FastAPI()app.add_middleware(TrustedHostMiddleware, allowed_hosts[example.com, www.example.com]
)四更多中间件 Starlette 官档 - 中间件 ASGI Awesome 列表 六一些有趣的自定义中间件
限流中间件限制每个IP在特定时间窗口内的请求次数。响应时间模拟中间件、为每个请求添加随机延迟。用于测试前端应用对不同响应时间的处理能力。可以模拟真实世界的网络延迟帮助发现潜在的超时问题。请求ID中间件为每个请求分配一个唯一的ID。方便跟踪和调试请求特别是在分布式系统中。响应内容修改中间件修改JSON响应中的特定内容。可以用于统一处理某些响应如敏感信息脱敏。日志中间件记录每个请求和响应的详细信息对于调试和监控非常有用。错误处理中间件全局捕获异常并自定义错误响应。安全头中间件添加安全相关的 HTTP 头提升应用安全性。统计中间件记录请求的统计信息如请求数量、响应时间等。HTTP 缓存中间件添加 HTTP 缓存头提高页面加载速度。请求重试中间件在特定情况下对请求进行重试。
import asyncio
import logging
import random
import timeimport uvicorn
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddlewareapp FastAPI()# 1. 限流中间件
class RateLimitMiddleware(BaseHTTPMiddleware):def __init__(self, app, max_requests: int, time_window: int):super().__init__(app)self.max_requests max_requestsself.time_window time_windowself.request_counts {}async def dispatch(self, request: Request, call_next):client_ip request.client.hostcurrent_time time.time()if client_ip in self.request_counts:if current_time - self.request_counts[client_ip][timestamp] self.time_window:self.request_counts[client_ip] {count: 1, timestamp: current_time}else:self.request_counts[client_ip][count] 1if self.request_counts[client_ip][count] self.max_requests:return JSONResponse(status_code429, content{error: Too many requests})else:self.request_counts[client_ip] {count: 1, timestamp: current_time}return await call_next(request)# 2. 响应时间模拟中间件
class ResponseTimeSimulatorMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):# 模拟0.1秒到1秒的随机延迟delay random.uniform(0.1, 1)await asyncio.sleep(delay)response await call_next(request)response.headers[X-Simulated-Delay] f{delay:.2f}sreturn response# 3. 请求ID中间件
class RequestIDMiddleware(BaseHTTPMiddleware):def __init__(self, app):super().__init__(app)self.request_id_counter 0async def dispatch(self, request: Request, call_next):self.request_id_counter 1request_id fREQ-{self.request_id_counter:05d}request.state.request_id request_idresponse await call_next(request)response.headers[X-Request-ID] request_idreturn response# 4. 响应内容修改中间件
class ResponseModifierMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):response await call_next(request)if isinstance(response, JSONResponse):content response.body.decode()modified_content content.replace(example, EXAMPLE)return Response(contentmodified_content, status_coderesponse.status_code,headersdict(response.headers), media_typeresponse.media_type)return responselogging.basicConfig(levellogging.INFO)
logger logging.getLogger(__name__)# 5. 日志中间件
class LoggingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):start_time time.time()response await call_next(request)process_time time.time() - start_timelogger.info(fRequest: {request.method} {request.url} completed in {process_time:.4f}s)return response# 6. 错误处理中间件
class ErrorHandlingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):try:response await call_next(request)return responseexcept Exception as exc:return JSONResponse(content{error: str(exc)}, status_code500)# 7. 安全头中间件
class SecurityHeadersMiddleware(BaseHTTPMiddleware):async def dispatch(self, request, call_next):response await call_next(request)response.headers[Content-Security-Policy] default-src selfresponse.headers[X-Content-Type-Options] nosniffresponse.headers[X-Frame-Options] DENYresponse.headers[X-XSS-Protection] 1; modeblockreturn response# 8. 统计中间件
class StatsMiddleware(BaseHTTPMiddleware):def __init__(self, app):super().__init__(app)self.total_requests 0self.total_time 0.0async def dispatch(self, request: Request, call_next):start_time time.time()response await call_next(request)process_time time.time() - start_timeself.total_requests 1self.total_time process_timeresponse.headers[X-Total-Requests] str(self.total_requests)response.headers[X-Total-Time] f{self.total_time:.2f}sresponse.headers[X-Process-Time] f{process_time:.2f}sreturn response# 9. HTTP 缓存中间件
class HTTPCacheMiddleware(BaseHTTPMiddleware):async def dispatch(self, request, call_next):response await call_next(request)response.headers[Cache-Control] public, max-age3600return response# 10. 请求重试中间件
class RetryMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):retries 3for attempt in range(retries):response await call_next(request)if response.status_code 200:return responsetime.sleep(1)return response# 添加中间件到应用
app.add_middleware(RateLimitMiddleware, max_requests5, time_window60)
app.add_middleware(ResponseTimeSimulatorMiddleware)
app.add_middleware(RequestIDMiddleware)
app.add_middleware(ResponseModifierMiddleware)
app.add_middleware(LoggingMiddleware)
app.add_middleware(ErrorHandlingMiddleware)
app.add_middleware(SecurityHeadersMiddleware)app.get(/)
async def root():return {message: This is an example response}if __main__ __name__:uvicorn.run(app, host127.0.0.1, port8088)
三使用中间件的注意事项
一顺序
中间件是按照添加的顺序依次执行的。顺序会影响请求和响应的处理流程所以要注意中间件的添加顺序。
假设我们有两个中间件一个是日志记录中间件另一个是 GZip 压缩中间件。我们希望日志记录在请求处理之前和之后都能记录信息而压缩应该在响应返回之前进行。
from fastapi import FastAPI, Request
from fastapi.middleware.gzip import GZipMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logginglogging.basicConfig(levellogging.INFO)
logger logging.getLogger(__name__)class LoggingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):start_time time.time()logger.info(fRequest: {request.method} {request.url})response await call_next(request)process_time time.time() - start_timelogger.info(fCompleted in {process_time:.4f}s)return responseapp FastAPI()# 日志记录中间件在 GZip 中间件之前
app.add_middleware(LoggingMiddleware)
app.add_middleware(GZipMiddleware, minimum_size1000)app.get(/)
async def read_root():return {message: Hello, World!}资源管理 - 中间件可能需要管理资源如数据库连接。 - 确保正确地打开和关闭资源考虑使用上下文管理器。
异步操作 - FastAPI支持异步操作中间件也应该尽可能是异步的。 - 使用 async/await 语法避免阻塞操作。
二性能
中间件会增加请求处理的开销要避免添加过多不必要的中间件尤其是在高并发场景下。
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logging
import oslogging.basicConfig(levellogging.INFO)
logger logging.getLogger(__name__)class LoggingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):start_time time.time()logger.info(fRequest: {request.method} {request.url})response await call_next(request)process_time time.time() - start_timelogger.info(fCompleted in {process_time:.4f}s)return responseapp FastAPI()if os.getenv(ENV) development:app.add_middleware(LoggingMiddleware)app.get(/)
async def read_root():return {message: Hello, World!}中间件会对每个请求都执行如果包含耗时操作会显著影响应用性能。 尽量保持中间件轻量避免在中间件中执行耗时操作。如果必须考虑异步操作或缓存策略。
三异常处理
中间件中处理的异常不会自动传递给 FastAPI 的全局异常处理器需要在中间件中捕获并处理异常。
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponseclass ErrorHandlingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):try:response await call_next(request)return responseexcept HTTPException as exc:return JSONResponse(content{error: exc.detail}, status_codeexc.status_code)except Exception as exc:return JSONResponse(content{error: Internal Server Error}, status_code500)app FastAPI()
app.add_middleware(ErrorHandlingMiddleware)app.get(/)
async def read_root():raise Exception(An unexpected error occurred)中间件可以捕获和处理异常但也可能掩盖重要的错误信息。 在捕获异常时确保记录足够的信息用于调试。考虑只捕获特定类型的异常。
四状态共享
使用 request.state 可以在中间件和路由处理器之间共享状态信息。
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddlewareclass RequestIDMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):request.state.request_id unique-request-idresponse await call_next(request)response.headers[X-Request-ID] request.state.request_idreturn responseapp FastAPI()
app.add_middleware(RequestIDMiddleware)app.get(/)
async def read_root(request: Request):return {request_id: request.state.request_id}读取请求体 - 读取请求体后它就被消耗了后续的处理包括路由处理函数将无法再次读取。 - 如果必须在中间件中读取请求体考虑缓存它或者使用 request.stream() 来允许多次读取。
五响应修改
中间件可以修改响应对象但需要确保不会破坏响应的结构和内容。
class ResponseModifierMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):response await call_next(request)if isinstance(response, JSONResponse):response.content {modified: response.content}return response修改响应可能会影响应用的预期行为特别是当多个中间件都修改响应时。 谨慎修改响应确保修改不会破坏响应的结构或语义。考虑使用装饰器而不是中间件来修改特定路由的响应。
