如何以灵活的方式将嵌套的 pydantic 模型用于 sqlalchemy

Posted 2023-02-15

【中文标题】如何以灵活的方式将嵌套的 pydantic 模型用于 sqlalchemy

【英文标题】：How to use nested pydantic models for sqlalchemy in a flexible way

【发布时间】：2021-02-01 10:39:21

【问题描述】：

from fastapi import Depends, FastAPI, HTTPException, Body, Request
from sqlalchemy import create_engine, Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, sessionmaker, relationship
from sqlalchemy.inspection import inspect
from typing import List, Optional
from pydantic import BaseModel
import json

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args="check_same_thread": False
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
app = FastAPI()


# sqlalchemy models

class RootModel(Base):
    __tablename__ = "root_table"
    id = Column(Integer, primary_key=True, index=True)
    someRootText = Column(String)
    subData = relationship("SubModel", back_populates="rootData")


class SubModel(Base):
    __tablename__ = "sub_table"
    id = Column(Integer, primary_key=True, index=True)
    someSubText = Column(String)
    root_id = Column(Integer, ForeignKey("root_table.id"))
    rootData = relationship("RootModel", back_populates="subData")


# pydantic models/schemas
class SchemaSubBase(BaseModel):
    someSubText: str

    class Config:
        orm_mode = True


class SchemaSub(SchemaSubBase):
    id: int
    root_id: int

    class Config:
        orm_mode = True


class SchemaRootBase(BaseModel):
    someRootText: str
    subData: List[SchemaSubBase] = []

    class Config:
        orm_mode = True


class SchemaRoot(SchemaRootBase):
    id: int

    class Config:
        orm_mode = True


class SchemaSimpleBase(BaseModel):
    someRootText: str

    class Config:
        orm_mode = True


class SchemaSimple(SchemaSimpleBase):
    id: int

    class Config:
        orm_mode = True


Base.metadata.create_all(bind=engine)


# database functions (CRUD)

def db_add_simple_data_pydantic(db: Session, root: SchemaRootBase):
    db_root = RootModel(**root.dict())
    db.add(db_root)
    db.commit()
    db.refresh(db_root)
    return db_root


def db_add_nested_data_pydantic_generic(db: Session, root: SchemaRootBase):

    # this fails:
    db_root = RootModel(**root.dict())
    db.add(db_root)
    db.commit()
    db.refresh(db_root)
    return db_root


def db_add_nested_data_pydantic(db: Session, root: SchemaRootBase):

    # start: hack: i have to manually generate the sqlalchemy model from the pydantic model
    root_dict = root.dict()
    sub_dicts = []

    # i have to remove the list form root dict in order to fix the error from above
    for key in list(root_dict):
        if isinstance(root_dict[key], list):
            sub_dicts = root_dict[key]
            del root_dict[key]

    # now i can do it
    db_root = RootModel(**root_dict)
    for sub_dict in sub_dicts:
        db_root.subData.append(SubModel(**sub_dict))

    # end: hack
    db.add(db_root)
    db.commit()
    db.refresh(db_root)
    return db_root


def db_add_nested_data_nopydantic(db: Session, root):
    print(root)
    sub_dicts = root.pop("subData")
    print(sub_dicts)
    db_root = RootModel(**root)

    for sub_dict in sub_dicts:
        db_root.subData.append(SubModel(**sub_dict))
    db.add(db_root)
    db.commit()
    db.refresh(db_root)

    # problem
    """
    if I would now "return db_root", the answer would be of this:
    
        "someRootText": "string",
        "id": 24
    

    and not containing "subData"
    therefore I have to do the following.
    Why?

    """
    from sqlalchemy.orm import joinedload

    db_root = (
        db.query(RootModel)
            .options(joinedload(RootModel.subData))
            .filter(RootModel.id == db_root.id)
            .all()
    )[0]
    return db_root


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


@app.post("/addNestedModel_pydantic_generic", response_model=SchemaRootBase)
def addSipleModel_pydantic_generic(root: SchemaRootBase, db: Session = Depends(get_db)):
    data = db_add_simple_data_pydantic(db=db, root=root)
    return data


@app.post("/addSimpleModel_pydantic", response_model=SchemaSimpleBase)
def add_simple_data_pydantic(root: SchemaSimpleBase, db: Session = Depends(get_db)):
    data = db_add_simple_data_pydantic(db=db, root=root)
    return data


@app.post("/addNestedModel_nopydantic")
def add_nested_data_nopydantic(root=Body(...), db: Session = Depends(get_db)):
    data = db_add_nested_data_nopydantic(db=db, root=root)
    return data


@app.post("/addNestedModel_pydantic", response_model=SchemaRootBase)
def add_nested_data_pydantic(root: SchemaRootBase, db: Session = Depends(get_db)):
    data = db_add_nested_data_pydantic(db=db, root=root)
    return data

说明

我的问题是：

如何以通用方式从嵌套的 pydantic 模型（或 python dicts）制作嵌套的 sqlalchemy 模型，并“一次性”将它们写入数据库。

我的示例模型名为RootModel，在subData 键中有一个名为“子模型”的子模型列表。

有关 pydantic 和 sqlalchemy 的定义，请参见上文。

示例： 用户提供一个嵌套的 json 字符串：

  "someRootText": "string",
  "subData": [
    
      "someSubText": "string"
    
  ]

打开浏览器并调用端点/docs。 您可以使用所有端点并从上面发布 json 字符串。

/addNestedModel_pydantic_generic

当您调用端点 /addNestedModel_pydantic_generic 时，它将失败，因为 sqlalchemy 无法直接从 pydantic 嵌套模型创建嵌套模型： AttributeError: 'dict' object has no attribute '_sa_instance_state'

​/addSimpleModel_pydantic

对于非嵌套模型，它可以工作。

其余端点正在展示解决嵌套模型问题的“hacks”。

/addNestedModel_pydantic

在此端点中生成根模型，并使用 pydantic 模型以非通用方式循环生成子模型。

/addNestedModel_pydantic

在此端点中生成根模型，并使用 python dicts 以非通用方式循环生成子模型。

我的解决方案只是 hack，我想要 一种从 pydantic（首选）或 python dict 创建嵌套 sqlalchemy 模型的通用方法。

环境

操作系统：Windows， FastAPI 版本：0.61.1 Python 版本：Python 3.8.5 sqlalchemy：1.3.19 pydantic : 1.6.1

【问题讨论】：

找到解决办法了吗？

这能回答你的问题吗？ List of object attributes in pydantic model

【参考方案1】：

我还没有在 pydantic/SQLAlchemy 中找到一个很好的内置方法来执行此操作。我是如何解决的：我给每个嵌套的 pydantic 模型一个 Meta 类，其中包含相应的 SQLAlchemy 模型。像这样：

from pydantic import BaseModel
from models import ChildDBModel, ParentDBModel

class ChildModel(BaseModel):
    some_attribute: str = 'value'
    class Meta:
        orm_model = ChildDBModel

class ParentModel(BaseModel):
    child: SubModel

这让我可以编写一个循环遍历 pydantic 对象并将子模型转换为 SQLAlchemy 模型的通用函数：

def is_pydantic(obj: object):
    """Checks whether an object is pydantic."""
    return type(obj).__class__.__name__ == "ModelMetaclass"


def parse_pydantic_schema(schema):
    """
        Iterates through pydantic schema and parses nested schemas
        to a dictionary containing SQLAlchemy models.
        Only works if nested schemas have specified the Meta.orm_model.
    """
    parsed_schema = dict(schema)
    for key, value in parsed_schema.items():
        try:
            if isinstance(value, list) and len(value):
                if is_pydantic(value[0]):
                    parsed_schema[key] = [schema.Meta.orm_model(**schema.dict()) for schema in value]
            else:
                if is_pydantic(value):
                    parsed_schema[key] = value.Meta.orm_model(**value.dict())
        except AttributeError:
            raise AttributeError("Found nested Pydantic model but Meta.orm_model was not specified.")
    return parsed_schema

parse_pydantic_schema 函数返回 pydantic 模型的字典表示，其中子模型被Meta.orm_model 中指定的相应 SQLAlchemy 模型替换。您可以使用此返回值一次性创建父 SQLAlchemy 模型：

parsed_schema = parse_pydantic_schema(parent_model)  # parent_model is an instance of pydantic ParentModel 
new_db_model = ParentDBModel(**parsed_schema)
# do your db actions/commit here

如果您愿意，您甚至可以扩展它以自动创建父模型，但这需要您还为所有 pydantic 模型指定 Meta.orm_model。

【讨论】：

非常好的实现@Daan。也许您可以调整它以使用 @root_validator 而不必手动调用该函数。

【参考方案2】：

不错的函数@dann，对于超过两层的嵌套，您可以使用这个递归函数：

def pydantic_to_sqlalchemy_model(schema):
    """
    Iterates through pydantic schema and parses nested schemas
    to a dictionary containing SQLAlchemy models.
    Only works if nested schemas have specified the Meta.orm_model.
    """
    parsed_schema = dict(schema)
    for key, value in parsed_schema.items():
        try:
            if isinstance(value, list) and len(value) and is_pydantic(value[0]):
                parsed_schema[key] = [
                    item.Meta.orm_model(**pydantic_to_sqlalchemy_model(item))
                    for item in value
                ]
            elif is_pydantic(value):
                parsed_schema[key] = value.Meta.orm_model(
                    **pydantic_to_sqlalchemy_model(value)
                )
        except AttributeError:
            raise AttributeError(
                f"Found nested Pydantic model in schema.__class__ but Meta.orm_model was not specified."
            )
    return parsed_schema

谨慎使用！是你有一个循环嵌套它会永远循环。

然后像这样称呼你数据转换器：

def create_parent(db: Session, parent: Parent_pydantic_schema):
    db_parent = Parent_model(**pydantic_to_sqlalchemy_model(intent))
    db.add(db_parent)
    db.commit()
    return db_parent

【参考方案3】：

使用验证器要简单得多：

SQLAlchemy 模型.py：

class ChildModel(Base):
    __tablename__ = "Child"
    name: str = Column(Unicode(255), nullable=False, primary_key=True)


class ParentModel(Base):
    __tablename__ = "Parent"
    some_attribute: str = Column(Unicode(255))
    children = relationship("Child", lazy="joined", cascade="all, delete-orphan")

    @validates("children")
    def adjust_children(self, _, value) -> ChildModel:
        """Instantiate Child object if it is only plain string."""
        if value and isinstance(value, str):
            return ChildModel(some_attribute=value)
        return value

Pydantic schema.py:

class Parent(BaseModel):
    """Model used for parents."""

    some_attribute: str
    children: List[str] = Field(example=["foo", "bar"], default=[])

    @validator("children", pre=True)
    def adjust_chidlren(cls, children):
        """Convert to plain string if it is a Child object."""
        if children and not isinstance(next(iter(children), None), str):
            return [child["name"] for child in children]
        return children