FastAPI接收GET参数存入数据库时遇Pydantic及SQLAlchemy映射错误

问题背景 #

我是FastAPI新手，正在开发一款应用，需要接收根URL的GET请求参数并将其存入数据库。但找到的示例大多针对POST请求，尝试多种方案后出现两类错误：

• Pydantic错误（提示需要POST请求体数据）
• sqlalchemy.orm.exc.UnmappedInstanceError：提示“Class 'app.main.User' is not mapped”

原始代码 #

app = FastAPI()

SQLALCHEMY_DATABASE_URL = "sqlite:///./camping.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args = {"check_same_thread": False})
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True, index=True)
    authtoken = Column(String(500))
    admin_id = Column(Integer)
    app_id = Column(Integer)
    lang = Column(String(20))
    Base.metadata.create_all(bind=engine)
    
class User(BaseModel):
    authtoken: str
    admin_id: int
    app_id: int
    lang: str
    class Config:
        from_attributes = True

# Dependency
def get_db():
    db = session() 
    try:
        yield db
    finally:
        db.close()

@app.get("/", response_model=User)
async def root(db : Session = Depends(get_db),authtoken: Union[str, None] = None, action: Union[str, None] = None,admin_id: Union[int, None] = None, chain_id: Union[int, None] = None, app_id: Union[int, None] = None, lang: Union[str, None] = None): 
    bk=User(authtoken=authtoken, action=action,admin_id=admin_id, chain_id=chain_id,app_id=app_id,lang=lang)
    db.add(bk)
    db.commit()
    db.refresh(bk)
    return User(**bk.dict())

修改后代码及报错 #

修改后的代码：

@app.get("/", response_model=User)
async def root(b1: User,db : Session = Depends(get_db),authtoken: Union[str, None] = None, action: Union[str, None] = None,admin_id: Union[int, None] = None, chain_id: Union[int, None] = None, app_id: Union[int, None] = None, lang: Union[str, None] = None): #,b1: User, db: Session = Depends(get_db)):
    bk=User(id=b1.id, authtoken=b1.authtoken, action=b1.action,admin_id=b1.admin_id, chain_id=b1.chain_id,app_id=b1.app_id,lang=b1.lang)
    db.add(bk)
    db.commit()
    db.refresh(bk)
    return User(**db.dict())

报错信息（中文翻译：请求体缺失，字段必填）：

{"detail":[{"type":"missing","loc":["body"],"msg":"Field required","input":null,"url":"https://errors.pydantic.dev/2.1.2/v/missing"}]}

解决方案 #

核心问题梳理

1. 类名冲突：同时定义了SQLAlchemy映射类User和Pydantic模型User，导致代码中混淆两种不同类型的对象——试图把Pydantic模型存入数据库，而非SQLAlchemy的映射实例，直接引发UnmappedInstanceError。
2. GET请求误用请求体：修改后的代码中b1: User参数会被FastAPI默认解析为POST请求体，但GET请求没有请求体，因此触发Pydantic的"请求体缺失"错误。
3. 数据库初始化时机错误：Base.metadata.create_all(bind=engine)放在User类内部，会在类定义时就执行，可能导致表创建不完整或重复执行。

修正后的完整代码

from fastapi import FastAPI, Depends
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from typing import Union

app = FastAPI()

SQLALCHEMY_DATABASE_URL = "sqlite:///./camping.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# SQLAlchemy 数据库映射类
class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True, index=True)
    authtoken = Column(String(500))
    admin_id = Column(Integer)
    app_id = Column(Integer)
    lang = Column(String(20))

# Pydantic 响应模型，重命名避免冲突
class UserResponse(BaseModel):
    authtoken: str
    admin_id: int
    app_id: int
    lang: str
    
    class Config:
        from_attributes = True

# 所有模型定义完成后，创建数据库表
Base.metadata.create_all(bind=engine)

# 数据库依赖
def get_db():
    db = SessionLocal() 
    try:
        yield db
    finally:
        db.close()

@app.get("/", response_model=UserResponse)
async def root(
    db: Session = Depends(get_db),
    authtoken: Union[str, None] = None,
    admin_id: Union[int, None] = None,
    app_id: Union[int, None] = None,
    lang: Union[str, None] = None
): 
    # 用GET参数构建SQLAlchemy的User实例
    db_user = User(
        authtoken=authtoken,
        admin_id=admin_id,
        app_id=app_id,
        lang=lang
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    # 用Pydantic模型序列化数据库对象返回
    return UserResponse.from_orm(db_user)

关键修改说明

• 重命名Pydantic模型：将原User Pydantic模型改为UserResponse，彻底避免和SQLAlchemy的User类冲突。
• 调整表创建时机：将Base.metadata.create_all(bind=engine)移到所有模型定义之后，确保所有表都能正确创建。
• 直接使用GET参数构建数据库对象：不再试图用请求体接收数据，直接从GET查询参数中取值，创建SQLAlchemy的User实例存入数据库。
• 正确序列化响应：使用UserResponse.from_orm(db_user)将数据库对象转换为Pydantic模型返回，符合FastAPI的响应规范。

测试方式

启动服务后，访问如下格式的URL即可测试：

http://localhost:8000/?authtoken=my_test_token&admin_id=123&app_id=456&lang=zh-CN

内容的提问来源于stack exchange，提问作者het.oosten