Flexibly Create Nested Database Entries from Incoming Pydantic/SQLModels

[hay-kot]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

# Sudo Code Based on Examples in Docs

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    headquarters: str


    heroes: List["Hero"] = Relationship(back_populates="team")



class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None

    team_id: Optional[int] = Field(default=None, foreign_key="team.id")

    team: Optional[Team] = Relationship(back_populates="heroes")


payload = {
    "name": "Team Name",
    "headquarters": "Whereever".
    "heroes": [
        "name": "Name 1"
        // Other Requied Fields... 👇
    ]
}

with Session(engine) as session:
    Team.create_all_nested(session, payload) # or something?

Description

I would like to do what is described in FastAPI issue #2194

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

In the example above, I'd like to pass in the payload to a method and the following to occur.

• Create new Team entry
• Create Hero entry and/or Relate the existing Hero to the Team

Similarly, I'd like the same to happen on update. Effectively making writing to the SQL database akin to writing to MongoDB

I don't believe this is supported or haven't gotten it to work, but my main questions are.

1. Is this supported?
2. If no, is this a use-case you've thought of?
3. Are you interested in a PR to support this either as a utility method or some sort of decorator?

Loving working with this so far, thanks for all your hard work!

Operating System

macOS

Operating System Details

No response

SQLModel Version

0.0.3

Python Version

3.9.6

Additional Context

I have accomplished this with SQLAlchemy in the past by using an auto_init decarator.

from functools import wraps
from typing import Union

from sqlalchemy.orm import MANYTOMANY, MANYTOONE, ONETOMANY


def handle_one_to_many_list(relation_cls, all_elements: list[dict]):
    elems_to_create = []
    updated_elems = []

    for elem in all_elements:
        elem_id = elem.get("id", None)

        existing_elem = relation_cls.get_ref(match_value=elem_id)

        if existing_elem is None:

            elems_to_create.append(elem)

        else:
            for key, value in elem.items():
                setattr(existing_elem, key, value)

            updated_elems.append(existing_elem)

    new_elems = []
    for elem in elems_to_create:
        new_elems = [relation_cls(**elem) for elem in all_elements]

    return new_elems


def auto_init(exclude: Union[set, list] = None):  # sourcery no-metrics
    """Wraps the `__init__` method of a class to automatically set the common
    attributes.

    Args:
        exclude (Union[set, list], optional): [description]. Defaults to None.
    """

    exclude = exclude or set()
    exclude.add("id")

    def decorator(init):
        @wraps(init)
        def wrapper(self, *args, **kwargs):  # sourcery no-metrics
            """
            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

            session = kwargs.get("session", None)

            for key, val in kwargs.items():
                if key in exclude:
                    continue

                if not hasattr(cls, key):
                    continue
                    # 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
                    use_list = relationships[key].uselist

                    if relation_dir == ONETOMANY.name and use_list:
                        instances = handle_one_to_many_list(relation_cls, val)
                        setattr(self, key, instances)

                    if relation_dir == ONETOMANY.name and not use_list:
                        instance = relation_cls(**val)
                        setattr(self, key, instance)

                    elif relation_dir == MANYTOONE.name and not use_list:
                        if isinstance(val, dict):
                            val = val.get("id")

                            if val is None:
                                raise ValueError(f"Expected 'id' to be provided for {key}")

                        if isinstance(val, (str, int)):
                            instance = relation_cls.get_ref(match_value=val, session=session)
                            setattr(self, key, instance)

                    elif relation_dir == MANYTOMANY.name:

                        if not isinstance(val, list):
                            raise ValueError(f"Expected many to many input to be of type list for {key}")

                        if len(val) > 0 and isinstance(val[0], dict):
                            val = [elem.get("id") for elem in val]

                        instances = [relation_cls.get_ref(elem, session=session) for elem in val]
                        setattr(self, key, instances)

            return init(self, *args, **kwargs)

        return wrapper

    return decorator

Usage

class AdminModel(SqlAlchemyBase, BaseMixins):
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)
    password = Column(String)
    is_superuser = Column(Boolean(), default=False)

    @auto_init(exclude={'is_superuser'})
    def __init__(self, **_):
        this.is_superuser = false

    @classmethod
    def get_ref(cls, match_value: str, match_attr: str = "id"):
        with SessionLocal() as session:
            eff_ref = getattr(cls, match_attr)
            return session.query(cls).filter(eff_ref == match_value).one_or_none()

```decorator

[khialb32]

Hello @hay-kot , i have tried your solution on nested models, and it works just fine, I will try with more nested pedantic models and see if it works, yet I would greatly appreciate if you could provide some explanation(documentation) of what you did because although I read it many times, I am still afraid to use it for production.

[hobbsAU]

SQLModel is fantastic as it cuts so much time out managing pydantic schemas and sqlalchemy models.

I too would like to know how to create database entries with incoming nested SQLModels.

Does anyone know how to do this yet?

[woprandi]

It's even worse than use separate pydantic and sqlalchemy models because the nested list on the payload received from client are always empty. So it's not even possible to create each element separarely

[scd75]

Agree with @hay-kot : it would be great to have the ability to instantiate nested objects from a nested SQLModel instance.
in the same vein as @hay-kot : i used with the past a custom constructor for SQLAlchemy Base class, and also proposed that to SQLAlchemey repo:
fastapi/fastapi#2194
sqlalchemy/sqlalchemy#6322
@tiangolo : any idea on how this could be achieved? Thanks!

[Fanna1119]

This would such an amazing feature. Any update on this?

[Sancho66]

Any update on this ?

[Sillocan]

This is a major feature that is missing from sqlmodel that is preventing myself from migrating a larger project from marshmallow. There is a lot of boilerplate code that needs to be added to handle nested objects

[zrajna]

I would also love to see this addition to the awesome SQLModel library.

[50Bytes-dev]

Started working with this wonderful library and encountered the same problem. Any updates?

[micoloth]

+1 on this.

First of all, this library looks amazing!

Last year, I was using Sqlalchemy and Pydantic models separately, but this library would be such a quality of life improvement..

Unfortunately, the fact that this feature is missing is a deal breaker, so i cannot start using it.

Last year, I also built a custom recursive solution to do this in Sqlalchemy+Pydantic, just like OP did.

But, I don't think it would be easy/a good idea to port it to SqlModels...

Any updates would be greatly appreciated!

[s-froghyar]

Any update on this? Would love to use hydra as a configuration management the way it is meant to be used (hierarchical and out of the box configurable with separation of concerns) rather than making a single flat yaml for all configs

[whugarra]

There any temporary guidance from best practices to avoid this issue?

[dataengineeringatfunderzgroup]

any updates on it? this is something that will be amazing for SQLModel to have.

[elliswilcox]

Hi, is there any updates or fixes for this?
Still having this issue in Nov 2024, a good 3 years after this issue was originally created.
I'm sure a lot of people (including me) would benefit from this being fixed.

@tiangolo

[UmarBelloKanwa]

Still facing these issues is there any updates about?

[muddassir-lateef]

+1 @tiangolo any updates for this feature. This is a great library but would love to have this feature.

[nathggns]

This makes it quite difficult to use SQLModel in LLM applications, as we need to build out our models from JSON output from an LLM. There doesn't seem to be an obviously supported way to do this right now.

[davletovalmir]

I ended up inheriting my SQLModel models from custom BaseTable class that is defined as shown below. Feel free to use and extend! It helped a lot in my project, hope someone else could also benefit from it!

class BaseTable(SQLModel):
    __abstract__ = True

    id: Optional[int] = Field(default=None, primary_key=True)

    # your other default columns / fields / methods here

    @classmethod
    def from_json(cls, json_str: str) -> Self:
        return cls.from_dict(json.loads(json_str))

    @classmethod
    def from_dict(cls, data: dict) -> Self:
        values = {}

        for field_name, raw_value in data.items():
            if field_name in {"id", "created_at", "updated_at"}:
                continue

            field_value = cls._build_field(field_name, raw_value)
            if field_value is None and raw_value is not None:  # skipped value
                continue
            values[field_name] = field_value

        return cls(**values)

    @classmethod
    def _build_field(cls, field_name: str, raw_value: Any) -> Any:
        if field_name in cls.model_fields:
            return raw_value

        rel = cls.__mapper__.relationships.get(field_name)
        if not rel:
            return None

        is_belongs_to_rel = f"{field_name}_id" in cls.model_fields
        if is_belongs_to_rel:
            return None

        FieldModel = rel.mapper.class_

        if rel.uselist:
            sub_items = []
            for item in raw_value:
                if isinstance(item, dict):
                    sub_items.append(FieldModel.from_dict(item))
                elif isinstance(item, FieldModel):
                    sub_items.append(item)
                else:
                    continue
            return sub_items

        if isinstance(raw_value, dict):
            return FieldModel.from_dict(raw_value)
        if isinstance(raw_value, FieldModel):
            return raw_value

        return None