ORM API

Extended Declarative Base

sqlalchemy_mate.ExtendedBase is a Mixin class that enables many convenient method for your ORM model. In this

import sqlalchemy as sa
from sqlalchemy.orm import declarative_base
import sqlalchemy_mate as sam

Base = declarative_base()

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

    id = sa.Column(sa.Integer, primary_key=True)
    name = 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, ]

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:

print(User.keys()) # class would work too, # ['id', 'name']
print(user.keys()) # ['id', 'name']
print(user.values()) # [1, 'Alice']
print(user.items()) # [('id', 1), ('name', 'Alice')]
print(user.to_dict()) # {'id': 1, 'name': 'Alice'}
print(user.to_OrderedDict()) # OrderedDict([('id', 1), ('name', 'Alice')])
print(user.pk_values()) # (1,)
user.glance() # User(id=1)

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
user_bob = User(name="Bob")
user.absorb(user_bob)
print(user) # User(id=1, name='Bob')

# update values based on generic python dict
user.revise({"name": "Cathy"})
print(user) # User(id=1, name='Cathy')

Insert, Select, Update

Talk is cheap, show me the Code.

from sqlalchemy.orm import Session

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)
print(User.count_all(engine)) # 3

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

# 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)
print(User.count_all(engine)) # 1000

# 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)

print(User.by_pk(engine, 999).name) # Alice
print(User.by_pk(engine, 1001).name) # Cathy
print(User.count_all(engine)) # 1002

# Run raw SQL query
results = User.by_sql(
    engine,
    sql="""
    SELECT *
    FROM users
    WHERE users.id >= 999
    """
)
# [User(id=999, name='Alice'), User(id=1000, name='Bob'), User(id=1001, name='Cathy'), User(id=1002, name='David')]
print(results)