SQLModel CRUD Utils

A set of CRUD utilities designed to streamline and expedite common database operations when using SQLModel, offering both synchronous and asynchronous support.

PyPI Version CI Status License Python
Get Started API Documentation

Overview

sqlmodel-crud-utils provides a convenient layer on top of SQLModel and SQLAlchemy to simplify common database interactions. It offers both synchronous and asynchronous functions for creating, reading, updating, and deleting data, along with helpers for bulk operations, filtering, pagination, and relationship loading. The goal is to reduce boilerplate code in projects using SQLModel.

What's New in v0.2.0

Version 0.2.0 brings significant enhancements focused on developer experience and production-ready features:

🚀 Public API Exports

No more deep imports! All functions are now available directly from the package:

Python
# Before (v0.1.0)
from sqlmodel_crud_utils.sync import get_row, update_row
from sqlmodel_crud_utils.a_sync import get_row as a_get_row

# After (v0.2.0)
from sqlmodel_crud_utils import get_row, update_row, a_get_row

⚠️ Custom Exception Hierarchy

Better error handling with detailed, context-aware exceptions:

  • RecordNotFoundError - When a record doesn't exist
  • MultipleRecordsError - When one record expected but multiple found
  • ValidationError - For data validation failures
  • BulkOperationError - For bulk operation failures with detailed stats
  • TransactionError - For transaction-related issues

🔒 Transaction Context Managers

Safer database operations with automatic commit and rollback:

Python
from sqlmodel_crud_utils import transaction, write_row, update_row

with transaction(session) as tx:
    user = write_row(User(name="Alice"), tx)
    update_row(user.id, {"email": "alice@example.com"}, User, tx)
    # Automatically commits on success, rolls back on error

📝 Audit Trail Mixins

Automatic timestamp tracking for record creation and updates:

Python
from sqlmodel_crud_utils import AuditMixin

class User(SQLModel, AuditMixin, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    # Automatically adds: created_at, updated_at, created_by, updated_by

🗑️ Soft Delete Support

Mark records as deleted without actually removing them:

Python
from sqlmodel_crud_utils import SoftDeleteMixin

class Product(SQLModel, SoftDeleteMixin, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    # Automatically adds: is_deleted, deleted_at, deleted_by

product.soft_delete(user="admin")  # Mark as deleted
product.restore()                  # Restore it
✅ 100% Backward Compatible - All v0.1.0 code continues to work without changes!

Key Features

⚡

Sync & Async Support

Parallel functions for both synchronous and asynchronous operations with simple a_ prefix for async versions.

📦

Simplified CRUD

High-level functions for common operations: get, create, update, delete, bulk operations, and upserts.

🔍

Flexible Filtering

Supports exact matches and comparisons (__like, __gte, __lte, __in) with keyword arguments.

📄

Built-in Pagination

Easy pagination support with offset and limit parameters for efficient data retrieval.

🔗

Relationship Loading

Supports eager loading (selectinload) and lazy loading (lazyload) for related data.

🎯

Type-Safe

Full type hints throughout for excellent IDE support and static type checking.

🔄

Dialect-Specific Upsert

Automatically uses the correct upsert syntax for PostgreSQL, SQLite, or MySQL based on configuration.

🛡️

Exception Handling

Custom exception hierarchy with detailed context for better error handling and debugging.

📊

Audit Trails

Built-in mixins for automatic timestamp and user tracking on all database operations.

Quick Start

Installation

Bash
pip install sqlmodel-crud-utils
# Or using uv:
uv pip install sqlmodel-crud-utils

Configuration

Set the SQL_DIALECT environment variable for upsert functionality:

Bash / .env
export SQL_DIALECT=postgresql  # or sqlite, mysql, etc

Basic Usage Examples

Simple CRUD Operations

Python
from sqlmodel import Session, SQLModel, create_engine, Field
from sqlmodel_crud_utils import get_one_or_create, get_rows, write_row

class MyModel(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    value: int | None = None

DATABASE_URL = "sqlite:///./mydatabase.db"
engine = create_engine(DATABASE_URL)
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    # Get or create an instance
    instance, created = get_one_or_create(
        session_inst=session,
        model=MyModel,
        name="Test Item",
        create_method_kwargs={"value": 123}
    )

    # Query with filtering
    success, rows = get_rows(
        session_inst=session,
        model=MyModel,
        value__gte=100,
        sort_field="name"
    )
    for row in rows:
        print(f"{row.name}: {row.value}")

Exception Handling

Python
from sqlmodel_crud_utils import (
    get_row,
    RecordNotFoundError,
    update_row,
    ValidationError
)

try:
    success, user = get_row(
        id_str=999,
        session_inst=session,
        model=User
    )
    if not success:
        raise RecordNotFoundError(
            model=User,
            id_value=999
        )
except RecordNotFoundError as e:
    print(f"Error: {e}")
    print(f"Model: {e.model.__name__}")
    print(f"ID: {e.id_value}")

Transactions

Python
from sqlmodel_crud_utils import (
    transaction,
    write_row,
    update_row,
    TransactionError
)

with Session(engine) as session:
    try:
        with transaction(session) as tx:
            # All operations succeed together
            # or all are rolled back
            user = write_row(
                User(name="Alice"),
                tx
            )
            profile = write_row(
                Profile(user_id=user.id),
                tx
            )
            update_row(
                user.id,
                {"verified": True},
                User,
                tx
            )
            # Commits automatically
    except TransactionError as e:
        print(f"Transaction failed: {e}")
        # Rolled back automatically

Async Operations

Python
from sqlmodel_crud_utils import (
    a_get_rows,
    a_write_row,
    a_transaction
)
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine(
    "sqlite+aiosqlite:///./db.db"
)

async with AsyncSession(async_engine) as session:
    # Async query
    success, users = await a_get_rows(
        session_inst=session,
        model=User,
        is_active=True
    )

    # Async transaction
    async with a_transaction(session) as tx:
        user = await a_write_row(
            User(name="Bob"),
            tx
        )

Audit Mixins

Python
from sqlmodel_crud_utils import (
    AuditMixin,
    write_row
)

class User(SQLModel, AuditMixin, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    # Auto-adds: created_at, updated_at,
    # created_by, updated_by

with Session(engine) as session:
    user = User(
        name="Alice",
        created_by="admin"
    )
    user = write_row(user, session)

    # Timestamps set automatically
    print(f"Created: {user.created_at}")

    # Update tracking
    user.email = "alice@example.com"
    user.updated_by = "admin"
    session.add(user)
    session.commit()
    print(f"Updated: {user.updated_at}")

Soft Deletes

Python
from sqlmodel_crud_utils import (
    SoftDeleteMixin,
    get_rows
)

class Product(SQLModel, SoftDeleteMixin, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    # Auto-adds: is_deleted, deleted_at,
    # deleted_by

with Session(engine) as session:
    product = Product(name="Widget")
    product = write_row(product, session)

    # Soft delete
    product.soft_delete(user="admin")
    session.add(product)
    session.commit()

    # Restore
    product.restore()
    session.commit()

    # Query non-deleted only
    success, active = get_rows(
        session_inst=session,
        model=Product,
        is_deleted=False
    )

Next Steps

📚 Explore Use Cases

See real-world scenarios and implementation patterns for REST APIs, microservices, admin dashboards, and more.

View Use Cases

🍳 Browse Recipes

Practical code patterns for complex filtering, pagination, batch processing, and performance optimization.

View Recipes

📖 API Reference

Complete API documentation with detailed function signatures, parameters, and return types.

API Docs