SQLAlchemy Mixins

2.0.5 · active · verified Wed Apr 15

SQLAlchemy Mixins provides an Active Record pattern, Django-like query methods, nested eager loading, and enhanced `__repr__` for SQLAlchemy models. It aims to simplify common ORM operations by adding convenient class methods. The current stable version is 2.0.5, with releases occurring semi-regularly as features and fixes are added.

Warnings

breaking Version 2.0.0 of `sqlalchemy-mixins` introduced support for SQLAlchemy 2.0. While it aims for backward compatibility with SQLAlchemy 1.4+, applications still using older SQLAlchemy 1.x specific patterns (e.g., `create_engine(..., strategy='threadlocal')` or explicit query objects without `.scalars()`) may encounter breaking changes or require migration.
Fix: Ensure your application's SQLAlchemy usage aligns with SQLAlchemy 2.0's API style, even if running on SQLAlchemy 1.4 in compatibility mode. Consider upgrading SQLAlchemy to version 2.0 alongside `sqlalchemy-mixins`.
gotcha By default, `ActiveRecordMixin` methods like `create`, `save`, `update`, and `delete` do not automatically commit changes to the database session. Prior to `v2.0.5`, manual `session.commit()` calls were always required. From `v2.0.5` onwards, a convenient `commit=True` keyword argument was added to these methods, but if omitted, manual commitment is still necessary.
Fix: Always explicitly call `session.commit()` after operations that modify data or ensure you consistently use the `commit=True` argument when available in versions 2.0.5 and newer.
gotcha The `set_session()` method assigns a global session to the model class for `ActiveRecordMixin` operations. In concurrent environments (e.g., multi-threaded, async web applications), this can lead to issues where objects belong to different sessions, or transactions are not isolated correctly. While `v2.1.0` adds async support, proper session management is crucial for `v2.0.5` and earlier.
Fix: For concurrent applications, consider using SQLAlchemy's `scoped_session` for managing sessions per-thread/request, or ensure sessions are explicitly passed and managed within your application's context (e.g., via dependency injection or a request context).

Install

pip install sqlalchemy-mixins Install stable version

Imports

ActiveRecordMixin

from sqlalchemy_mixins import ActiveRecordMixin

InspectionMixin

from sqlalchemy_mixins import InspectionMixin

Quickstart

This quickstart demonstrates how to set up `ActiveRecordMixin` with a SQLAlchemy declarative model. It covers common CRUD operations (create, find, update, delete) using the mixin's class methods, as well as basic query filtering. It highlights the use of `commit=True` which simplifies transaction management from version 2.0.5 onwards.

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy_mixins import ActiveRecordMixin
import os

# Setup SQLAlchemy engine and session (using in-memory sqlite for example)
# For a real app, you might get this from an environment variable
database_url = os.environ.get('DATABASE_URL', 'sqlite:///test.db')
engine = create_engine(database_url)
Session = sessionmaker(bind=engine)
session = Session()

# Define the declarative base for models
Base = declarative_base()

# Define a User model inheriting from Base and ActiveRecordMixin
class User(Base, ActiveRecordMixin):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String, unique=True)

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}')>"

# Create database tables
Base.metadata.create_all(engine)

# Set the session for ActiveRecordMixin for this model
User.set_session(session)

# --- Usage Examples ---

# 1. Create a new user (commit=True is available from v2.0.5)
user1 = User.create(name='Alice', email='alice@example.com', commit=True)
print(f"Created user: {user1}")

# 2. Find a user by ID
found_user = User.find(user1.id)
print(f"Found user by ID: {found_user}")

# 3. Find a user by attributes (Django-like filter)
filtered_user = User.where(name='Alice').first()
print(f"Found user by name: {filtered_user}")

# 4. Update a user
user1.update(name='Alicia', commit=True)
print(f"Updated user: {user1}")

# 5. Get all users
all_users = User.all()
print(f"Total users: {len(all_users)}")

# 6. Delete a user
user1.delete(commit=True)
print(f"User {user1.name} deleted.")

# Clean up and close the session
session.close()

view raw JSON →