Flexibly Create Nested Database Entries from Incoming Pydantic/SQLModels #6
Description
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.
- 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 decoratorUsage
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