从 Sqlalchemy 到 Pydantic 的映射问题 - from_orm 失败

Question

我正在尝试使用 from_orm 方法将我的结果字典从 sqlalchemy 自动获取到 Fastapi 的 Pydantic 输出到映射，但我总是收到验证错误。

File "pydantic\main.py", line 508, in pydantic.main.BaseModel.from_orm pydantic.error_wrappers.ValidationError: 2 validation errors for Category name field required (type=value_error.missing) id field required (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

或者是否有 another/better 方法将 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}]]

Models.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

Answer 1

这是我自己的新手，所以不能保证最好的答案，但我注意到如果你只是简单但在模式中是可选的，它就可以工作。

`

 class Category(BaseModel):
        name: Optional[str]
        parentid: int = None
        id: Optional[int]

        class Config:
            orm_mode = True
`

仍在响应中返回该信息：

[
  {
    "name": "games",
    "parentid": null,
    "id": 1
  },
  {
    "name": "computer",
    "parentid": null,
    "id": 2
  },
  {
    "name": "household",
    "parentid": null,
    "id": 3
  },
  {
    "name": "test",
    "parentid": null,
    "id": 10
  }
]

可能仍然存在某种验证错误，但目前看来是一个可用的解决方法。

Answer 2

我刚刚遇到了同样的问题。我认为它与 pydantic nonethless 有关。请查看此 link 了解更多信息 https://github.com/samuelcolvin/pydantic/issues/506。

但是改变了我的模型：

class Student(BaseModel):
    id: Optional[int] --- changed to optional
    name: Optional [str]
    surname: Optional [str]
    email: Optional [str]

错误验证消失。这是一个有趣的错误 - 假设我的数据库中的条目仍然用值更新......我也是 fastAPI 的新手所以解决方法和错误现在没有真正意义......但是是的它有效。谢谢

从 Sqlalchemy 到 Pydantic 的映射问题 - from_orm 失败

Mapping issues from Sqlalchemy to Pydantic - from_orm failed

python

sqlalchemy

pydantic

fastapi

Models.py

Schemas.py

main.py