How to use pydantic and sqlalchemy models with relationship

[mxdev88]

Hello,

I'm trying to pass pydantic models to sqlalchemy models based on sql-databases doc.

It works well for single models but fails to work with relationship. I'm expecting to receive an objet nesting several other objects in my endpoint.

Is there a way to do that using the BaseModel.dict() like in the documentation or do I need to map my pydantic model to my sqlalchemy model in order to achieve this?

Example

from typing import Optional, List

from fastapi import Depends, FastAPI, HTTPException

from pydantic import BaseModel

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, sessionmaker, relationship
from sqlalchemy import ForeignKey, Column, Integer, String

Base = declarative_base()


class Parent(Base):
    __tablename__ = "parents"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(255), index=True)

    children = relationship("Child", back_populates="parent")


class Child(Base):
    __tablename__ = "children"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(255), index=True)
    parent_id = Column('parent_id', Integer(), ForeignKey('parents.id'), nullable=False)

    parent = relationship("Parent", back_populates="children")


class ChildSchema(BaseModel):
    id: Optional[int]
    name: str

    class Config:
        orm_mode = True


class ParentSchema(BaseModel):
    id: Optional[int]
    name: str

    children: List[ChildSchema] = []

    class Config:
        orm_mode = True


SQLALCHEMY_DATABASE_URL = "sqlite:///./test_app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)

api = FastAPI()


def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


@api.post("/parents/", response_model=ParentSchema)
def post_parent(parent: ParentSchema, db: Session = Depends(get_db)):
    db_parent = Parent(**parent.dict())
    db.add(db_parent)
    db.commit()
    db.refresh(db_parent)
    return ParentSchema.from_orm(db_parent)


if __name__ == "__main__":
    from fastapi.testclient import TestClient

    client = TestClient(api)

    # this works
    data1 = """{"name": "string"}"""
    response = client.post("/parents/", data=data1)
    assert response.status_code == 200

    # this does not work
    data2 = """{"name": "string", "children": [{"name": "string"}]}"""
    response = client.post("/parents/", data=data2)
    assert response.status_code == 200

Description

data1 works well as expected.
data2 raise AttributeError: 'dict' object has no attribute '_sa_instance_state'

Environment

• FastAPI Version: 0.58.0
• Python version: Python 3.7.7

[tsh356]

Try using fastapi's jsonable_encoder instead of pydantic's dict() method:

https://fastapi.tiangolo.com/tutorial/encoder/#using-the-jsonable_encoder

from fastapi.encoders import jsonable_encoder

@api.post("/parents/", response_model=ParentSchema)
def post_parent(parent: ParentSchema, db: Session = Depends(get_db)):
    parent_data = jsonable_encoder(parent)
    db_parent = Parent(**parent_data)
    db.add(db_parent)
    db.commit()
    db.refresh(db_parent)
    return ParentSchema.from_orm(db_parent)

[imjohsep]

@mxdev88

@api.post("/parents", response_model=ParentSchema)
def post_parent(parent: ParentSchema, db: Session = Depends(get_db)):
    db_parent = Parent(id=parent.id, name=parent.name)
    db.add(db_parent)
    db.commit()

    db_children = [Child(name=c.name, parent_id=db_parent.id) for c in parent.children]
    db.add_all(db_children)
    db.commit()

    return db_parent

I agree that we should be able to construct the relationships from the nested orm schemas within the parent schema, but I've only really been able to get this to work by explicitly defining and inserting the children. If there is a better way to do this, I would be interested.

[kielerrr]

I agree that we should be able to construct the relationships from the nested orm schemas within the parent schema, but I've only really been able to get this to work explicitly defining and inserting the children. If there is a better way to do this, I would be interested.

I agree as well. It seems counter to the entire spirit of modeling in fastapi. Maybe we're missing something.

[avico78]

def post_parent(parent: ParentSchema, db: Session = Depends(get_db)):
parent_data = jsonable_encoder(parent)
db_parent = Parent(**parent_data)
db.add(db_parent)
db.commit()
db.refresh(db_parent)
return ParentSchema.from_orm(db_parent)

@imjohsep - solution is working ,thanks .
did u find a better way to do it ?
seem for more complex structure/relation it would required lot of code to handle it,
lets say one parent of child of Childs .
im curious to understand if this a bug or limitation
as when we do

def post_parent(parent: ParentSchema, db: Session = Depends(get_db)):
    parent_data = jsonable_encoder(parent)

parent_data hold all nested trees correctly ....
but when trying to intiliaze by the Parent model:

db_parent = Parent(**parent_data)

it somehow breaking it ...

@tsh356 - are u sure ur solution solving this ?

fatapi V0.63
I tried this solution and im getting same error for input:

{
  "name": "AA",
  "children": [
    {

      "name": "A"
    }
  ]
}

error:

nd_event
    child_state, child_dict = instance_state(child), instance_dict(child)
AttributeError: 'dict' object has no attribute '_sa_instance_state'

models:

class Parent(Base):
    __tablename__ = "parents"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(255), index=True)

    children = relationship("Child", back_populates="parent")


class Child(Base):
    __tablename__ = "children"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(255), index=True)
    parent_id = Column('parent_id', Integer(), ForeignKey('parents.id'), nullable=False)

    parent = relationship("Parent", back_populates="children")

Schemas:

class ChildSchema(BaseModel):
    id: Optional[int]
    name: str

    class Config:
        orm_mode = True


class ParentSchema(BaseModel):
    id: Optional[int]
    name: str

    children: List[ChildSchema] = None

    class Config:
        orm_mode = True

fastapi:

@customer_router.post("/parents/", response_model=ParentSchema)
def post_parent(parent: ParentSchema, db: Session = Depends(get_db)):
    parent_data = jsonable_encoder(parent)
    db_parent = Parent(**parent_data)
    db.add(db_parent)
    db.commit()
    db.refresh(db_parent)
    return ParentSchema.from_orm(db_parent)

[ccrvlh]

The solution proposed to use jsonable_encoder also doesn't work for me.

The only way I can handle this at the moment is to split the models and then pass the sqlalchemy child instance manually to the parent model class.

At the moment I have a model with about ~50 fields, 10 of them are relationships (both one-to-one and one-to-many, the later would require using sqlalchemy's append method), so I've been cracking my head open to find a way to make this work without writing hundreds of lines of code to manipulate the schemas/models that would most definitely lead to a complex and error prone code.

And I have to say I wonder if this is even possible... I could try to help on this, but I confess at the moment I have no idea how.
Maybe this would have to be a pydantic's feature, or even sqlalchemy's?

[minyoung90]

The solution proposed to use jsonable_encoder also doesn't work for me too.

https://fastapi.tiangolo.com/tutorial/sql-databases/

The example above has no clue for creating an object with relationship.
The tables have a relationship but, not to use when it is created so, It could be done.

[hecjhs]

@mxdev88 did you have any luck with this?

[ipxwork]

I have a generic base class for CRUD, and creating parent with kids as an array I did next:

I create the parent from fields that haven't relations and then add relations fields to the object created and commit it.
I get the model class of the kids from relationship fields

...

def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
        obj_in_dict = jsonable_encoder(obj_in, exclude_defaults=True)
        obj_in_parent = self.object_as_dict(obj_in_dict)
        obj_in_data = self.object_as_dict(obj_in_dict, relationships=True)
        db_obj = self.model(**obj_in_parent)  # type: ignore

        db.add(db_obj)
        db.commit()

        for key, value in obj_in_data.items():
            if key in obj_in_parent or not hasattr(self.model, key):
                continue

            children = [
                getattr(self.model, key).prop.entity._identity_class(**child, item_id=db_obj.id) for child in value
            ]

            setattr(db_obj, key, children)

        db.commit()
        db.refresh(db_obj)
        return db_obj


    def object_as_dict(self, obj, relationships=False):
        """
        Converts an SQLAlchemy instance to a dictionary.

        :param relationships: If true, also include relationships in the output dict
        """
        obj_dict = jsonable_encoder(obj, exclude_defaults=True)
        properties = inspect(self.model).mapper.all_orm_descriptors

        if not relationships:
            return {
                key: obj_dict[key]
                for key, value in properties.items()
                if (
                    not hasattr(value, "prop")
                    or not isinstance(value.prop, RelationshipProperty)
                )
                and key in obj_dict
            }
        else:
            return {key: obj_dict[key] for key, value in properties.items() if key in obj_dict}

...

[darshanputtaswamy]

Did anyone find a better approach to saving and loading objects?