响应式网站下载,自己做的网站可以发布吗,六类网线制作,招商网站设计【FastAPI】中间件 一、概述二、作用2.1 日志记录与监控2.2 身份验证与授权2.3 CORS#xff08;跨域资源共享#xff09;2.4 Gzip压缩2.5 会话管理2.6 自定义功能2.7 执行顺序 三、 总结四、相关链接 一、概述
FastAPI的中间件提供了一种强大的机制#xff0c;允许开发者在… 【FastAPI】中间件 一、概述二、作用2.1 日志记录与监控2.2 身份验证与授权2.3 CORS跨域资源共享2.4 Gzip压缩2.5 会话管理2.6 自定义功能2.7 执行顺序 三、 总结四、相关链接 一、概述
FastAPI的中间件提供了一种强大的机制允许开发者在请求被处理之前以及响应返回给客户端之前对它们进行拦截和操作。 这为实现诸如日志记录、身份验证、性能监控、错误处理等功能提供了极大的灵活性。
二、作用
2.1 日志记录与监控
中间件可以用来记录所有进入系统的HTTP请求及其对应的响应。 例如你可以创建一个中间件来测量每个请求的处理时间并将这个信息作为自定义头部X-Process-Time添加到响应中。 这对于性能分析非常有用因为它可以帮助你了解哪些请求消耗了较多的时间从而优化系统性能。
import json
import logging
import os
import time
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware# 设置日志格式和级别
LOG_LEVEL os.getenv(LOG_LEVEL, INFO).upper()
logging.basicConfig(levelLOG_LEVEL,format%(asctime)s - %(levelname)s - %(message)s
)class LoggingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next) - Response:start_time time.perf_counter()response await call_next(request)process_time time.perf_counter() - start_timelog_data {client_ip: request.client.host,method: request.method,path: request.url.path,status_code: response.status_code,process_time: f{process_time:.4f}s,timestamp: datetime.datetime.fromtimestamp(time.time()).strftime(%Y-%m-%d %H:%M:%S)}# 根据环境变量决定是否打印详细的日志内容if LOG_LEVEL DEBUG:log_data[headers] dict(request.headers)log_data[query_params] dict(request.query_params)logging.info(json.dumps(log_data))return responseapp FastAPI()
app.add_middleware(LoggingMiddleware)app.get(/)
async def root():return {message: Hello World}if __name__ __main__:import uvicornuvicorn.run(app, host0.0.0.0, port8000, log_levelLOG_LEVEL.lower())测试方法
启动应用python main.py使用浏览器或Postman访问http://localhost:8000/查看命令行输出的日志信息确认日志已正确记录。
2024-12-25 19:21:56,568 - INFO - {client_ip: 127.0.0.1, method: GET, path: /, status_code: 200, process_time: 0.0003s, timestamp: 2024-12-25 19:21:56}
INFO: 127.0.0.1:53956 - GET / HTTP/1.1 200 OK设置日志配置 这里设置了日志记录器的基本配置。通过环境变量 LOG_LEVEL 来控制日志级别默认为 INFO。 这意味着除非设置了更低的日志级别如 DEBUG否则不会输出调试信息。 此外还指定了日志格式包括时间戳、日志级别和消息内容。
LOG_LEVEL os.getenv(LOG_LEVEL, INFO).upper()
logging.basicConfig(levelLOG_LEVEL,format%(asctime)s - %(levelname)s - %(message)s
)自定义中间件 LoggingMiddleware 中间件是用来处理请求和响应之间逻辑的理想场所。 在这个例子中LoggingMiddleware 被用来捕获每次请求的时间点然后在请求完成后计算处理所需的时间并将相关信息以 JSON 格式记录下来。这有助于监控 API 性能并诊断问题。
开始计时使用 time.perf_counter() 获取当前时间。调用下一个处理器通过 await call_next(request) 将请求传递给下一个中间件或最终的路由处理器。结束计时再次调用 time.perf_counter() 计算总耗时。构建日志数据创建一个字典 log_data 包含客户端 IP、HTTP 方法、路径、状态码及处理时间等信息。条件性添加更多信息如果日志级别为 DEBUG还会额外添加请求头和查询参数到日志中以便于更详细的调试。记录日志最后将构造好的日志数据转换成 JSON 字符串并通过 logging.info() 输出。
定义 FastAPI 应用程序 这里创建了一个 FastAPI 实例 app 并注册了之前定义的 LoggingMiddleware。随后定义了一个 GET 方法对应的路由 /当访问根路径时会返回一个 JSON 响应其中包含了一条简单的欢迎信息 “Hello World”。
app FastAPI()
app.add_middleware(LoggingMiddleware)app.get(/)
async def root():return {message: Hello World}启动 Uvicorn 服务器 这段代码确保只有当脚本作为主模块运行时才会启动 Uvicorn 服务器。 Uvicorn 是一个用于运行 ASGI 应用的异步服务器它将托管 FastAPI 应用程序并监听指定的主机地址 (0.0.0.0) 和端口 (8000)。 同时也将日志级别传递给 Uvicorn以保证一致性。
if __name__ __main__:import uvicornuvicorn.run(app, host0.0.0.0, port8000, log_levelLOG_LEVEL.lower())2.2 身份验证与授权
通过中间件可以在请求到达路由处理器之前检查用户的身份验证状态。
如果请求未通过身份验证则可以直接返回401 Unauthorized响应如果通过了身份验证但没有足够的权限访问特定资源则可以返回403 Forbidden响应。
这种方式确保了只有经过适当授权的用户才能访问受保护的API端点。
from datetime import datetime, timedelta
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from pydantic import BaseModel
from typing import Optional
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.base import BaseHTTPMiddlewareSECRET_KEY your_secret_key
ALGORITHM HS256
ACCESS_TOKEN_EXPIRE_MINUTES 30oauth2_scheme OAuth2PasswordBearer(tokenUrltoken)class Token(BaseModel):access_token: strtoken_type: strclass TokenData(BaseModel):username: Optional[str] Nonefake_users_db {johndoe: {username: johndoe,full_name: John Doe,email: johndoeexample.com,hashed_password: fakehashedsecret,disabled: False,}
}def verify_password(plain_password, hashed_password):return plain_password hashed_passworddef get_user(db, username: str):if username in db:user_dict db[username]return UserInDB(**user_dict)def authenticate_user(fake_db, username: str, password: str):user get_user(fake_db, username)if not user:return Falseif not verify_password(password, user.hashed_password):return Falsereturn userdef create_access_token(data: dict, expires_delta: Optional[timedelta] None):to_encode data.copy()if expires_delta:expire datetime.utcnow() expires_deltaelse:expire datetime.utcnow() timedelta(minutes15)to_encode.update({exp: expire})encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM)return encoded_jwtasync def get_current_user(token: str Depends(oauth2_scheme)):credentials_exception HTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailCould not validate credentials,headers{WWW-Authenticate: Bearer},)try:payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM])username: str payload.get(sub)exp: int payload.get(exp)if username is None or exp is None:raise credentials_exceptionif datetime.utcfromtimestamp(exp) datetime.utcnow():raise credentials_exceptiontoken_data TokenData(usernameusername)except JWTError:raise credentials_exceptionuser get_user(fake_users_db, usernametoken_data.username)if user is None:raise credentials_exceptionreturn userclass AuthMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next) - Response:token request.headers.get(Authorization)if token:try:await get_current_user(token.split()[1]) # Assuming Bearer tokenexcept HTTPException as e:return JSONResponse(status_codee.status_code, content{detail: e.detail})else:return JSONResponse(status_code401, content{detail: Unauthorized})return await call_next(request)app FastAPI()
app.add_middleware(AuthMiddleware)app.post(/token, response_modelToken)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm Depends()):user authenticate_user(fake_users_db, form_data.username, form_data.password)if not user:raise HTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailIncorrect username or password,headers{WWW-Authenticate: Bearer},)access_token_expires timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES)access_token create_access_token(data{sub: user.username}, expires_deltaaccess_token_expires)return {access_token: access_token, token_type: bearer}app.get(/users/me, response_modelUser)
async def read_users_me(current_user: User Depends(get_current_user)):return current_userif __name__ __main__:import uvicornuvicorn.run(app, host0.0.0.0, port8000)2.3 CORS跨域资源共享
为了使Web应用程序能够从不同的域名发出AJAX请求你需要配置CORS策略。 FastAPI内置了CORSMiddleware它允许你轻松地设置允许的来源、方法、头等参数从而实现安全且灵活的跨域通信2。 下面是一个支持CORS配置的应用程序示例允许来自特定源的请求并处理预检请求Preflight Requests。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp FastAPI()origins [http://localhost:3000,
]app.add_middleware(CORSMiddleware,allow_originsorigins,allow_credentialsTrue,allow_methods[*],allow_headers[*],expose_headers[X-Total-Count],max_age86400
)app.get(/)
async def root():return {message: Hello World}if __name__ __main__:import uvicornuvicorn.run(app, host0.0.0.0, port8000)2.4 Gzip压缩
对于需要提高传输效率的情况可以使用Gzip中间件来压缩响应内容。这样不仅可以减少网络带宽占用还能加快页面加载速度特别是在移动设备上效果尤为明显2。
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddlewareapp FastAPI()app.add_middleware(GZipMiddleware,minimum_size1000,compresslevel9
)app.get(/)
async def root():large_text a * 1024 # 创建一个较大的字符串用于测试压缩效果return {message: large_text}if __name__ __main__:import uvicornuvicorn.run(app, host0.0.0.0, port8000)2.5 会话管理
借助于SessionMiddleware你可以方便地管理用户的会话信息。 比如在登录后存储用户的认证令牌或其他临时数据这些数据可以通过cookie传递并在后续请求中读取2。
// A code block
var foo bar;2.6 自定义功能
除了上述提到的功能之外中间件还可以用于执行其他任何你想要在每次请求或响应时运行的任务。例如你可以编写一个中间件来解析并标准化查询字符串参数或者根据某些条件修改请求体中的内容7。
// A code block
var foo bar;2.7 执行顺序
值得注意的是当多个中间件同时存在于一个应用中时它们按照“后进先出”的原则被执行。 也就是说最后添加的那个中间件会在最前面被调用并且最先完成其工作之后再交给下一个中间件继续处理。
// A code block
var foo bar;三、 总结
总之FastAPI中的中间件是一个非常强大且灵活的工具它使得开发者能够在不改变核心业务逻辑的前提下轻松地向应用程序添加各种辅助性功能和服务。 无论是为了增强安全性、改善用户体验还是简化开发流程合理利用中间件都能带来显著的好处。
四、相关链接
【FastAPI】日志 【FastAPI】中间件 【FastAPI】简介 【FastAPI】BaseModel类
