FastAPI访问/products端点提示request字段缺失问题排查

我正在跟随FastAPI官方文档的简单教程开发，项目原本已使用SQLAlchemy，新增FastAPI依赖后尝试运行，代码如下：

import re
import json
import copy
import traceback
import urllib.parse
from models import *
from mangum import Mangum
from datetime import datetime
from sqlalchemy.orm import Session
from sqlalchemy import *
from sqlalchemy import create_engine
from collections import defaultdict
from fastapi import FastAPI, Depends, Request, Response

app = FastAPI()


@app.middleware("http")
async def db_session_middleware(request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        engine = create_engine(
            "some db"
        )
        base.metadata.create_all(engine)
        request.state.db = Session(engine)
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response


def get_db(request):
    return request.state.db


@app.get("/")
def get_root():
    return {"Status": "OK"}


@app.get("/products/{sku}")
def get_product(sku, db=Depends(get_db)):
    pass


@app.get("/products", status_code=200)
def get_products(page: int = 1, page_size: int = 50, db: Session = Depends(get_db)):
    try:
        result, sku_list = [], []
        for row in (
            db.query(Product, Image)
            .filter(Product.sku == Image.sku)
            .limit(page_size)
            .offset(page * page_size)
        ):
            if row[0].sku not in sku_list:
                result.append(
                    {
                        "sku": row[0].sku,
                        "brand": row[0].brand,
                        "image": row[1].image_url,
                        "title": row[0].product_title,
                        "price": row[0].original_price,
                        "reviewCount": row[0].total_reviews,
                        "rating": row[0].overall_rating,
                    }
                )
                sku_list.append(row[0].sku)

        print(f"Result: {result}")
        return {"body": {"message": "Success", "result": result}, "statusCode": 200}
    except Exception as err:
        print(traceback.format_exc(err))
        return {
            "body": {"message": "Failure", "result": traceback.format_exc(err)},
            "statusCode": 500,
        }

通过URL http://127.0.0.1:8000/products?page=1&page_size=100 访问products端点时，收到如下响应：

{"detail":[{"loc":["query","request"],"msg":"field required","type":"value_error.missing"}]}

不清楚响应中提示的“query”和“request”缺失是什么意思，请问哪里出错了？

核心原因 #

get_db函数的参数未正确声明依赖：FastAPI处理依赖注入时，会自动解析函数的所有参数。你的get_db函数定义了request参数，但没有将其标记为依赖项，FastAPI会把它当作查询参数去解析，所以才会提示查询参数中的request字段缺失。

解决方法 #

有两种可行的修改方式：

方式一：为request参数添加Depends()标记

修改get_db函数，让FastAPI自动注入Request对象：

from fastapi import Request, Depends

def get_db(request: Request = Depends()):
    return request.state.db

方式二：直接在依赖中注入Request

也可以在路由的依赖声明中直接指定Request注入，但更推荐方式一的写法：

# 修改get_db函数
def get_db(request: Request):
    return request.state.db

# 在路由中使用时
@app.get("/products", status_code=200)
def get_products(page: int = 1, page_size: int = 50, db: Session = Depends(lambda req: get_db(req))):
    # 函数逻辑不变

额外优化建议 #

1. 数据库引擎优化：不要在中间件中每次请求都创建create_engine，应该将引擎定义在全局作用域，避免重复创建连接池，提升性能：

# 放在文件顶部，全局创建一次
engine = create_engine("some db")
base.metadata.create_all(engine)

@app.middleware("http")
async def db_session_middleware(request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        request.state.db = Session(engine)
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response

2. 分页逻辑修正：当前的offset(page * page_size)会导致第一页丢失数据（比如page=1时会跳过前100条），应该修改为：

.offset((page - 1) * page_size)

内容的提问来源于stack exchange，提问作者Amol Borkar