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

see more Info,
following sqlalchemy- creating one -> one relation:
https://docs.sqlalchemy.org/en/14/orm/basic_relationships.html

Basically, simple relation just adding "uselist=False"
for simple input:

{
  "customer_no": 1,
  "address": {
    "streetAddress": "some address"
  }
}

I have customer table
and address

I expect each customer insreted with a single address

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':
                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)



class CustomerModel(Base):
    __tablename__ = "customer"
    id = Column(Integer, primary_key=True, index=True)
    customer_no = Column(Integer,nullable=False, index=True)
    address = relationship("AddressModel",uselist=False, 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")

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

class SchemaAddress(SchemaAddressBase):
    id: int
    address_customer: int
    class Config:
        orm_mode = True

class SchemaCustomerBase(BaseModel):
    customer_no: int
    address: SchemaAddressBase=None
    class Config:
        orm_mode = True

class SchemaCustomer(SchemaCustomerBase):
    id: int
    class Config:
        orm_mode = True
```
Endpoint :  

```
def db_add_nested_data_pydantic(root: SchemaCustomerBase,db: Session):

    db_root = CustomerModel(**root.dict())
    db.add(db_root)
    db.commit()
    db.refresh(db_root)
    return db_root

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


error:

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 "./nested/schema.py", line 51, in _declarative_constructor_auto_instantiate_nested
new_elem = childclass(**elem)
TypeE

[scd75]

Hi @avico78 , it is expected to not work with one-to-one relationships, as I only added the case of one-to-many relationships in the following code for init:

        if k in relationships.keys():
            if relationships[k].direction.name == 'ONETOMANY':

I guess you can easily add here a handling of "ONETOONE" relationships (and by the way I cleaned a little bit the code):

        if k in relationships.keys():
            if relationships[k].direction.name in ['ONETOMANY', 'ONETOONE']
                childclass = getattr(sys.modules[self.__module__], relationships[k].argument)
                setattr(self, k, [childclass(**elem) for elem in kwargs[k]])

let me know if that works.. But the question is: are you sure that want to "Create" a new object for the corresponding one-to-one relationship? (this was my use case for one-to-many, but just make sure that you also want to create the target to the one-to-one. By 'create' I mean, not link to an existing object)

[avico78]

@scd75 ,

thanks much for your reply ,
I'd explain the use case ,
I have set of tables which can be map as a Json document ,
the relation between them is based on how I want the Json will look like.
before i explain the mapping rule ,
see example:

{
  "main_table_filed1": "value",
  "sub_table1": [
    {
      "sub_table1_filed": "value1"
    },
    {
      "sub_table1_filed": "value2"
    }
  ]
}

first level objects(flat ones) - has a single occurrence
will come from main table/leading table ,
Other objects , which has one2may (see sub_table1) - will come from
different table (sub_table1)
the dependencies will always be down --> top,
meaning sub table will have a Fk for the upper object ,

So the schema should be :

class MainTable(Base):
    __tablename__ = "main_table"
    id = Column(Integer, primary_key=True, index=True)
    sub_table1= relationship("Sub_table1",uselist=True, back_populates="main_table")


class Sub_table1(Base):
    __tablename__ = "sub_table1"
    id = Column(Integer, primary_key=True,index=True )
    sub_table1_filed = Column(String)
    main2sub_table1= Column(Integer, ForeignKey("main.id"))
    main= relationship("MainTable", back_populates="sub_table1")

Pydanmtic :

class SchemaSubTable1Base(BaseModel):
    sub_table1_filed: Optional[str]
    class Config:
        orm_mode = True

class SchemaSubTable1(SchemaSubTable1Base):
    id: int
    sub_table1_filed: int
    class Config:
        orm_mode = True

class SchemaMainBase(BaseModel):
    main_table_filed1: str
    sub_table1: List[SchemaSubTable1Base]=None
    class Config:
        orm_mode = True

class SchemaMain(SchemaMainBase):
    id: int
    class Config:
        orm_mode = True

Now , sub_table1 in SchemaMainBase define as List as it can multiple occurrences,
But let add another dictionary object :

{
  "main_table_filed1": "value",
  "sub_table1": [
    {
      "sub_table1_filed": "value1"
    },
    {
      "sub_table1_filed": "value2"
    }
  ],
  "dict_obj": {
    "main_table_filed2": "value"
  }
}

So to create such object we can either map it from main table (he has single occurrence )
or from another sub table (one2one relation).
I'm not sure how it can be implement ,
by the pydantic class it only a change of

class SchemaMainBase(BaseModel):
    main_table_filed1: str
    sub_table1: List[SchemaSubTable1Base]=None
    dict_obj:    SchemaSubTable1Base ={}
#or  if coming from same MainTable
    dict_obj:    dict 
    class Config:
        orm_mode = True

those mapping rules can be define by the user so it open for manipluation
in order to get the correct json ,
So assumption the Fk will be always from main--> sub.

Sadly , I lost by mistake the code I start over writing based your new Base function .
if that ok with u , I'd be really appreciate if i can directly communicate you .

Thanks again !

my email avico78@gmail.com

[avico78]

@scd75 ,

after loosing my whole code folder I just finish coding the whole thing again .
Im getting an error now from the function ,
not sure what im doing wrong:

File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/state.py", line 430, in _initialize_instance
return manager.original_init(*mixed[1:], **kwargs)
File "./databases_config/db.py", line 79, in _declarative_constructor_auto_instantiate_nested
childclass = getattr(sys.modules[self.__module__], relationships[k].argument)
TypeError: getattr(): attribute name must be string

I guess it may related to they i setup the scripts /Base location.
I did the following:

base_class.py
(using ur last version)

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 k in relationships.keys():
            if relationships[k].direction.name in ['ONETOMANY', 'ONETOONE']:
                childclass = getattr(sys.modules[self.__module__], relationships[k].argument)
                setattr(self, k, [childclass(**elem) for elem in kwargs[k]])


_declarative_constructor_auto_instantiate_nested.__name__ = "__init__"

Base = declarative_base(
    constructor=_declarative_constructor_auto_instantiate_nested)

base.py

from app_sa.base_class import Base
from app_sa.models import *

Now im taking the Base from base.py
main.py

from app_sa.base import Base
Base.metadata.create_all(bind=postgress_customer_engine)

Running the endpoint with proper input,
routers.py:

from typing import List, Optional
import sqlalchemy
from fastapi import APIRouter, Depends, HTTPException,Body
from databases_config.db import get_db
from sqlalchemy.orm import Session
from app_sa.schema import SchemaCustomerBase
from app_sa.models import CustomerModel


app_sa_router = APIRouter()

def db_add_nested_data_pydantic_generic(db: Session, root: SchemaCustomerBase):
    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", response_model=SchemaCustomerBase)
def add_nested_data_pydantic_generic(root: SchemaCustomerBase, db: Session = Depends(get_db)):
    data = db_add_nested_data_pydantic_generic(db=db, root=root)
    return data

Tables being drop/create successfully from main :
main.py

from typing import Optional
from fastapi import FastAPI, Depends, HTTPException,Request
from databases_config.db import metadata,postgress_customer_engine
from app_sa.routes import app_sa_router
from app_sa.base import Base

app = FastAPI(title=settings.PROJECT_NAME,version=settings.PROJECT_VERSION)

Base.metadata.drop_all(bind=postgress_customer_engine)
Base.metadata.create_all(bind=postgress_customer_engine)

You mention :

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)

I did tried placing the function and class also in module - same result .

[scd75]

@avico78 , understand it is an issue with the finding of the related model name.
try running in debug mode, and inspect the following object: relationships[k].argument (it should be a string)

what does it look like?

PS: maybe would be good to carry on this discussion on another thread to not pollute the initial one

[avico78]

@scd75 ,thanks
Following ur suggestion ,i added some prints inside the function ,
seems the relationships[k].argument actually doesn't return a str
but

<sqlalchemy.ext.declarative.clsregistry._class_resolver object at 0x7f2fc598ce50>>

Test:

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 k in relationships.keys():
             if relationships[k].direction.name in ['ONETOMANY', 'ONETOONE']:
                  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]])


_declarative_constructor_auto_instantiate_nested.__name__ = "__init__"

Base = declarative_base(
    constructor=_declarative_constructor_auto_instantiate_nested)


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

class SubscriberModel(Base):
    __tablename__ = "subscriber"
    id = Column(Integer, primary_key=True, index=True)
    subscriber_no = Column(Integer)
    sub_customer = Column(Integer, ForeignKey("customer.id"))
    customer = relationship("CustomerModel", back_populates="subscriber")

class SchemaSubscriberBase(BaseModel):
    subscriber_no: int
    class Config:
        orm_mode = True

class SchemaSubscriber(SchemaSubscriberBase):
    id: int
    sub_customer:int
    class Config:
         orm_mode = True

class SchemaCustomerBase(BaseModel):
    customer_no: int
    subscriber: List[SchemaSubscriberBase] = None
    class Config:
        orm_mode = True

class SchemaCustomer(SchemaCustomerBase):
    id: int
    class Config:
        orm_mode = True
 

debug prints:

Current Key: subscriber relationships_direction ONETOMANY argument <bound method _class_resolver._resolve_name of <sqlalchemy.ext.declarative.clsregistry._class_resolver object at 0x7f2fc598ce50>>

Not sure what is it mean? it didnt return the related object(subscriber as string) but as object?
How can i fix it?
Im sure this something related to configuration as previously it was working

[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]