python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > FastAPI处理CORS跨域

FastAPI优雅处理CORS跨域与日志记录的实战指南

作者:小庄-Python办公

在当今的 Web 开发领域,API(应用程序接口)是连接前后端以及不同服务之间的桥梁,FastAPI 的核心优势在于其高性能和开发效率的完美结合,下面我们就来看看FastAPI如何处理CORS跨域与日志记录吧

第一章:为什么 FastAPI 成为现代 Python Web 开发的首选?

在当今的 Web 开发领域,API(应用程序接口)是连接前后端以及不同服务之间的桥梁。Python 作为一门简洁且强大的语言,拥有众多优秀的 Web 框架,而 FastAPI 正是近年来异军突起的新星。

FastAPI 的核心优势在于其高性能开发效率的完美结合。它基于 Python 3.6+ 的类型提示(Type Hints),利用 Pydantic 进行数据验证和序列化,不仅保证了代码的可读性,还自动生成了交互式的 API 文档(Swagger UI 和 ReDoc)。

然而,当我们从简单的 “Hello World” 迈向生产级应用时,两个问题几乎是不可避免的:

本篇文章将深入探讨如何在 FastAPI 中结合 Python 的标准库和中间件机制,优雅且专业地解决这两个核心痛点。

第二章:跨域资源共享 (CORS) 的原理与 FastAPI 实战

什么是 CORS

CORS(Cross-Origin Resource Sharing)是一种基于 HTTP 头的机制,它允许服务器声明哪些源(Origin)有权访问其资源。浏览器出于安全考虑,默认遵循“同源策略”,即如果你的前端运行在 localhost:3000,而后端在 localhost:8000,浏览器会拦截它们之间的请求。

在 FastAPI 中解决 CORS

FastAPI 提供了一个非常便捷的工具 CORSMiddleware 来处理这个问题。它的原理是通过在请求到达你的路由处理函数之前,拦截并添加正确的 HTTP 响应头(如 Access-Control-Allow-Origin)。

实战代码:配置 CORS

不要在每个路由上手动添加 Header,而是使用中间件全局处理。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 1. 定义允许的源列表
# 在生产环境中,请务必替换为具体的域名,而不是使用 "*" 通配符
# 例如: origins = ["https://your-frontend-domain.com"]
origins = [
    "http://localhost",
    "http://localhost:3000",
    "http://127.0.0.1",
    "http://127.0.0.1:3000",
]

# 2. 添加中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,           # 允许访问的源
    allow_credentials=True,          # 允许携带 Cookie
    allow_methods=["*"],             # 允许的方法 (GET, POST, PUT, DELETE, OPTIONS 等)
    allow_headers=["*"],             # 允许的请求头
)

@app.get("/")
async def root():
    return {"message": "Hello World with CORS enabled!"}

专家提示

第三章:构建生产级日志系统

日志是软件的“黑匣子”。在 Python 中,虽然 print 很方便,但它在生产环境中存在诸多缺陷(如无法区分级别、难以定向输出、性能较差)。我们需要使用 Python 内置的 logging 模块来构建结构化的日志系统。

1. 基础配置:格式化与输出

一个良好的日志应该包含时间戳、日志级别、文件位置以及具体的日志信息。

import logging
import sys

# 配置日志格式
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(name)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    handlers=[
        logging.StreamHandler(sys.stdout)  # 输出到控制台
    ]
)

logger = logging.getLogger(__name__)

# 测试日志
logger.info("系统启动,日志配置已加载")

2. 进阶:在 FastAPI 中实现请求日志记录

仅仅记录应用启动的日志是不够的。我们需要记录每一个进来的请求(URL、方法、耗时)以及每一个抛出的异常。FastAPI 的事件处理器 (Event Handlers)中间件 (Middleware) 是完成这个任务的最佳搭档。

方案 A:使用中间件记录请求耗时

中间件可以包裹每一个请求,计算处理时间。

import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

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_time) * 1000  # 转换为毫秒
        formatted_time = f"{process_time:.2f}ms"
        
        logger.info(
            f"{request.method} {request.url.path} - "
            f"Status: {response.status_code} - "
            f"Duration: {formatted_time}"
        )
        
        return response

方案 B:捕获全局异常

为了防止异常发生时只有默认的报错信息,我们可以注册一个全局异常处理器。

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    logger.error(f"HTTPException occurred: {exc.status_code} - {exc.detail} at {request.url.path}")
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

将这些组件整合到主应用中:

app = FastAPI()

# 添加 CORS (来自第二章)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"], 
    allow_methods=["*"], 
    allow_headers=["*"]
)

# 添加日志中间件
app.add_middleware(LoggingMiddleware)

# ... 路由定义 ...

3. 日志轮转 (Log Rotation)

在生产环境中,日志文件如果不加控制,会迅速占满磁盘空间。Python 的 logging.handlers 模块提供了 RotatingFileHandlerTimedRotatingFileHandler

代码示例(按大小切割):

from logging.handlers import RotatingFileHandler

file_handler = RotatingFileHandler(
    "app.log", 
    maxBytes=10*1024*1024,  # 10MB
    backupCount=5,          # 保留5个旧文件
    encoding="utf-8"
)
file_handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
logger.addHandler(file_handler)

第四章:终极整合与最佳实践

将 CORS 和日志结合在一起,我们就能构建一个健壮的 FastAPI 应用骨架。以下是完整的生产级模板代码:

import logging
import time
import sys
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

# --- 配置区域 ---

# 1. 日志配置
logger = logging.getLogger("FastAPI_Prod")
logger.setLevel(logging.INFO)

# 控制台输出
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
logger.addHandler(console_handler)

# 文件输出 (可选,需确保目录存在)
# file_handler = RotatingFileHandler("logs/app.log", maxBytes=10000000, backupCount=5)
# file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
# logger.addHandler(file_handler)

# 2. CORS 配置
origins = [
    "http://localhost:3000",
    # 生产环境域名
    # "https://www.yourdomain.com"
]

# --- 中间件类 ---

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_time) * 1000
        logger.info(f"{request.method} {request.url.path} - {response.status_code} - {process_time:.2f}ms")
        return response

# --- 应用初始化 ---

app = FastAPI(title="Production Ready API", version="1.0.0")

# 注册中间件(注意顺序:CORS 通常建议在较前的位置,但 Logging 可以包裹它)
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(LoggingMiddleware)

# --- 异常处理 ---

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    logger.error(f"Unhandled Exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal Server Error"},
    )

# --- 路由 ---

@app.get("/")
async def root():
    return {"message": "API is running with advanced logging and CORS"}

@app.get("/error")
async def trigger_error():
    # 触发一个异常来测试日志捕获
    raise HTTPException(status_code=400, detail="Bad Request Test")

# --- 启动命令 ---
# uvicorn main:app --reload

最佳实践总结

以上就是FastAPI优雅处理CORS跨域与日志记录的实战指南的详细内容,更多关于FastAPI处理CORS跨域的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文