【发布时间】：2021-01-08 19:29:26
【问题描述】：
我正在尝试使用 from_orm 方法将我的结果字典从 sqlalchemy 自动获取到 Fastapi 的 Pydantic 输出到映射，但我总是遇到验证错误。
文件“pydantic\main.py”，第 508 行，在 pydantic.main.BaseModel.from_orm pydantic.error_wrappers.ValidationError：类别的 2 个验证错误 姓名 必填字段（type=value_error.missing） ID 必填字段（type=value_error.missing）
如果我自己使用 Pydantic 模式创建对象并将它们添加到列表中，则该方法有效。 我必须改变什么才能使 from_orm 工作？ 我是否可能遗漏了文档中的某些内容？ https://pydantic-docs.helpmanual.io/usage/models/#orm-mode-aka-arbitrary-class-instances https://fastapi.tiangolo.com/tutorial/sql-databases/#use-pydantics-orm_mode
或者是否有另一种/更好的方法可以将 ResultProxy 转换为 Pydantic 输出？
我从数据库方法得到的输出如下：
[{'id': 1, 'name': 'games', 'parentid': None}, {'id': 2, 'name': 'computer', 'parentid': None}, {'id': 3, 'name': 'household', 'parentid': None}, {'id': 10, 'name': 'test', 'parentid': None}]]
模型.py
>from sqlalchemy import BigInteger, Column, DateTime, ForeignKey, Integer, Numeric, String, Text, text, Table
from sqlalchemy.orm import relationship, mapper
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
metadata = Base.metadata
category = Table('category', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(200)),
Column('parentid', Integer),
)
class Category(object):
def __init__(self, cat_id, name, parentid):
self.id = cat_id
self.name = name
self.parentid = parentid
mapper(Category, category)
Schemas.py
>from pydantic import BaseModel, Field
class Category(BaseModel):
name: str
parentid: int = None
id: int
class Config:
orm_mode = True
main.py
>def result_proxy_to_Dict(results: ResultProxy):
d, a = {}, []
for rowproxy in results:
# rowproxy.items() returns an array like [(key0, value0), (key1, value1)]
for column, value in rowproxy.items():
# build up the dictionary
d = {**d, **{column: value}}
a.append(d)
return a
def crud_read_cat(db: Session) -> dict:
# records = db.query(models.Category).all()
#query = db.query(models.Category).filter(models.Category.parentid == None)
s = select([models.Category]). \
where(models.Category.parentid == None)
result = db.execute(s)
#print(type(result))
#print(result_proxy_to_Dict(result))
#results = db.execute(query)
# result_set = db.execute("SELECT id, name, parentid FROM public.category;")
# rint(type(result_set))
# for r in result_set:
# print(r)
# return [{column: value for column, value in rowproxy.items()} for rowproxy in result_set]
# return await databasehelper.database.fetch_all(query)
return result_proxy_to_Dict(result)
#return results
@router.get("/category/", response_model=List[schemas.Category], tags=["category"])
async def read_all_category(db: Session = Depends(get_db)):
categories = crud_read_cat(db)
context = []
print(categories)
co_model = schemas.Category.from_orm(categories)
# print(co_model)
for row in categories:
print(row)
print(row.get("id", None))
print(row.get("name", None))
print(row.get("parentid", None))
tempcat = schemas.Category(id=row.get("id", None), name=row.get("name", None),
parentid=row.get("parentid", None))
context.append(tempcat)
#for dic in [dict(r) for r in categories]:
# print(dic)
# print(dic.get("category_id", None))
# print(dic.get("category_name", None))
# print(dic.get("category_parentid", None))
# tempcat = schemas.Category(id=dic.get("category_id", None), name=dic.get("category_name", None),
# parentid=dic.get("category_parentid", None))
# context.append(tempcat)
return context
【问题讨论】：
标签： python sqlalchemy fastapi pydantic