FastAPI+SQLModel查询Item报错:'SQLModelMetaclass'对象不可迭代
问题场景
我有一个FastAPI应用,需要提供GET接口从PostgreSQL 13数据库中获取所有Item数据。相关代码如下:
数据库模型(其他应用定义)
class ItemDb(SQLModel, table=True): __tablename__ = "items" id: str = Field( default_factory=uuid.uuid4, primary_key=True) name: str
class GadgetDb(SQLModel, table=True): __tablename__ = "gadgets" id: str = Field( default_factory=uuid.uuid4, primary_key=True) item_id: str = Field(..., foreign_key="items.id") name: str
返回数据模型
class Item(SQLModel): id: str key_name: str
CRUD类及方法
class CRUDItem(CRUDBase[Item, None, None]): def list(self, db: Session) -> List[Item]: statement = select(self.model) results = db.exec(statement) return results.all() item = CRUDItem(Item)
端点函数
from app.crud.crud_item import item @router.get( "/", response_model=List[Item], status_code=200, response_model_exclude_unset=True, ) def list_items( session: Session = Depends(get_session), ) -> Any: items = item.list(session) return items
CRUDBase基类
ModelType = TypeVar("ModelType", bound=Base) CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel) UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel) class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): def __init__(self, model: Type[ModelType]): """ CRUD object with default methods to Create, Read, Update, Delete (CRUD). **Parameters** * `model`: A SQLAlchemy model class * `schema`: A Pydantic model (schema) class """ self.model = model def get(self, db: Session, id: Any) -> Optional[ModelType]: return db.query(self.model).filter(self.model.id == id).first() def list( self, db: Session, *, skip: int = 0, limit: int = 100 ) -> List[ModelType]: return db.query(self.model).offset(skip).limit(limit).all() def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType: obj_in_data = jsonable_encoder(obj_in) db_obj = self.model(**obj_in_data) # type: ignore db.add(db_obj) db.commit() db.refresh(db_obj) return db_obj
触发的错误
调用接口时出现TypeError: 'SQLModelMetaclass' object is not iterable,错误栈:
INFO: 172.30.0.1:51432 - "GET /api/v1/items/ HTTP/1.1" 500 Internal Server Error ERROR: Exception in ASGI application Traceback (most recent call last): File "/usr/local/lib/python3.9/site-packages/uvicorn/protocols/http/httptools_impl.py", line 404, in run_asgi result = await app( # type: ignore[func-returns-value] File "/usr/local/lib/python3.9/site-packages/uvicorn/middleware/proxy_headers.py", line 78, in __call__ return await self.app(scope, receive, send) File "/usr/local/lib/python3.9/site-packages/fastapi/applications.py", line 270, in __call__ await super().__call__(scope, receive, send) File "/usr/local/lib/python3.9/site-packages/starlette/applications.py", line 124, in __call__ await self.middleware_stack(scope, receive, send) File "/usr/local/lib/python3.9/site-packages/starlette/middleware/errors.py", line 184, in __call__ raise exc File "/usr/local/lib/python3.9/site-packages/starlette/middleware/errors.py", line 162, in __call__ await self.app(scope, receive, _send) File "/usr/local/lib/python3.9/site-packages/starlette/middleware/cors.py", line 84, in __call__ await self.app(scope, receive, send) File "/usr/local/lib/python3.9/site-packages/starlette/middleware/exceptions.py", line 75, in __call__ raise exc File "/usr/local/lib/python3.9/site-packages/starlette/middleware/exceptions.py", line 64, in __call__ await self.app(scope, receive, sender) File "/usr/local/lib/python3.9/site-packages/fastapi/middleware/asyncexitstack.py", line 21, in __call__ raise e File "/usr/local/lib/python3.9/site-packages/fastapi/middleware/asyncexitstack.py", line 18, in __call__ await self.app(scope, receive, send) File "/usr/local/lib/python3.9/site-packages/starlette/routing.py", line 680, in __call__ await route.handle(scope, receive, send) File "/usr/local/lib/python3.9/site-packages/starlette/routing.py", line 275, in handle await self.app(scope, receive, send) File "/usr/local/lib/python3.9/site-packages/starlette/routing.py", line 65, in app response = await func(request) File "/usr/local/lib/python3.9/site-packages/fastapi/routing.py", line 231, in app raw_response = await run_endpoint_function( File "/usr/local/lib/python3.9/site-packages/fastapi/routing.py", line 162, in run_endpoint_function return await run_in_threadpool(dependant.call, **values) File "/usr/local/lib/python3.9/site-packages/starlette/concurrency.py", line 41, in run_in_threadpool return await anyio.to_thread.run_sync(func, *args) File "/usr/local/lib/python3.9/site-packages/anyio/to_thread.py", line 31, in run_sync return await get_asynclib().run_sync_in_worker_thread( File "/usr/local/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread return await future File "/usr/local/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 867, in run result = context.run(func, *args) File "/app/./app/api/api_v1/endpoints/items.py", line 29, in list_items results = session.exec(select(Item)).all() File "/usr/local/lib/python3.9/site-packages/sqlmodel/sql/expression.py", line 450, in select return SelectOfScalar._create(*entities, **kw) # type: ignore File "/usr/local/lib/python3.9/site-packages/sqlalchemy/sql/selectable.py", line 5309, in _create return cls.create_legacy_select(*args, **kw) File "<string>", line 2, in create_legacy_select File "/usr/local/lib/python3.9/site-packages/sqlalchemy/util/deprecations.py", line 402, in warned return fn(*args, **kwargs) File "/usr/local/lib/python3.9/site-packages/sqlalchemy/sql/selectable.py", line 5165, in create_legacy_select self._raw_columns = [ TypeError: 'ModelMetaclass' object is not iterable
错误原因
核心问题是混淆了数据库模型与返回数据模型的作用:
- CRUD操作需要操作绑定数据库表的
ItemDb(带table=True的SQLModel),但代码中传入CRUDItem的是仅用于返回数据的Item模型,该模型未关联数据库表,无法被SQLAlchemy的select()函数识别为有效查询实体。 - 从错误栈可见,执行
select(Item)时触发异常,因为Item不是可被SQLAlchemy查询的数据库模型类,导致框架尝试迭代其列定义时失败。
解决方案
明确区分数据库模型和返回模型的职责,按以下步骤修改:
1. 修正CRUD类的初始化参数
将CRUDItem的实例化参数改为ItemDb,确保CRUD操作针对数据库表:
# 修改crud_item.py中的代码 class CRUDItem(CRUDBase[ItemDb, None, None]): def list(self, db: Session) -> List[ItemDb]: statement = select(self.model) results = db.exec(statement) return results.all() item = CRUDItem(ItemDb) # 传入数据库模型ItemDb而非返回模型Item
2. 端点函数中转换数据格式
从数据库查询到ItemDb实例后,将其转换为返回模型Item,注意字段映射(数据库中的name对应返回模型的key_name):
# 修改端点函数 @router.get( "/", response_model=List[Item], status_code=200, response_model_exclude_unset=True, ) def list_items( session: Session = Depends(get_session), ) -> Any: db_items = item.list(session) # 手动映射字段转换模型 return [Item(id=item.id, key_name=item.name) for item in db_items]
3. 可选:用SQLModel简化模型转换
如果字段映射较多,可给返回模型字段添加别名,使用from_orm方法自动转换:
# 修改返回模型Item class Item(SQLModel): id: str key_name: str = Field(alias="name") # 别名对应数据库表的name字段 # 端点中简化转换逻辑 return [Item.from_orm(item) for item in db_items]
修改完成后,CRUD操作会正确查询items表,再将结果转换为指定的返回模型,接口即可正常返回数据。
内容的提问来源于stack exchange,提问作者lr_optim
相关产品推荐
相关产品推荐

