Declarative Static Data to be Created on Function Call

[dl-lim]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question 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

from sqlmodel import create_engine, Session, SQLModel, select, Field

class Hero(SQLModel, table=True):
    hero_id: int | None = Field(default=None, primary_key=True)
    hero_name: str
    hero_avatar: str

class Team(SQLModel, table=True):
    team_id: int | None = Field(default=None, primary_key=True)
    team: str


class Item(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    description: str


def get_engine():
    return create_engine("sqlite:///database.db")

def create_db_and_tables():
    engine = get_engine()
    SQLModel.metadata.create_all(engine)

def create_static_data():
    engine = get_engine()
    with Session(engine) as db:
        # heros = [
        #     Hero(hero_id=1, hero_name="Deadpond", hero_avatar="Dag Worthlows.png"),
        #     Hero(hero_id=2, hero_name="Spider-boy", hero_avatar="Pedro Parquet.png"),
        #     Hero(hero_id=3, hero_name="Rusty-Man", hero_avatar="Tommy Spark.png"),
        # ]
        # _sync_static_data(db, Hero, heros)

        # teams = [
        #     Team(team_id=1, team="Apprehenders"),
        #     Team(team_id=2, team="Preventers"),
        # ]
        # _sync_static_data(db, Team, teams)

        items = [
            Item(id=1, name="Item A", description="Lorem ipsum"),
            Item(id=2, name="Item B", description="dolor"),
            Item(id=3, name="Item C", description="sit amet"),
        ]
        _sync_static_data(db, Item, items)

def _sync_static_data(db, model, data):

    existing_entries = {item.id: item for item in db.exec(select(model)).all()}
    new_entries = {item.id: item for item in data}

    # Delete items that are not in the new list
    for id in list(existing_entries.keys()):
        if id not in new_entries:
            item_to_delete = existing_entries.pop(id)
            db.delete(item_to_delete)

    # Add or update items
    for id, new_item in new_entries.items():
        if id in existing_entries:
            # Update existing item
            existing_item = existing_entries[id]
            for key, value in new_item.model_dump(exclude_unset=True).items():
                setattr(existing_item, key, value)
        else:
            # Add new item
            new_item_instance = model(**new_item.model_dump())
            db.add(new_item_instance)

    db.commit()


if __name__ == '__main__':
    create_db_and_tables()
    create_static_data()

Description

Consider the above sample code. I would call create_static_data() in the lifespan module of fastapi.

I'm looking for a way to declaratively define some static data at the beginning of each run.

It aims to be a more of a MERGE statement rather than an INSERT if blank situation. If it exists and is different, update it. If it doesn't exist, create it. If it exists but is not in the new list, delete it.

If at any time anyone changes the code in create_static_data(), it should change the static data in the database too.

In the above codebase, I have a sample _sync_static_data() which works with the Items model, given that the primary key is just id. In an ideal world, all models can just have id as their primary key and this function would work.

However, if you uncomment the heroes and teams declarations, the _sync_static_data() code will error as their id's are hero_id and team_id respectively.

What's a better implementation to resolve this? Possible Solutions:

1. I've considered extracting primary keys only, but there may be more than one primary key with link tables or other situations. This approach may also be quite clunky (and I've not tested this deeply yet)
2. Is there a way to perform the MERGE statement or _sync_static_data() more seamlessly and natively?

Operating System

Linux, Windows

Operating System Details

WSL

SQLModel Version

0.0.18

Python Version

3.10.5

Additional Context

No response

[monchin]

I'm not sure whether I understood your requirement correctly, but I guess you can have a look at if this is what you need

from sqlmodel import SQLModel, Field

class Hero(SQLModel, table=True):
    hero_id: int | None = Field(default=None, primary_key=True)
    hero_name: str
    hero_avatar: str

item = Hero(hero_id=1, hero_name="John", hero_avatar="https://example.com/avatar.jpg")
item_class = str(type(item)).strip("<").strip(">").strip("'").rsplit(".")[-1]
item_id = getattr(item, f"{item_class.lower()}_id")
print(item_id)

[dl-lim]

Hmm, thanks, that's one way, I suppose - an approach to the Possible Solution 1 described in the opening post - and it has a strict requirement that the id field has to have the same form as f"{item_class.lower()}_id".

Consider:

class EconomicHero(SQLModel, table=True):
    ecohero_id: int | None = Field(default=None, primary_key=True)
    ...

then the described approach wouldn't work.

Hoping to spark more ideas and other more elegant way to achieve declaration of objects in a db, thanks!

[monchin]

I think it's difficult because the program can't know what your primary key name is, and for one table, there may be more than one primary keys.

So why don't you set all primary key name as id?

[dl-lim]

So why don't you set all primary key name as id?

Several reasons - if you have downstream applications that already use a specific hero_id, changing that would mean mutating the data.

When you have more models and if you ever had to do a join on them, just id is going to get very confusing very quickly too.

But these are pros and cons of different approaches and perhaps in some situations just id is acceptable.

[monchin]

I guess it would better to ask on stackoverflow or sqlalchemy, the main point is whether there is a way to get the primary key of each sqlalchemy model