sadel 0.1.0

Helper class for upserting records to database

Sadel

Sadel is a helper class for upserting records with SQLModel. Sadel combines SQLAlchemy ('sa') and SQLmodel ('del'), and in Danish, 'sadel' means 'saddle' symbolizing taking over the burden of managing upserts.

Installation

pip install sadel

Features

• Upsert and batch_upsert functions.
• For auditing, automatically adds and manages created_on and modified_on columns to your table (timezones are supported).
• Validates your data before upserting using Pydantic validate_model method (not supported in SQLModel)
• Asyncio
• Compatible with Alembic
• Specify the (PK) columns to use for upserting using _upsert_index_elements attribute
• Ignore specific columns from updating using _upsert_exclude_fields attribute

Example upsert

from sadel import Sadel
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel import Field, create_engine, select, or_
from sqlmodel.ext.asyncio.session import AsyncSession

class Hero(Sadel, table=True):
    __tablename__ = "hero" 
    _upsert_index_elements = {"id"}

    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None
    
sqlite_url = f"sqlite+aiosqlite::///database.db" 
engine = create_engine(sqlite_url, echo=True, future=True)

async_engine = create_async_engine(sqlite_url_async, echo=True, future=True)
hero = Hero(name="Deadpond", secret_name="Dive Wilson")

async with AsyncSession(async_engine) as session:
    # Upsert the record
    await Hero.upsert(hero, session)
    
    # Fetch the upserted record
    result = (
        (await session.exec(select(Hero).where(Hero.name == "Deadpond")))
        .all()
    )

    print(result)

Output:

[Hero(id=1, name='Deadpond', secret_name='Dive Wilson', age=None, created_on=datetime.datetime(2024, 8, 1, 19, 39, 7), modified_on=None)]

Example batch upsert

hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)

async with AsyncSession(async_engine) as session:
    await Hero.batch_upsert([hero_1, hero_2, hero_3], session)
    result = (
        (
            await session.exec(
               select(Hero).where(or_(Hero.name == "Deadpond",  Hero.name == "Spider-Boy", Hero.name == "Rusty-Man"))
            )
        )
        .all()
    )
    print(result)

Output:

[Hero(id=1, name='Deadpond', secret_name='Dive Wilson',age=None, created_on=datetime.datetime(2024, 8, 1, 19, 39, 7), modified_on=None), 
Hero(id=2, name='Spider-Boy", secret_name='Pedro Parqueador',age=None, created_on=datetime.datetime(2024, 8, 1, 19, 39, 7), modified_on=None),
Hero(id=3, name='Rusty-Man', secret_name='Tommy Sharp', age=48, created_on=datetime.datetime(2024, 8, 1, 19, 39, 7), modified_on=None)]

Example update record

async with AsyncSession(async_engine) as session:
    # Upsert the record
    hero = Hero(name="Deadpond", secret_name="Dive Wilson", age=25)
    await Hero.upsert(hero, session)

    # Update the record
    hero.age = 30
    # Upsert the updated record
    await Hero.upsert(hero, session)

    # Fetch the updated record
    result = (
        (await session.exec(select(Hero).where(Hero.name == "Deadpond")))
        .scalars()
        .all()
    )

    print(result)

Output:

[Hero(id=1, name='Deadpond', secret_name='Dive Wilson', age=30, created_on=datetime.datetime(2024, 8, 1, 19, 39, 7), modified_on=datetime.datetime(2024, 8, 1, 19, 39, 8))]

Contributing

• Fork the repository
• Create a new branch
• Make your changes
• Raise a PR

Development

# install dependencies
rye sync 
# run tests, linting, formatting, and type checking, 
rye run all

License

This project is licensed under the terms of the MIT License