Storing Pydantic models at database¶
In some cases you might need to be able to store Pydantic models as a JSON data instead of create new table for them. You can do it now with new SqlAlchemy type PydanticJSonType mapped to BaseModel inside SqlModel.
For example let's add some stats to our heroes and save them at database as JSON data.
At first, we need to create new class Stats inherited from pydantic BaseModel or even SqlModel:
class Stats(BaseModel):
strength: int
dexterity: int
constitution: int
intelligence: int
wisdom: int
charisma: int
Then create new field stats to Hero model
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int]
stats: Optional[Stats]
👀 Full tutorial preview
import random
from typing import Optional
from pydantic import BaseModel
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Stats(BaseModel):
strength: int
dexterity: int
constitution: int
intelligence: int
wisdom: int
charisma: int
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int]
stats: Optional[Stats]
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def random_stat():
random.seed()
return Stats(
strength=random.randrange(1, 20, 2),
dexterity=random.randrange(1, 20, 2),
constitution=random.randrange(1, 20, 2),
intelligence=random.randrange(1, 20, 2),
wisdom=random.randrange(1, 20, 2),
charisma=random.randrange(1, 20, 2),
)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", stats=random_stat())
hero_2 = Hero(
name="Spider-Boy", secret_name="Pedro Parqueador", stats=random_stat()
)
hero_3 = Hero(
name="Rusty-Man", secret_name="Tommy Sharp", age=48, stats=random_stat()
)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Deadpond")
results = session.exec(statement)
hero_1 = results.one()
print("Hero 1:", hero_1)
statement = select(Hero).where(Hero.name == "Rusty-Man")
results = session.exec(statement)
hero_2 = results.one()
print("Hero 2:", hero_2)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
Here we define new Pydantic model Stats contains statistics of our hero and map this model to SqlModel class Hero.
Then we create new instances of Hero model with random generated stats and save it at database.
How to watch for mapped model changes at runtime¶
In previous example we have one non bug but feature - Stats model isn't mutable and if we try to load our Hero form database and then change some stats and call session.commit() there no changes will be saved.
Let's see how to avoid it.
At first, we need to inherit our Stats model from sqlalchemy.ext.mutable.Mutable:
class Stats(BaseModel, Mutable):
strength: int
dexterity: int
constitution: int
intelligence: int
wisdom: int
charisma: int
@classmethod
Then map Stats to Hero as shown bellow:
stats: Stats = Field(
default_factory=None,
sa_column=Column(Stats.as_mutable(PydanticJSONType(type=Stats))),
)
After all of these actions we can change mutated model, and it will be saved to database after we call session.commit()
statement = select(Hero).where(Hero.name == "Deadpond")
results = session.exec(statement)
hero_1 = results.one()
print("Hero 1:", hero_1.stats)
hero_1.stats.strength = 100500
session.commit()
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Deadpond")
results = session.exec(statement)
hero_1 = results.one()
print("Hero 1 strength:", hero_1.stats.strength)
print("Hero 1:", hero_1)
👀 Full tutorial preview
import random
from typing import Any, Optional
from pydantic import BaseModel
from sqlalchemy import Column
from sqlalchemy.ext.mutable import Mutable
from sqlmodel import Field, Session, SQLModel, create_engine, select
from sqlmodel.sql.sqltypes import PydanticJSONType
class Stats(BaseModel, Mutable):
strength: int
dexterity: int
constitution: int
intelligence: int
wisdom: int
charisma: int
@classmethod
def coerce(cls, key: str, value: Any) -> Optional[Any]:
return value
def __setattr__(self, key, value):
# set the attribute
object.__setattr__(self, key, value)
# alert all parents to the change
self.changed()
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int]
stats: Stats = Field(
default_factory=None,
sa_column=Column(Stats.as_mutable(PydanticJSONType(type=Stats))),
)
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def random_stat():
random.seed()
return Stats(
strength=random.randrange(1, 20, 2),
dexterity=random.randrange(1, 20, 2),
constitution=random.randrange(1, 20, 2),
intelligence=random.randrange(1, 20, 2),
wisdom=random.randrange(1, 20, 2),
charisma=random.randrange(1, 20, 2),
)
def create_hero():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", stats=random_stat())
with Session(engine) as session:
session.add(hero_1)
session.commit()
def mutate_hero():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Deadpond")
results = session.exec(statement)
hero_1 = results.one()
print("Hero 1:", hero_1.stats)
hero_1.stats.strength = 100500
session.commit()
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Deadpond")
results = session.exec(statement)
hero_1 = results.one()
print("Hero 1 strength:", hero_1.stats.strength)
print("Hero 1:", hero_1)
def main():
create_db_and_tables()
create_hero()
mutate_hero()
if __name__ == "__main__":
main()