Fastapi: How to do flexibly use nested pydantic models for sqlalchemy ORM

[j-gimbel]

First check

• [x ] I added a very descriptive title to this issue.
• [ x] I used the GitHub search to find a similar issue and didn't find it.
• [ x] I searched the FastAPI documentation, with the integrated search.
• [x ] I already searched in Google "How to X in FastAPI" and didn't find any information.
• [ x] I already read and followed all the tutorial in the docs and didn't find an answer.
• [ x] I already checked if it is not related to FastAPI but to Pydantic.
• [x ] I already checked if it is not related to FastAPI but to Swagger UI.
• [ x] I already checked if it is not related to FastAPI but to ReDoc.
• [ x] After submitting this, I commit to one of:
  
  
  
  • Read open issues with questions until I find 2 issues where I can help someone and add a comment to help there.
  
  
  • I already hit the "watch" button in this repository to receive notifications and I commit to help at least 2 people that ask questions in the future.
  
  
  • Implement a Pull Request for a confirmed bug.
  
  

Example

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"
    there for 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 add_nested_data_pydantic_generic(root: SchemaRootBase, db: Session = Depends(get_db)):
    data = db_add_nested_data_pydantic_generic(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

Description

My Question is:

How to make nested sqlalchemy models from nested pydantic models (or python dicts) in a generic way and write them to the datase in "one shot".

My example model is called "root model" and has a list of submodels called "sub models" in "subData" key.
Please see above for pydantic and sql alchemy definitions.

Example:
The user provides a nested json string:

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

Open the browser and call the endpoint /docs.
You can play around with all endpoints and POST the json string from above.

/addNestedModel_pydantic_generic

When you call the endpoint /addNestedModel_pydantic_generic it will fail, because sqlalchemy cannot create the nested model from pydantic nested model directly:
AttributeError: 'dict' object has no attribute '_sa_instance_state'

​/addSimpleModel_pydantic

With a non-nested model it works.

The remaining endpoints are showing "hacks" to solve the problem of nested models.

/addNestedModel_pydantic

In this endpoint is generate the root model and andd the submodels with a loop in a non-generic way with pydantic models.

/addNestedModel_pydantic

In this endpoint is generate the root model and andd the submodels with a loop in a non-generic way with python dicts.

My solutions are only hacks, I want a generic way to create nested sqlalchemy models either from pydantic (preferred) or from a python dict.

Environment

• OS: Windows,
• FastAPI Version : 0.61.1
• Python version: Python 3.8.5
• sqlalchemy: 1.3.19
• pydantic : 1.6.1

[ycd]

Have you tried @tiangolo's Pydantic SQLAlchemy ?

[j-gimbel]

Thank you for the contribution, but your proposal is for converting sqlalchemy model to pydantic, I need a way to convert a nested model from pydantic to sqlalchemy.

[j-gimbel]

I am thinking about a recursive function to do what I want. I currently need an external "maping" dict that translates the pydantic sub class to the related sqlalchemy class by their name. Would be helpful if someone could provide an example of a recursive function for two dicts. Added difficulty that sub model could be lists or direct keys.

[ycd]

@j-gimbel I'll play with this more but this is what I came up with.

# Too lazy to delete unused imports :( 
from typing import Optional, Union, List, Dict, Any, Mapping, Type, TypeVar, Generic


def pydantic_to_sqlalchemy(schema: pydantic.main.ModelMetaclass) -> Any:
    __fields_dict__ = {}

    def recurse(obj: pydantic.main.ModelMetaclass, temp_key: str = "") -> None:
        if isinstance(obj, pydantic.main.ModelMetaclass):
            for key, value in obj.schema().items():
                recurse(obj=value, temp_key=temp_key + key if temp_key else key)

        if isinstance(obj, dict):
            for key, value in obj.items():
                recurse(obj=value, temp_key=temp_key + key if temp_key else key)

        if isinstance(obj, list):
            for item in range(len(obj)):
                recurse(
                    obj=obj[item],
                    temp_key=temp_key + str(item) if temp_key else str(item),
                )
        else:
            __fields_dict__[temp_key] = obj

    recurse(schema)
    return __fields_dict__

Not sure how this would be helpful to you, maybe just for inspiration. But I'll keep playing with this.

Out: {'': <class '__main__.SchemaRoot'>,
 'definitions': {'SchemaSubBase': {'properties': {'someSubText': {'title': 'Somesubtext',
                                                                  'type': 'string'}},
                                   'required': ['someSubText'],
                                   'title': 'SchemaSubBase',
                                   'type': 'object'}},
 'definitionsSchemaSubBase': {'properties': {'someSubText': {'title': 'Somesubtext',
                                                             'type': 'string'}},
                              'required': ['someSubText'],
                              'title': 'SchemaSubBase',
                              'type': 'object'},
 'definitionsSchemaSubBaseproperties': {'someSubText': {'title': 'Somesubtext',
                                                        'type': 'string'}},
 'definitionsSchemaSubBasepropertiessomeSubText': {'title': 'Somesubtext',
                                                   'type': 'string'},
 'definitionsSchemaSubBasepropertiessomeSubTexttitle': 'Somesubtext',
 'definitionsSchemaSubBasepropertiessomeSubTexttype': 'string',
 'definitionsSchemaSubBaserequired0': 'someSubText',
 'definitionsSchemaSubBasetitle': 'SchemaSubBase',
 'definitionsSchemaSubBasetype': 'object',
 'properties': {'id': {'title': 'Id', 'type': 'integer'},
                'someRootText': {'title': 'Someroottext', 'type': 'string'},
                'subData': {'default': [],
                            'items': {'$ref': '#/definitions/SchemaSubBase'},
                            'title': 'Subdata',
                            'type': 'array'}},
 'propertiesid': {'title': 'Id', 'type': 'integer'},
 'propertiesidtitle': 'Id',
 'propertiesidtype': 'integer',
 'propertiessomeRootText': {'title': 'Someroottext', 'type': 'string'},
 'propertiessomeRootTexttitle': 'Someroottext',
 'propertiessomeRootTexttype': 'string',
 'propertiessubData': {'default': [],
                       'items': {'$ref': '#/definitions/SchemaSubBase'},
                       'title': 'Subdata',
                       'type': 'array'},
 'propertiessubDataitems': {'$ref': '#/definitions/SchemaSubBase'},
 'propertiessubDataitems$ref': '#/definitions/SchemaSubBase',
 'propertiessubDatatitle': 'Subdata',
 'propertiessubDatatype': 'array',
 'required0': 'someRootText',
 'required1': 'id',
 'title': 'SchemaRoot',
 'type': 'object'}

[ArcLightSlavik]

Tortoise has some stuff with pydantic conversion.
Link: https://tortoise-orm.readthedocs.io/en/latest/examples/pydantic.html
May be helpful.

[j-gimbel]

Thank you both, I will check this out tomorrow.

[bazakoskon]

Hey @j-gimbel ! Have you managed to find a solution? 👀