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

Question

我正在尝试使用 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)

模式.py

from pydantic import BaseModel, Field

class Category(BaseModel):
    name: str
    parentid: int = None
    id: int
    class Config:
        orm_mode = True

主文件

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

标签： pythonsqlalchemyfastapipydantic

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
  }
]

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