如何灵活地使用嵌套的简单模型进行SQLALCHIZY [英] How to use nested pydantic models for sqlalchemy in a flexible way

问题描述

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

说明

我的问题是：

如何以通用方式从嵌套的简单模型(或Python字典)创建嵌套的SQLALChemy模型，并将它们写入到&Quot；One Sort&Quot；中的数据库。

我的示例模型名为RootModel，并且在subData键中有一个名为&subModels"；的子模型列表。

请参阅上面的pydtic和sqlalChemy定义。

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

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

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

/addNestedModel_PYDANIC_GENERIC 当您调用端点/addNestedModel_PYDANIC_GENERIC时，它将失败，因为SQLalChemy不能直接从PYDNIC嵌套模型创建嵌套模型： AttributeError: 'dict' object has no attribute '_sa_instance_state'

​/addSimpleModel_PYDANIC

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

其余终结点正在显示用于解决嵌套模型问题的&q；hack&q；。

/addNestedModel_PYDANIC

在此端点中，以一种非泛型的方式生成根模型，并以一种非泛型的方式使用子模型来生成子模型。

/addNestedModel_PYDANIC

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

我的解决方案只是一些技巧，我希望有一种通用的方法来创建嵌套的SQLALCHEMY模型，要么是从pydual(首选)，要么是从python词典。

环境

• 操作系统：Windows
• FastAPI版本：0.61.1
• PYTHON版本：PYTHON3.8.5
• SqlalChemy：1.3.19
• PUDANIC：1.6.1

推荐答案

我还没有找到一种很好的内置方法来实现这一点。我是如何解决这个问题的：我为每个嵌套的简单模型提供了一个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

这使我能够编写一个泛型函数，该函数循环遍历PYDANIC对象，并将子模型转换为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函数返回简单模型的字典表示，其中子模型被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

如果需要，您甚至可以将其扩展为还自动创建父模型，但这还需要您为所有二进制模型指定Meta.orm_model。

这篇关于如何灵活地使用嵌套的简单模型进行SQLALCHIZY的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！