ORM API

Extended Declarative Base

sqlalchemy_mate.api.ExtendedBase is a Mixin class that enables many convenient methods for your ORM model.

import sqlalchemy as sa
import sqlalchemy.orm as orm
import sqlalchemy_mate.api as sam
from rich import print as rprint

Base = orm.declarative_base()


# add sqlalchemy_mate.ExtendedBase Mixin class
class User(Base, sam.ExtendedBase):
    __tablename__ = "users"

    id: orm.Mapped[int] = orm.mapped_column(sa.Integer, primary_key=True)
    name: orm.Mapped[str] = orm.mapped_column(sa.String, nullable=True)
    # you can also do this
    # id: int = sa.Column(sa.Integer, primary_key=True)
    # name: str = sa.Column(sa.String, nullable=True)

    # put important columns here
    # you can choose to print those columns only with ``.glance()`` method.
    _settings_major_attrs = [
        id,
    ]

Python dict liked interface

A data model should be have a nicer print.

user = User(id=1, name="Alice")
print(user)

User(id=1, name='Alice')

Convert data model to / from generic python data type should be easy.

rprint(User.keys())

['id', 'name']

rprint(user.keys())

['id', 'name']

rprint(user.values())

[1, 'Alice']

rprint(user.items())

[('id', 1), ('name', 'Alice')]

rprint(user.to_dict())

{'id': 1, 'name': 'Alice'}

rprint(user.to_OrderedDict())

OrderedDict([('id', 1), ('name', 'Alice')])

rprint(user.pk_values())

(1,)

user.glance()  # only print important columns

User(id=1)

Absorb and Revise

Python dict can update values based on another dict, A data model should be able to do it to.

# update values based on another data model, it is similar to ``dict.update()``
user_bob = User(name="Bob")
user.absorb(user_bob)
rprint(user)

# update values based on generic python dict, it is similar to ``dict.update()``
user.revise({"name": "Cathy"})
rprint(user)

User(id=1, name='Bob')

User(id=1, name='Cathy')

Insert, Select, Update with ORM Model

engine = sa.create_engine("sqlite:///:memory:")
sam.test_connection(engine, timeout=3)

Base.metadata.create_all(engine)
# Delete all data, make sure we have a fresh start
User.delete_all(engine)

# Count rows in a table
print(User.count_all(engine))

0

# Bulk insert
user_list = [
    User(id=57),
    User(id=264),
    User(id=697),
]
User.smart_insert(engine, user_list)
rprint(User.count_all(engine))

3

# Get single object by primary key values
user = User.by_pk(engine, 57)
rprint(user)  # User(id=57)

User(id=57, name=None)

Smart Insert Handles Primary Key conflicts efficiently

# Bulk insert, handle primary key conflicts efficiently
user_list = [User(id=id_) for id_ in range(1, 1000 + 1)]
User.smart_insert(engine, user_list)
rprint(User.count_all(engine))  # 1000

1000

Bulk Upsert (Insert + Update) should be easy

# before upsert
rprint(User.by_pk(engine, 999))
rprint(User.by_pk(engine, 1000))
rprint(User.by_pk(engine, 1001))
rprint(User.by_pk(engine, 1002))

User(id=999, name=None)

User(id=1000, name=None)

None

None

# Bulk update + insert, locate rows by primary key values
user_list = [
    User(id=999, name="Alice"),
    User(id=1000, name="Bob"),
    User(id=1001, name="Cathy"),
    User(id=1002, name="David"),
]
User.upsert_all(engine, user_list)

(2, 2)

# after upsert
rprint(User.by_pk(engine, 999))
rprint(User.by_pk(engine, 1000))
rprint(User.by_pk(engine, 1001))
rprint(User.by_pk(engine, 1002))

User(id=999, name='Alice')

User(id=1000, name='Bob')

User(id=1001, name='Cathy')

User(id=1002, name='David')

Run raw SQL query

The Class.by_sql() assumes that the returned row is the given ORM model. You cannot do this with JOIN.

results = User.by_sql(
    engine,
    sql="""
    SELECT *
    FROM users
    WHERE users.id >= 999
    """,
)
rprint(results)

[User(id=999, name='Alice'), User(id=1000, name='Bob'), User(id=1001, name='Cathy'), User(id=1002, name='David')]