sqlmodel-crud-utilities 0.2.1

A set of CRUD utilities to expedite operations with SQLModel

SQLMODEL_CRUD_UTILS

A set of CRUD (Create, Read, Update, Delete) utilities designed to streamline and expedite common database operations when using SQLModel, offering both synchronous and asynchronous support.

Built with the tools and technologies:

Table of Contents

• Overview
• What's New in v0.2.0
• Features
• Project Structure
  • Project Index
• Getting Started
  • Prerequisites
  • Configuration
  • Installation
  • Usage
  • Testing
• Project Roadmap
• Contributing
• License
• Acknowledgments

---

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:

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

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": "[e-mail]"}, User, tx)
    # Automatically commits on success, rolls back on error

Audit Trail Mixins

Automatic timestamp tracking for record creation and updates:

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:

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!

---

Features

• Sync & Async Support: Provides parallel functions in sqlmodel_crud_utils.sync and sqlmodel_crud_utils.a_sync.
• Public API Exports: Simple imports from the main package with a_ prefix for async functions.
• Simplified CRUD: Offers high-level functions:
  • get_one_or_create: Retrieves an existing record or creates a new one.
  • get_row: Fetches a single row by primary key.
  • get_rows: Fetches multiple rows with flexible filtering, sorting, and pagination.
  • get_rows_within_id_list: Fetches rows matching a list of primary keys.
  • update_row: Updates fields of an existing row.
  • delete_row: Deletes a row by primary key.
  • write_row: Inserts a single new row.
  • insert_data_rows: Inserts multiple new rows with fallback for individual insertion on bulk failure.
  • bulk_upsert_mappings: Performs bulk insert-or-update operations (dialect-aware).
• Custom Exception Hierarchy: Detailed exceptions for better error handling and debugging.
• Transaction Context Managers: Safe transaction handling with automatic commit/rollback.
• Audit Trail Mixins: Automatic timestamp and user tracking (AuditMixin).
• Soft Delete Support: Mark records as deleted without removing them (SoftDeleteMixin).
• Relationship Loading: Supports eager loading (selectinload) and lazy loading (lazyload) via parameters in get_row and get_rows.
• Flexible Filtering: get_rows supports filtering by exact matches (filter_by) and common comparisons (__like, __gte, __lte, __gt, __lt, __in) using keyword arguments.
• Pagination: Built-in pagination for get_rows.
• Dialect-Specific Upsert: Automatically uses the correct upsert syntax (e.g., ON CONFLICT DO UPDATE for PostgreSQL/SQLite) based on the SQL_DIALECT environment variable.
• Type-Safe: Full type hints for excellent IDE support and type checking.

---

Project Structure

└── sqlmodel_crud_utils/
    ├── __init__.py          # Public API exports
    ├── a_sync.py            # Asynchronous CRUD functions
    ├── sync.py              # Synchronous CRUD functions
    ├── utils.py             # Shared utilities
    ├── exceptions.py        # Custom exception hierarchy
    ├── transactions.py      # Transaction context managers
    └── mixins.py            # Audit and soft-delete mixins

Project Index

sqlmodel_crud_utils/

__root__

__init__.py	Public API exports for easy importing of all CRUD functions, exceptions, mixins, and transaction managers.
a_sync.py	Contains asynchronous versions of the CRUD utility functions, designed for use with `asyncio` and async database drivers (e.g., `aiosqlite`, `asyncpg`).
sync.py	Contains synchronous versions of the CRUD utility functions for standard execution environments.
utils.py	Provides shared helper functions used by both `sync.py` and `a_sync.py`, such as environment variable retrieval and dynamic dialect-specific import logic for upsert statements.
exceptions.py	Custom exception hierarchy for better error handling including RecordNotFoundError, ValidationError, BulkOperationError, and TransactionError.
transactions.py	Transaction context managers for safe database operations with automatic commit and rollback functionality.
mixins.py	Reusable mixins for common patterns like audit trails (AuditMixin) and soft deletes (SoftDeleteMixin).

---

Getting Started

Prerequisites

• Python: Version 3.9+ required.
• Database: A SQLAlchemy-compatible database (e.g., PostgreSQL, SQLite, MySQL).
• SQLModel: Your project should be using SQLModel for ORM definitions.

Configuration

This package requires the SQL_DIALECT environment variable to be set for the upsert functionality to work correctly across different database backends.

Set it in your environment:

export SQL_DIALECT=postgresql # or sqlite, mysql, etc

Or add it to a .env file in your project root (will be loaded automatically via python-dotenv):

SQL_DIALECT=postgresql

Refer to SQLAlchemy Dialects for a list of supported dialect names.

Installation

Install from PyPI (Recommended):

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

Build from source:

1. Clone the sqlmodel_crud_utils repository:

git clone https://github.com/fsecada01/SQLModel-CRUD-Utilities.git

2. Navigate to the project directory:

cd sqlmodel_crud_utils

3. Install the project dependencies:

uv pip install -r core_requirements.txt
# For testing/development
uv pip install -r dev_requirements.txt

(Alternatively, use pip install -r requirements.txt && pip install .)

Usage

Import the desired functions from the main package and use them with your SQLModel session and models.

Basic CRUD Operations (Synchronous)

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

# Define your model
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}
    )
    print(f"Instance ID: {instance.id}, Was created: {not created}")

    # Get rows matching criteria
    success, rows = get_rows(
        session_inst=session,
        model=MyModel,
        value__gte=100,
        sort_field="name"
    )
    if success:
        print(f"Found {len(rows)} rows with value >= 100:")
        for row in rows:
            print(f"- {row.name} (ID: {row.id})")

Using Public API Imports

# Import everything you need from the main package
from sqlmodel_crud_utils import (
    # Sync functions
    get_row, get_rows, write_row, update_row, delete_row,
    # Async functions (with a_ prefix)
    a_get_row, a_get_rows, a_write_row, a_update_row,
    # Exceptions
    RecordNotFoundError, ValidationError,
    # Transaction managers
    transaction, a_transaction,
    # Mixins
    AuditMixin, SoftDeleteMixin
)

Exception Handling

from sqlmodel_crud_utils import get_row, RecordNotFoundError

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}")  # User with id=999 not found
    print(f"Model: {e.model.__name__}")  # Access exception details
    print(f"ID: {e.id_value}")

Transaction Context Managers

Synchronous:

from sqlmodel_crud_utils import transaction, write_row, update_row

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", email="[e-mail]"), tx)
            profile = write_row(Profile(user_id=user.id, bio="Developer"), tx)
            update_row(user.id, {"verified": True}, User, tx)
            # Automatically commits here if no exceptions
    except TransactionError as e:
        print(f"Transaction failed: {e}")
        # Automatically rolled back

Asynchronous:

from sqlmodel_crud_utils import a_transaction, a_write_row, a_update_row
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine

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

async with AsyncSession(async_engine) as session:
    try:
        async with a_transaction(session) as tx:
            user = await a_write_row(User(name="Bob"), tx)
            await a_update_row(user.id, {"email": "[e-mail]"}, User, tx)
            # Automatically commits here if no exceptions
    except TransactionError as e:
        print(f"Transaction failed: {e}")
        # Automatically rolled back

Using AuditMixin

from datetime import datetime
from sqlmodel import SQLModel, Field
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
    email: str
    # AuditMixin automatically adds:
    # - created_at: datetime
    # - updated_at: datetime | None
    # - created_by: str | None
    # - updated_by: str | None

with Session(engine) as session:
    # Create user with audit tracking
    user = User(name="Alice", email="[e-mail]", created_by="admin")
    user = write_row(user, session)

    # created_at is automatically set to current UTC time
    print(f"User created at: {user.created_at}")

    # When updating
    user.email = "[e-mail]"
    user.updated_by = "admin"
    session.add(user)
    session.commit()
    session.refresh(user)

    # updated_at is automatically updated
    print(f"User updated at: {user.updated_at}")

Using SoftDeleteMixin

from sqlmodel import SQLModel, Field
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
    price: float
    # SoftDeleteMixin automatically adds:
    # - is_deleted: bool
    # - deleted_at: datetime | None
    # - deleted_by: str | None

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

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

    print(f"Deleted: {product.is_deleted}")  # True
    print(f"Deleted at: {product.deleted_at}")
    print(f"Deleted by: {product.deleted_by}")  # admin

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

    print(f"Deleted: {product.is_deleted}")  # False

    # Query non-deleted products
    from sqlmodel import select
    success, products = get_rows(
        session_inst=session,
        model=Product,
        is_deleted=False  # Filter out soft-deleted records
    )

Combining Mixins

class Order(SQLModel, AuditMixin, SoftDeleteMixin, table=True):
    id: int | None = Field(default=None, primary_key=True)
    customer_id: int
    total: float
    # Now has both audit trail AND soft delete support!

with Session(engine) as session:
    order = Order(customer_id=1, total=99.99, created_by="system")
    order = write_row(order, session)

    # Track creation
    print(f"Order created at {order.created_at} by {order.created_by}")

    # Soft delete with tracking
    order.soft_delete(user="admin")
    session.commit()

    print(f"Order deleted at {order.deleted_at} by {order.deleted_by}")

Testing

Ensure development dependencies are installed (uv pip install -r dev_requirements.txt or pip install -r dev_requirements.txt).

Run the test suite using pytest:

python -m pytest

This will execute all tests in the tests/ directory and provide coverage information based on the pytest.ini or pyproject.toml configuration.

---

Project Roadmap

• Alpha Release: Initial working version with core CRUD functions.
• Testing: Achieve 100% test coverage via Pytest.
• CI/CD: Implement GitHub Actions for automated testing, build, and release.
• Beta Release: Refine features based on initial testing and usage.
• v0.2.0 Release: Public API, exceptions, transactions, audit trails, soft deletes.
• Community Feedback: Solicit feedback from users.
• 360 Development Review: Comprehensive internal review of code, docs, and tests.
• Official 1.0 Release: Stable release suitable for production use.

---

Contributing

Contributions are welcome! Please feel free to submit issues, feature requests, or pull requests.

• 💬 Join the Discussions: Share your insights, provide feedback, or ask questions.
• 🐛 Report Issues: Submit bugs found or log feature requests for the sqlmodel_crud_utils project.
• 💡 Submit Pull Requests: Review open PRs, and submit your own PRs.

Contributing Guidelines

1. Fork the Repository: Start by forking the project repository to your GitHub account.
2. Clone Locally: Clone the forked repository to your local machine.

   git clone https://github.com/fsecada01/SQLModel-CRUD-Utilities.git
   

3. Create a New Branch: Always work on a new branch for your changes.

      git checkout -b feature/your-new-feature
   

4. Make Your Changes: Implement your feature or bug fix. Add tests!
5. Test Your Changes: Run pytest to ensure all tests pass.
6. Format and Lint: Ensure code follows project standards (e.g., using black, ruff, pre-commit).
7. Commit Your Changes: Commit with a clear and concise message.

   git commit -m "feat: Implement the new feature."
   

8. Push to GitHub: Push the changes to your forked repository.

   git push origin feature/your-new-feature
   

9. Submit a Pull Request: Create a PR against the main branch of the original repository. Clearly describe your changes.
10. Review: Wait for code review and address any feedback.

Contributor Graph

---

License

This project is protected under the MIT License. For more details, refer to the LICENSE file.

---

Acknowledgments

• Inspiration drawn from the need to streamline CRUD operations across multiple projects utilizing SQLModel.
• Built upon the excellent foundations provided by SQLModel and SQLAlchemy.
• Utilizes Loguru for optional logging and Factory Boy for test data generation.
• Special thanks to all contributors and users who provide feedback and improvements.

---