sqlmodel-object-helpers 0.0.3

Generic async query helpers for SQLModel: filtering, eager loading, pagination

sqlmodel-object-helpers

Generic async query helpers for SQLModel: filtering, eager loading, pagination, and mutations with security limits and full type safety.

Features

• Async-First - All query and mutation functions are async, designed for AsyncSession
• Flexible Filtering - LogicalFilter with recursive AND/OR/condition trees, plus flat dict filters with dot-notation for relationship traversal
• 15 Operators - eq, ne, gt, lt, ge, le, in_, not_in, like, ilike, between, is, isnot, match, exists
• Smart Eager Loading - Automatic selectinload for one-to-many and joinedload for many-to-one
• Pagination - Page/per_page with total count and configurable max_per_page limit
• Projections - Select specific columns across joins with dot-notation and SQL aliases
• CRUD Mutations - add_object, update_object, delete_object plus bulk update_objects, delete_objects with flush-only semantics for composable transactions
• Count & Exists - count_objects (single SELECT count(*)) and exists_object (SELECT EXISTS(...)) without loading data
• Row Locking - for_update parameter on get_object for SELECT ... FOR UPDATE
• Time Filtering - TimeFilter for created_at/updated_at range filtering with half-open interval semantics
• Relationship Safety Check - check_for_related_records pre-deletion inspection of ONETOMANY dependencies
• Security Limits - Configurable depth, list size, and pagination caps to prevent abuse
• Type Safety - Full type annotations with PEP 695 generics and py.typed marker (PEP 561)

Installation

pip install sqlmodel-object-helpers

One import, everything through dot notation:

import sqlmodel_object_helpers as soh

soh.get_object(session, ...)
soh.add_object(session, ...)

Quick Start

from datetime import datetime

from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import SQLModel, Field

import sqlmodel_object_helpers as soh


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    is_active: bool = True


async def example(session: AsyncSession):
    # Create
    user = await soh.add_object(session, User(name="Alice"))

    # Get one by PK
    found = await soh.get_object(session, User, pk={"id": user.id})

    # Get many with filtering
    active_users = await soh.get_objects(
        session, User,
        filters={"is_active": {soh.Operator.EQ: True}},
    )

    # Paginated
    page = await soh.get_objects(
        session, User,
        pagination=soh.Pagination(page=1, per_page=25),
    )
    # page.data -> list[User], page.pagination.total -> int

Configuration

Override security limits at application startup:

import sqlmodel_object_helpers as soh

soh.settings.max_per_page = 300
soh.settings.max_filter_depth = 50

Setting	Default	Description
max_filter_depth	100	Maximum recursion depth for AND/OR nesting
max_and_or_items	100	Maximum sub-filters in a single AND/OR list
max_load_depth	10	Maximum depth of eager-loading chains
max_in_list_size	1000	Maximum elements in an IN(...) list
max_per_page	500	Maximum value for per_page in pagination

Query Operations

get_object

Fetch a single object by primary key or filters. Supports ORM mode (model instances with eager loading) and SQL mode (flat dicts from specific columns).

import sqlmodel_object_helpers as soh

# By primary key
user = await soh.get_object(session, User, pk={"id": 5})

# By LogicalFilter
user = await soh.get_object(
    session, User,
    filters=soh.LogicalFilter(condition={"name": {soh.Operator.EQ: "Alice"}}),
)

# With eager loading
user = await soh.get_object(
    session, User,
    pk={"id": 5},
    load_paths=["posts", "posts.comments"],
)

# Graceful mode (returns None instead of raising)
user = await soh.get_object(session, User, pk={"id": 999}, suspend_error=True)

# Row-level locking (SELECT ... FOR UPDATE)
user = await soh.get_object(session, User, pk={"id": 5}, for_update=True)

get_objects

Fetch multiple objects with filtering, pagination, sorting, and eager loading.

import sqlmodel_object_helpers as soh

# Simple filter
users = await soh.get_objects(session, User, filters={"is_active": {soh.Operator.EQ: True}})

# With pagination + sorting
result = await soh.get_objects(
    session, User,
    pagination=soh.Pagination(page=1, per_page=25),
    order_by=soh.OrderBy(sorts=[soh.OrderAsc(asc="name")]),
)
# result.data -> list[User]
# result.pagination -> PaginationR(page=1, per_page=25, total=100)

# OR logic between conditions
users = await soh.get_objects(
    session, User,
    filters={"is_active": {soh.Operator.EQ: True}, "role": {soh.Operator.EQ: "admin"}},
    logical_operator="OR",
)

# Relationship filters (dot-notation)
users = await soh.get_objects(
    session, Attempt,
    filters={"application.applicant.last_name": {soh.Operator.EQ: "Smith"}},
    load_paths=["application.applicant"],
)

# Time filtering
from datetime import datetime

recent = await soh.get_objects(
    session, User,
    time_filter=soh.TimeFilter(
        created_after=datetime(2026, 1, 1),
        created_before=datetime(2026, 2, 1),
    ),
)

count_objects

Return the count of records matching filters without loading any data.

import sqlmodel_object_helpers as soh

# Total count
total = await soh.count_objects(session, User)

# Filtered count
active = await soh.count_objects(session, User, filters={"is_active": {soh.Operator.EQ: True}})

# With time filter
recent = await soh.count_objects(
    session, User,
    time_filter=soh.TimeFilter(created_after=datetime(2026, 1, 1)),
)

exists_object

Check whether at least one record matches the criteria. Uses EXISTS (SELECT ... LIMIT 1) -- the database stops at the first match.

import sqlmodel_object_helpers as soh

# By PK
found = await soh.exists_object(session, User, pk={"id": 5})

# By filter
has_admin = await soh.exists_object(session, User, filters={"role": {soh.Operator.EQ: "admin"}})

get_projection

Fetch specific columns from related tables. Returns flat dicts instead of ORM instances.

import sqlmodel_object_helpers as soh

rows = await soh.get_projection(
    session,
    Attempt,
    columns=[
        "id",
        ("application.applicant.last_name", "applicant_name"),
        ("schedule.unit.address", "unit_address"),
    ],
    outer_joins=["schedule"],
    limit=100,
)
# [{"id": 1, "applicant_name": "Smith", "unit_address": "123 Main St"}, ...]

Mutations

All mutation functions use session.flush() instead of session.commit() -- the caller manages transaction boundaries. This allows composing multiple mutations into a single atomic transaction:

import sqlmodel_object_helpers as soh

async with session.begin():
    await soh.add_object(session, billing)
    await soh.add_object(session, payment)
    # commit happens automatically when the block exits

add_object / add_objects

import sqlmodel_object_helpers as soh

# Single
user = await soh.add_object(session, User(name="Alice"))
# user.id is now populated (server-generated)

# Bulk
users = await soh.add_objects(session, [User(name="Alice"), User(name="Bob")])

update_object

import sqlmodel_object_helpers as soh

user = await soh.get_object(session, User, pk={"id": 1})
updated = await soh.update_object(session, user, {"name": "Alice Updated", "is_active": False})

Field names are validated against the model before touching the database. Raises MutationError if a key does not exist on the model.

delete_object

import sqlmodel_object_helpers as soh

# By instance
await soh.delete_object(session, User, instance=user)

# By PK
await soh.delete_object(session, User, pk={"id": 5})

update_objects / delete_objects

Bulk operations that issue a single SQL statement without loading objects:

import sqlmodel_object_helpers as soh

# Bulk update: UPDATE users SET is_active=False WHERE role='guest'
count = await soh.update_objects(
    session, User,
    data={"is_active": False},
    filters={"role": {soh.Operator.EQ: "guest"}},
)
# count -> number of rows updated

# Bulk delete: DELETE FROM users WHERE is_active=False
count = await soh.delete_objects(
    session, User,
    filters={"is_active": {soh.Operator.EQ: False}},
)
# count -> number of rows deleted

Filters are required for safety -- empty filters are rejected. Dot-notation (relationship) filters are not supported in bulk operations.

check_for_related_records

Pre-deletion check that inspects all ONETOMANY relationships:

import sqlmodel_object_helpers as soh

deps = await soh.check_for_related_records(session, Organization, pk={"id": 1})
if deps:
    print(deps)
    # ["Related record found in 'Unit (units_lkp)' (id=1)", ...]

Returns None if no related records exist, or a list of human-readable dependency descriptions.

Filtering

LogicalFilter (AND/OR/condition)

Recursive filter structure used by get_object. Exactly one of AND, OR, or condition must be set:

import sqlmodel_object_helpers as soh

# Simple condition
f = soh.LogicalFilter(condition={"status_id": {soh.Operator.EQ: 10}})

# OR
f = soh.LogicalFilter(OR=[
    soh.LogicalFilter(condition={"status_id": {soh.Operator.EQ: 10}}),
    soh.LogicalFilter(condition={"is_blocked": {soh.Operator.EQ: True}}),
])

# Nested AND + OR
f = soh.LogicalFilter(AND=[
    soh.LogicalFilter(condition={"is_active": {soh.Operator.EQ: True}}),
    soh.LogicalFilter(OR=[
        soh.LogicalFilter(condition={"role": {soh.Operator.EQ: "admin"}}),
        soh.LogicalFilter(condition={"role": {soh.Operator.EQ: "manager"}}),
    ]),
])

Flat dict filters

Used by get_objects. Nested dicts are auto-flattened via flatten_filters:

# Nested form (auto-flattened)
{"application": {"applicant": {"last_name": {soh.Operator.EQ: "test"}}}}

# Equivalent flat form (dot-notation)
{"application.applicant.last_name": {soh.Operator.EQ: "test"}}

Operators

Enum member	SQL	Example
Operator.EQ	=	{soh.Operator.EQ: 10}
Operator.NE	!=	{soh.Operator.NE: 0}
Operator.GT	>	{soh.Operator.GT: 5}
Operator.LT	<	{soh.Operator.LT: 100}
Operator.GE	>=	{soh.Operator.GE: 1}
Operator.LE	<=	{soh.Operator.LE: 50}
Operator.IN	IN (...)	{soh.Operator.IN: [1, 2, 3]}
Operator.NOT_IN	NOT IN (...)	{soh.Operator.NOT_IN: [4, 5]}
Operator.LIKE	LIKE	{soh.Operator.LIKE: "%test%"}
Operator.ILIKE	ILIKE	{soh.Operator.ILIKE: "%test%"}
Operator.BETWEEN	BETWEEN	{soh.Operator.BETWEEN: [1, 10]}
Operator.IS	IS	{soh.Operator.IS: None}
Operator.IS_NOT	IS NOT	{soh.Operator.IS_NOT: None}
Operator.MATCH	MATCH	{soh.Operator.MATCH: "query"}
"exists"	IS NOT NULL / .any()	{"exists": True}

Operator is a StrEnum — each member equals its string value (Operator.EQ == "eq"), so string keys still work but enum members provide type safety and autocompletion.

Multiple operators can be combined on a single field: {"age": {soh.Operator.GE: 18, soh.Operator.LE: 65}}.

Typed Filter Models

Pydantic models for type-safe filter schemas in API endpoints:

import sqlmodel_object_helpers as soh
from pydantic import BaseModel

class UserFilter(BaseModel):
    age: soh.FilterInt | None = None       # eq, ne, gt, lt, ge, le, in_
    name: soh.FilterStr | None = None      # eq, ne, like, ilike
    is_active: soh.FilterBool | None = None  # eq, ne
    posts: soh.FilterExists | None = None  # exists: bool

Available: FilterInt, FilterStr, FilterDate, FilterDatetime, FilterTimedelta, FilterBool, FilterExists.

Eager Loading

The library automatically selects the optimal loading strategy:

• selectinload for one-to-many (uselist=True) -- avoids cartesian products
• joinedload for many-to-one (uselist=False) -- single-query efficiency

Dot-notation chaining resolves each level independently:

# Each segment gets the optimal strategy
load_paths=["application.applicant"]
# application -> joinedload (many-to-one)
# applicant   -> joinedload (many-to-one)

load_paths=["comments"]
# comments -> selectinload (one-to-many)

Maximum chain depth is controlled by settings.max_load_depth (default: 10).

Pagination & Sorting

Pagination

import sqlmodel_object_helpers as soh

result = await soh.get_objects(
    session, User,
    pagination=soh.Pagination(page=2, per_page=25),
)
result.data         # list[User]
result.pagination   # PaginationR(page=2, per_page=25, total=150)

per_page is validated against settings.max_per_page (default: 500).

Sorting

import sqlmodel_object_helpers as soh

order = soh.OrderBy(sorts=[
    soh.OrderAsc(asc="last_name"),
    soh.OrderDesc(desc="created_at"),
])

result = await soh.get_objects(session, User, order_by=order)

API Reference

Settings

• soh.settings -- Module-level QueryHelperSettings instance (mutable at runtime)
• soh.QueryHelperSettings -- Pydantic model for security/performance limits

Query Functions

• soh.get_object(session, model, ...) -- Single object by PK or filters (supports for_update row locking)
• soh.get_objects(session, model, ...) -- Multiple objects with filtering, pagination, sorting, time_filter
• soh.count_objects(session, model, ...) -- Count matching records (SELECT count(*))
• soh.exists_object(session, model, ...) -- Check existence (SELECT EXISTS(...))
• soh.get_projection(session, model, columns, ...) -- Column projection across joins

Mutation Functions

• soh.add_object(session, instance) -- Add single, flush + refresh
• soh.add_objects(session, instances) -- Add multiple, flush + refresh each
• soh.update_object(session, instance, data) -- Update fields from dict
• soh.update_objects(session, model, data, filters) -- Bulk update via single UPDATE ... WHERE
• soh.delete_object(session, model, *, instance, pk) -- Delete by reference or PK
• soh.delete_objects(session, model, filters) -- Bulk delete via single DELETE ... WHERE
• soh.check_for_related_records(session, model, pk) -- Pre-deletion dependency check

Filter Builders

• soh.build_filter(model, filters, ...) -- Build SQLAlchemy expression from LogicalFilter dict
• soh.build_flat_filter(model, filters, ...) -- Build from flat dot-notation dict (aliased joins)
• soh.flatten_filters(filters) -- Convert nested dicts to flat dot-notation

Operators

• soh.Operator -- StrEnum of operator names
• soh.SUPPORTED_OPERATORS -- Dict mapping operator names to SQLAlchemy lambdas
• soh.SPECIAL_OPERATORS -- Frozenset of operators with extended handling ({"exists"})

Loaders

• soh.build_load_chain(model, path, ...) -- Build loader option from string/list path
• soh.build_load_options(model, attrs) -- Build single loader option chain

Exceptions

• soh.QueryError -- Base exception (has status_code attribute for HTTP mapping)
• soh.ObjectNotFoundError -- 404 Not Found
• soh.InvalidFilterError -- 400 Bad Request
• soh.InvalidLoadPathError -- 400 Bad Request
• soh.DatabaseError -- 500 Internal Server Error
• soh.MutationError -- 400 Bad Request

Filter Types

• soh.FilterInt, soh.FilterStr, soh.FilterDate, soh.FilterDatetime, soh.FilterTimedelta, soh.FilterBool, soh.FilterExists
• soh.OrderAsc, soh.OrderDesc, soh.OrderBy
• soh.LogicalFilter
• soh.TimeFilter -- created_after, created_before, updated_after, updated_before with half-open interval [after, before)

Pagination Types

• soh.Pagination -- Request model (page, per_page)
• soh.PaginationR -- Response model (page, per_page, total)
• soh.GetAllPagination[T] -- Generic wrapper (data: list[T], pagination: PaginationR | None)

Projection Types

• soh.ColumnSpec -- Type alias: str | tuple[str, str]

Development

pip install -e ".[dev]"
pytest tests/

Syncing with Remote

# Fetch commits and tags
git pull origin main --tags

# If local branch is behind remote
git reset --hard origin/main

Verify sync status

git log --oneline origin/main -5
git diff origin/main

CI/CD

The pipeline consists of two stages:

1. auto_tag — on push to main, reads __version__ from __init__.py and automatically creates a tag (if it doesn't exist)

2. mirror_to_github — on tag creation, mirrors the repository to GitHub (removing .gitlab-ci.yml)

Release flow

1. Update __version__ in src/sqlmodel_object_helpers/__init__.py
2. Push to main
3. CI creates tag → mirrors to GitHub → publishes to PyPI

Versioning

This project follows Semantic Versioning (MAJOR.MINOR.PATCH):

• PATCH — bug fixes, documentation, metadata
• MINOR — new features (backwards compatible)
• MAJOR — breaking changes

License

This project is licensed under the PolyForm Noncommercial License 1.0.0.

You may use this software for noncommercial purposes only.

See LICENSE for the full license text, or visit polyformproject.org.