tiangolo
Flexibly Create Nested Database Entries from Incoming Pydantic/SQLModels
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 SQLModel documentation, with the integrated search.
- [X] I already searched in Google "How to X in SQLModel" 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 SQLModel but to Pydantic.
- [X] I already checked if it is not related to SQLModel but to SQLAlchemy.
Commit to Help
- [X] 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.
- Is this supported?
- If no, is this a use-case you've thought of?
- 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
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.
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?
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
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: https://github.com/tiangolo/fastapi/issues/2194 https://github.com/sqlalchemy/sqlalchemy/discussions/6322 @tiangolo : any idea on how this could be achieved? Thanks!
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
Started working with this wonderful library and encountered the same problem. Any updates?
+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!
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
