问题描述
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,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,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 dict）中创建嵌套的sqlalchemy模型，并“一次性”将其写入数据库。
我的示例模型称为RootModel，并且在subData键中有一个称为“子模型”的子模型列表。
请参见上面的pydantic和sqlalchemy定义。
示例： 用户提供了一个嵌套的json字符串：
{
"someRootText": "string","subData": [
{
"someSubText": "string"
}
]
}
打开浏览器并调用端点/docs。
您可以使用所有端点，并从上方发布json字符串。
当调用端点/ addNestedModel_pydantic_generic时，它将失败，因为sqlalchemy无法直接从pydantic嵌套模型创建嵌套模型：
AttributeError: 'dict' object has no attribute '_sa_instance_state'
使用非嵌套模型，它可以工作。
其余端点显示“ hacks”以解决嵌套模型的问题。
/ addNestedModel_pydantic在此端点中生成根模型，并以非泛型方式对子模型和pydantic模型进行循环。
/ addNestedModel_pydantic在此端点中，使用python dicts以非通用的方式生成根模型并使用循环将子模型和化。
我的解决方案只是黑客，我想一种从pydantic（首选）或python dict创建嵌套sqlalchemy模型的通用方式。
环境
- 操作系统：Windows，
- FastAPI版本：0.61.1
- Python版本：Python 3.8.5
- sqlalchemy：1.3.19
- pydantic：1.6.1
解决方法
暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！
如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@）