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? 👀

[Wouterkoorn]

Hey @bazakoskon, maybe you have managed to find a solution? 👀

[scd75]

Hello,
bumped into the same issue. One solution that I found interesting is simply to override the "init" constructor in SQLAlchemy classes, so that all nested elements are directly instantiated from nested objects passed into the arguments:

I reused the "_declarative_constructor" function code (which is the by-default init method for SQLAlchemy Base class)

class user(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    
    my_nested_list = relationship("MyNestedObjectClass", back_populates="owner", cascade='all, delete-orphan')

    def __init__(self, **kwargs):
        cls_ = type(self)
        for k in kwargs:
            if not hasattr(cls_, k):
                raise TypeError(
                    "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
                )
            if k == 'my_nested_list':
                for e in kwargs[k]:
                    new_nested_element = MyNestedObjectClass(**e)
                    self.my_nested_list.append(new_nested_element)
            else:
                setattr(self, k, kwargs[k])

I guess the overrided "_declarative_constructor" function can be generalized (and used as "contructor" parameter when calling declarative_base()) to handle automatically iterables found in the list of kwargs, based on the class relationships.

EDIT:
you can create the following function, and use it instead of the default constructor in declarative_base(). Works well on my side:

def _declarative_constructor_auto_instantiate_nested(self, **kwargs):
    """A simple constructor that allows initialization from kwargs.

    Sets attributes on the constructed instance using the names and
    values in ``kwargs``.

    Only keys that are present as
    attributes of the instance's class are allowed. These could be,
    for example, any mapped columns or relationships.

    EDITED by Samuel to auto_instantiate nested lists's elements as child classes
    """
    cls_ = type(self)
    relationships = self.__mapper__.relationships
    for k in kwargs:
        if not hasattr(cls_, k):
            raise TypeError(
                "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
            )
        if k in relationships.keys():
            if relationships[k].direction.name == 'ONETOMANY':
                childclass = getattr(sys.modules['models'], relationships[k].argument)
                nestedattribute = getattr(self, k)
                for elem in kwargs[k]:
                    new_elem = childclass(**elem)
                    nestedattribute.append(new_elem)
        else:
            setattr(self, k, kwargs[k])

_declarative_constructor_auto_instantiate_nested.__name__ = "__init__"

Base = declarative_base(
    constructor=_declarative_constructor_auto_instantiate_nested)

[avico78]

Have you tried @tiangolo's Pydantic SQLAlchemy ?

I did check this for nested object and it doesnt work

[avico78]

Hello,
bumped into the same issue. One solution that I found interesting is simply to override the "init" constructor in SQLAlchemy classes, so that all nested elements are directly instantiated from nested objects passed into the arguments:

I reused the "_declarative_constructor" function code (which is the by-default init method for SQLAlchemy Base class)

class user(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    
    my_nested_list = relationship("MyNestedObjectClass", back_populates="owner", cascade='all, delete-orphan')

    def __init__(self, **kwargs):
        cls_ = type(self)
        for k in kwargs:
            if not hasattr(cls_, k):
                raise TypeError(
                    "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
                )
            if k == 'my_nested_list':
                for e in kwargs[k]:
                    new_nested_element = MyNestedObjectClass(**e)
                    self.my_nested_list.append(new_nested_element)
            else:
                setattr(self, k, kwargs[k])

I guess the overrided "_declarative_constructor" function can be generalized (and used as "contructor" parameter when calling declarative_base()) to handle automatically iterables found in the list of kwargs, based on the class relationships.

EDIT:
you can create the following function, and use it instead of the default constructor in declarative_base(). Works well on my side:

def _declarative_constructor_auto_instantiate_nested(self, **kwargs):
    """A simple constructor that allows initialization from kwargs.

    Sets attributes on the constructed instance using the names and
    values in ``kwargs``.

    Only keys that are present as
    attributes of the instance's class are allowed. These could be,
    for example, any mapped columns or relationships.

    EDITED by Samuel to auto_instantiate nested lists's elements as child classes
    """
    cls_ = type(self)
    relationships = self.__mapper__.relationships
    for k in kwargs:
        if not hasattr(cls_, k):
            raise TypeError(
                "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
            )
        if k in relationships.keys():
            if relationships[k].direction.name == 'ONETOMANY':
                childclass = getattr(sys.modules['models'], relationships[k].argument)
                nestedattribute = getattr(self, k)
                for elem in kwargs[k]:
                    new_elem = childclass(**elem)
                    nestedattribute.append(new_elem)
        else:
            setattr(self, k, kwargs[k])

_declarative_constructor_auto_instantiate_nested.__name__ = "__init__"

Base = declarative_base(
    constructor=_declarative_constructor_auto_instantiate_nested)

Would be thankful if u can add some full example,i tried to implement it and couldn't make it work ,thanks

[avico78]

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.

Did u manage to find for this a solution?

[scd75]

@avico78
please see attached full example. Just create the "Base" class as follows, and have all you SQLAlchemy models inheritate from this "Base" class. It should work..

import sys
from sqlalchemy.ext.declarative import declarative_base


def _declarative_constructor_auto_instantiate_nested(self, **kwargs):
    """A simple constructor that allows initialization from kwargs.

    Sets attributes on the constructed instance using the names and
    values in ``kwargs``.

    Only keys that are present as
    attributes of the instance's class are allowed. These could be,
    for example, any mapped columns or relationships.

    EDITED by @scd75 to auto_instantiate nested lists' elements as child classes
    """
    cls_ = type(self)
    relationships = self.__mapper__.relationships
    for k in kwargs:
        if not hasattr(cls_, k):
            raise TypeError(
                "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
            )
        if k in relationships.keys():
            if relationships[k].direction.name == 'ONETOMANY':
                childclass = getattr(sys.modules[self.__module__], relationships[k].argument)
                nestedattribute = getattr(self, k)
                for elem in kwargs[k]:
                    new_elem = childclass(**elem)
                    nestedattribute.append(new_elem)
        else:
            setattr(self, k, kwargs[k])


_declarative_constructor_auto_instantiate_nested.__name__ = "__init__"

Base = declarative_base(
    constructor=_declarative_constructor_auto_instantiate_nested)

Please note that the proposed implementation requires that nested models are in the same module as parent models, as you can see here:

childclass = getattr(sys.modules[self.__module__], relationships[k].argument)

But this can be easily modified if needed

[avico78]

@scd75 - great work man!

I've tested with ~3-4 level of json level depth and it work perfectly,
for both/read and write,

[avico78]

@scd75 - found some issue ,
In case we have relation of 1 -> 1 ,
i.e:

class CustomerModel(Base):
    __tablename__ = "customer"
    id = Column(Integer, primary_key=True, index=True)
    customer_no = Column(Integer,nullable=False)
    address = relationship("AddressModel", back_populates="customer")


class AddressModel(Base):
    __tablename__ = "address"
    id = Column(Integer, primary_key=True, index=True)
    streetAddress = Column(String)
    address_customer = Column(Integer, ForeignKey("customer.id"))
    customer = relationship("CustomerModel", back_populates="address")

And by the pyadntic class we define the relation as one-->one

class SchemaAddressBase(BaseModel):
    streetAddress: str
    class Config:
        orm_mode = True

class SchemaCustomerBase(BaseModel):
    customer_no: int
    address:  SchemaAddressBase

meaning the json would look like:

{
  "customer_no": 6,
  "address": {
    "streetAddress": "7777"
}
}

It failed with :

File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/state.py", line 434, in _initialize_instance

return manager.original_init(*mixed[1:], **kwargs)

File "./DynPydantic/schema.py", line 47, in _declarative_constructor_auto_instantiate_nested

new_elem = childclass(**elem)

TypeError: DeclarativeMeta object argument after ** must be a mapping, not str

Maybe im wrong , but seems it doesn't cover the case of nested dictionaries?

[avico78]

@scd75 ,

"if we actually made this a feature, then it would be a magnet for new user requests and issues that would be better solved if they just made their own constructor that does what they want.
"
I really don't get that...i think it could be really nice feature,
actually , i think this feature is a must/fundamental for any ORM package.

I'm still struggling making it work,
it failed here:

childclass = getattr(sys.modules[self.__module__], relationships[k].argument)

following your suggestion I added debug print:

from sqlalchemy.ext.declarative import declarative_base
import sys

def _declarative_constructor_auto_instantiate_nested(self, **kwargs):
    cls_ = type(self)
    relationships = self.__mapper__.relationships
    
    for k in kwargs:
        if not hasattr(cls_, k):
            raise TypeError(
                "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
            )
        if k in relationships.keys():
            if relationships[k].direction.name == 'ONETOMANY':
                print("Current Key:",k,"relationships_direction:",relationships[k].direction.name,"argument:",relationships[k].argument)
                #childclass = getattr(sys.modules[self.__module__], relationships[k].argument)
                #setattr(self, k, [childclass(**elem) for elem in kwargs[k]])
        else:
            setattr(self, k, kwargs[k])

_declarative_constructor_auto_instantiate_nested.__name__ = "__init__"

Base = declarative_base(
    constructor=_declarative_constructor_auto_instantiate_nested)

router (i commented some part to see the print of relationships[k].argument):

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

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

@app_sa_router.post("/addNestedModel_pydantic_generic")
def add_nested_data_pydantic_generic(root: SchemaCustomerBase, db: Session = Depends(get_db)):
    print("im root",root)
    data = db_add_nested_data_pydantic_generic(db=db, root=root)
    return {"ok":1}

And I get

#coming from router
im root customer_no=0 first_name='string' last_name='string' subscriber=[SchemaSubscriberBase(subscriber_no=0)]

#coming from function
Current Key: subscriber relationships_direction: ONETOMANY argument: <bound method _class_resolver._resolve_name of <sqlalchemy.ext.declarative.clsregistry._class_resolver object at 0x7f060942dbe0>>

Any idea?

[Wouterkoorn]

@avico78 @scd75 Last few days I have created a library for this, mostly because I wanted to learn how to do that. It seems I went with a different approach. Please let me know what you think, feedback is greatly appriciated.

Github: https://github.com/Wouterkoorn/sqlalchemy-pydantic-orm
Github pages: https://wouterkoorn.github.io/sqlalchemy-pydantic-orm/
PyPI: https://pypi.org/project/sqlalchemy-pydantic-orm/

[avico78]

@Wouterkoorn - tnx for your update i surely check it,
I found also orm ormar Which based on sqlalchemy and actually provide very efficient way and simple handle nested objects.
see:
https://github.com/collerek/ormar

[Sinclert]

Hi 👋🏻

I ended up arriving to this issue when looking for ways to enable nested model auto-instantiation in my SQLAlchemy projects. My comment here is not strictly related to FastAPI, but could improve the solution provided in @scd75 's comment.

My proposal includes two major changes:

• Replacing the cumbersome sys.modules["models"] part by relationships[key].mapper.entity.
• Defining the function as a usual __init__ method.

In order to define our custom constructor as a common __init__ method, in the latest version of SQLAlchemy (v1.4.X), the constructor argument of the declarative_base() function needs to be None. According to its code explanation, this is what would disable SQLAlchemy's default constructor, allowing the custom one to kick in (normal Python behaviour).

Proposed constructor:

from sqlalchemy.orm import ONETOMANY
from sqlalchemy.orm import declarative_base


Base = declarative_base(constructor=None)


class BaseModel:
    """Base class for all the Python data models"""

    def __init__(self, **kwargs):
        """
        Custom initializer that allows nested children initialization.
        Only keys that are present as instance's class attributes are allowed.
        These could be, for example, any mapped columns or relationships.

        Code inspired from GitHub.
        Ref: https://github.com/tiangolo/fastapi/issues/2194
        """

        cls = self.__class__
        model_columns = self.__mapper__.columns
        relationships = self.__mapper__.relationships

        for key, val in kwargs.items():

            if not hasattr(cls, key):
                raise TypeError(f"Invalid keyword argument: {key}")

            if key in model_columns:
                setattr(self, key, val)
                continue

            if key in relationships:
                relation_dir = relationships[key].direction.name
                relation_cls = relationships[key].mapper.entity

                if relation_dir == ONETOMANY.name:
                    instances = [relation_cls(**elem) for elem in val]
                    setattr(self, key, instances)


class ChildModel(Base, BaseModel):
    ...


class ParentModel(Base, BaseModel):
    ...

[scd75]

Hi 👋🏻

I ended up arriving to this issue when looking for ways to enable nested model auto-instantiation in my SQLAlchemy projects. My comment here is not strictly related to FastAPI, but could improve the solution provided in @scd75 's comment.

My proposal includes two major changes:

• Replacing the cumbersome sys.modules["models"] part by relationships[key].mapper.entity.
• Defining the function as a usual __init__ method.

In order to define our custom constructor as a common __init__ method, in the latest version of SQLAlchemy (v1.4.X), the constructor argument of the declarative_base() function needs to be None. According to its code explanation, this is what would disable SQLAlchemy's default constructor, allowing the custom one to kick in (normal Python behaviour).

Proposed constructor:

from sqlalchemy.orm import ONETOMANY
from sqlalchemy.orm import declarative_base


Base = declarative_base(constructor=None)


class BaseModel:
    """Base class for all the Python data models"""

    def __init__(self, **kwargs):
        """
        Custom initializer that allows nested children initialization.
        Only keys that are present as instance's class attributes are allowed.
        These could be, for example, any mapped columns or relationships.

        Code inspired from GitHub.
        Ref: https://github.com/tiangolo/fastapi/issues/2194
        """

        cls = self.__class__
        model_columns = self.__mapper__.columns
        relationships = self.__mapper__.relationships

        for key, val in kwargs.items():

            if not hasattr(cls, key):
                raise TypeError(f"Invalid keyword argument: {key}")

            if key in model_columns:
                setattr(self, key, val)
                continue

            if key in relationships:
                relation_dir = relationships[key].direction.name
                relation_cls = relationships[key].mapper.entity

                if relation_dir == ONETOMANY.name:
                    instances = [relation_cls(**elem) for elem in val]
                    setattr(self, key, instances)


class ChildModel(Base, BaseModel):
    ...


class ParentModel(Base, BaseModel):
    ...

Hi @Sinclert , thanks for your comment.
Just , out of curiosity, what would be the issues with just modifying the name of the custom constructor fucntion to "init" ?

as in here:

_declarative_constructor_auto_instantiate_nested.__name__ = "__init__"

[Sinclert]

Hey @scd75 ,

I believe the is no technical difference.

However, defining the constructor as you would normally do with any other __init__ method, improves readability. Furthermore, there are some static code analyzers that do not recognize a function whose name gets set to "__init__" (marks any instantiation with a warning ⚠️ ), but correctly identify any normal __init__ method (i.e. PyCharm code analysis).

[yinzixie]

Hi 👋🏻

I ended up arriving to this issue when looking for ways to enable nested model auto-instantiation in my SQLAlchemy projects. My comment here is not strictly related to FastAPI, but could improve the solution provided in @scd75 's comment.

My proposal includes two major changes:

• Replacing the cumbersome sys.modules["models"] part by relationships[key].mapper.entity.
• Defining the function as a usual __init__ method.

In order to define our custom constructor as a common __init__ method, in the latest version of SQLAlchemy (v1.4.X), the constructor argument of the declarative_base() function needs to be None. According to its code explanation, this is what would disable SQLAlchemy's default constructor, allowing the custom one to kick in (normal Python behaviour).

Proposed constructor:

from sqlalchemy.orm import ONETOMANY
from sqlalchemy.orm import declarative_base


Base = declarative_base(constructor=None)


class BaseModel:
    """Base class for all the Python data models"""

    def __init__(self, **kwargs):
        """
        Custom initializer that allows nested children initialization.
        Only keys that are present as instance's class attributes are allowed.
        These could be, for example, any mapped columns or relationships.

        Code inspired from GitHub.
        Ref: https://github.com/tiangolo/fastapi/issues/2194
        """

        cls = self.__class__
        model_columns = self.__mapper__.columns
        relationships = self.__mapper__.relationships

        for key, val in kwargs.items():

            if not hasattr(cls, key):
                raise TypeError(f"Invalid keyword argument: {key}")

            if key in model_columns:
                setattr(self, key, val)
                continue

            if key in relationships:
                relation_dir = relationships[key].direction.name
                relation_cls = relationships[key].mapper.entity

                if relation_dir == ONETOMANY.name:
                    instances = [relation_cls(**elem) for elem in val]
                    setattr(self, key, instances)


class ChildModel(Base, BaseModel):
    ...


class ParentModel(Base, BaseModel):
    ...

Not sure If I am wrong.
But in my case, there is no ONETOONE in sqlachemy.

# no ONETOONE error
from sqlalchemy.orm import ONETOMANY, ONETOONE

And I always get a ONETOMANY name from relationships[key].direction.name, even if the modles are exactly flow the ONETOONE standard.

Also, the official document said:

Within the ORM, “one-to-one” is considered as a convention where the ORM expects that only one related row will exist for any parent row.

So I guess ONETOONE is also regarded as ONETOMANY in Sqlachemy ?

And I slightly change the code usingif isinstance(val, list) to replace if relation_dir == ONETOMANY.name. It works well in my case.

if isinstance(val, list):
    instances = [relation_cls(**elem) for elem in val]
    setattr(self, key, instances)
elif isinstance(val, dict):
    instance = relation_cls(**val)
    setattr(self, key, instance)

[Sinclert]

@yinzixie I think you are right. There is no ONETOONE constant on the ORM package.

In my initial testing, all the relationships[key].direction.name properties hold a "ONETOMANY" value, although I am sure that if that property exists, is because it could hold a different value if the proper levers are pulled.

For instance, I noticed that the relationship constructor receives an argument called single_parent (False by default). Not sure if setting that value True is enough to demonstrate my hypothesis, though, but it is a good start.

[don4of4]

@tiangolo Can we sponsor a fix to this issue or have you opine here? This issue is perhaps the messiest part of this whole FastApi framework.

We have SqlAlchemy -> Pydantic models working very nicely, but going the other direction is involved -- especially when you throw in relationships and composite primary keys.

We have a modification of https://github.com/Wouterkoorn/sqlalchemy-pydantic-orm in the works to see if that will solve our issue (adding non "id" pk support, and adding composite pks).

[jrlopes2005]

Agree with @don4of4. I tried @Wouterkoorn solution and it works great with nested models. @tiangolo please take a look on this if you can.

[Wouterkoorn]

We have a modification of https://github.com/Wouterkoorn/sqlalchemy-pydantic-orm in the works to see if that will solve our issue (adding non "id" pk support, and adding composite pks).

You're also welcome to make a PR. I'll be happy to spend some more time on it as well, it was indeed just a first POC version (although we do use it in production)

[orionmontoyac]

I used __ init__ method in the BaseModel class to convert the nested Dictonary into an ORM object.

class Star(db.Model, BaseModelMixin):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String, nullable=False)
    constellation_id = db.Column(db.Integer, db.ForeignKey('constellation.id'), nullable=False)

    def __init__(self, name: str):
        self.name = name


class Constellation(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String, nullable=False)
    abbr = db.Column(db.String, nullable=False)
    right_ascension = db.Column(db.String, nullable=False)
    stars = db.relationship('Star', backref='constellation', lazy=False, cascade='all, delete-orphan')

    def __init__(self, name: str, abbr: str, right_ascension: str, stars=[]):
        self.name = name
        self.abbr = abbr
        self.right_ascension = right_ascension
        self.stars = [Star(**star) for star in stars]

[Wouterkoorn]

From the moment I created my PoC, my idea was to integrate it into https://github.com/tiangolo/pydantic-sqlalchemy.
But at that point in time, SQLModel was not a thing yet, which might now somewhat deprecate the needs of the pydantic-sqlalchemy project.

I have no experience with SQLModel, is the topic of this discussion still an issue using that library? I could see this being solved by not needing to switch between models anymore

[MarijnStam]

Using SQLModel still persists this issue. Consider the following models:

class IngredientBase(SQLModel):
    name: str = Field(index=True, unique=True)
    
class Ingredient(IngredientBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    meals: List["Meal"] = Relationship(back_populates="ingredients", link_model=MealIngredient)
    
class IngredientCreate(IngredientBase):
    pass

class IngredientRead(IngredientBase):
    id: int

class MealBase(SQLModel):
    name: str = Field(index=True, unique=True)

class Meal(MealBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    ingredients: List[Ingredient] = Relationship(back_populates="meals", link_model=MealIngredient)
    
class MealCreate(MealBase):
    ingredients: List[IngredientCreate] = []

I have a many-to-many relationship between Meal and Ingredients with a simple link table. I want to be able to receive a Meal with a set of ingredients, (see definition of MealCreate). However upon receiving the payload, I still have to manually populate the nested classed:

@app.put("/new_meal/", response_model=MealReadAll)
async def new_meal(*, session: Session = Depends(get_session), new_meal: MealCreate):
    db_meal = Meal.from_orm(new_meal)   #-> db_meal does not contain any ingredients after this
    ingredients = [Ingredient(name=x.name) for x in new_meal.ingredients]  #-> Manually instantiate ingredients as found in the payload
    setattr(db_meal, 'ingredients', new_ingredients)  #-> Link them to the Meal 

After doing this, adding the Meal to the db using session.add() works perfectly fine and also inserts the join table, however it would be nice if this would be automatically parsed.

Python is still fairly new for me and SQLModel/FastAPI is really new for me, so please point out if what I am describing is not related to the issue at hand and I am doing something else wrong.

[Skorpyon]

Thats why I love Django framework...

[DennyWeinberg]

How does Django solve this?

[Guddi11]

Afaik Django does not combine API Schema/Serializer and ORM but you need to declare both models (=ORM) and serializer separate. The greater flexibility comes at the cost of a little more code here, but I guess for more complex data structures it might be worth it.

[hasansezertasan]

Hi 👋 everyone. Is there an update on this?

[don4of4]

Not to dogpile, but this is one of the weaker areas of fastapi. Our team recently switched to Django since the more complete, opinionated framework was easier than fussing with issues like this.

[DennyWeinberg]

Hi @don4of4 , really appreciate your conment. We have big discussions and I'm doing a lot of research because of that topic. Could you please elaborate on your findings? I'm especially interested in loading, manipulating and storing nested objects wirh foreign keys and manys.

[hasansezertasan]

@don4of4 I don't think it's a weakness of FastAPI. Dealing with ORM Models is not FastAPIs job.