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
• Idempotent Bulk Insert - insert_objects_ignore_conflicts: one INSERT ... ON CONFLICT DO NOTHING per chunk (PostgreSQL, SQLite), safe to replay
• Bulk Upsert - upsert_objects: one INSERT ... ON CONFLICT DO UPDATE per chunk (PostgreSQL, SQLite); update expressions run on every call
• 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)
• Dynamic Table Metadata - build_dynamic_meta derives TableMeta + list[ColumnMeta] from a SQLModel class by reading PostgreSQL pg_description with fallback to Column(comment=...). Supports per-role label/row_link overrides via || comment format, TTL-cached
• Standalone Mode - configure() + import sqlmodel_object_helpers.standalone for auto-session usage without DI
• Session Lifecycle Logging - Transparent session open/commit/rollback logging with hex session IDs and timing

Installation

pip install sqlmodel-object-helpers

Two usage modes:

# DI mode — caller provides session (FastAPI Depends, etc.)
import sqlmodel_object_helpers as soh

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

# Standalone mode — auto-creates session per call
import sqlmodel_object_helpers.standalone as soh_sa

soh_sa.get_object(User, pk={"id": 1})
soh_sa.add_object(User(name="Alice"))

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

Standalone Mode

For projects that don't use FastAPI DI or need simple one-call-one-transaction semantics.

Important: The session factory must use expire_on_commit=False. After each standalone call the session commits and closes — with the default True, all attributes on returned objects would be expired and inaccessible (DetachedInstanceError).

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
import sqlmodel_object_helpers as soh
import sqlmodel_object_helpers.standalone as soh_sa

engine = create_async_engine("postgresql+asyncpg://...")
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)

# 1. Configure the session factory once at startup
soh.configure(async_session_factory)

# 2. Use standalone wrappers — each call creates its own session and commits
user = await soh_sa.get_object(User, pk={"id": 1})
new_user = await soh_sa.add_object(User(name="Alice"))
await soh_sa.delete_object(User, pk={"id": 5})

# All 12 functions are available:
# Queries:  get_object, get_objects, count_objects, exists_object, get_projection
# Mutations: add_object, add_objects, update_object, update_objects,
#            delete_object, delete_objects, check_for_related_records

Each standalone call creates a session, executes the operation, commits on success, and rolls back on error. No session parameter needed.

auto_session - multi-operation transactions

When you need multiple operations in a single atomic transaction:

import sqlmodel_object_helpers as soh

async with soh.auto_session() as session:
    billing = await soh.add_object(session, Billing(...))
    payment = await soh.add_object(session, Payment(...))
    # commit happens automatically on exit
    # if any operation fails — ALL are rolled back

Session Management

DI mode — create_session_dependency()

For FastAPI projects with dependency injection:

import sqlmodel_object_helpers as soh
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

get_session = soh.create_session_dependency(async_session_factory)
DbSession = Annotated[AsyncSession, Depends(get_session)]

@router.get("/users/{user_id}")
async def get_user(user_id: int, session: DbSession):
    return await soh.get_object(session, User, pk={"id": user_id})

One session per HTTP request. The dependency commits on success, rolls back on error.

Standalone mode — configure()

For projects without DI or for scripts/CLI:

import sqlmodel_object_helpers as soh

soh.configure(async_session_factory)
# Now standalone functions and auto_session() are available

Session Lifecycle Logging

All session operations are logged via logging.getLogger("sqlmodel_object_helpers"):

DEBUG  auto_session[1a2b3c4d] opened
DEBUG  auto_session[1a2b3c4d] committed (0.015s)
WARNING auto_session[1a2b3c4d] rollback (0.003s) — MutationError: ...

• Hex session ID (1a2b3c4d) correlates all log lines for the same session
• Prefix distinguishes mode: auto_session (standalone/auto_session) vs di_session (DI dependency)
• Timing shows elapsed time from session open to commit/rollback
• Level: DEBUG for normal flow, WARNING for rollbacks and commit failures

Enable with:

import logging
logging.getLogger("sqlmodel_object_helpers").setLevel(logging.DEBUG)

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

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=soh.LogicalOperator.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, timezone

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

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, tzinfo=timezone.utc)),
)

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")])

Choose the insert helper by its contract:

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.

insert_objects_ignore_conflicts

Idempotent bulk insert -- INSERT ... ON CONFLICT (...) DO NOTHING, no objects loaded or refreshed:

import sqlmodel_object_helpers as soh

inserted = await soh.insert_objects_ignore_conflicts(
    session, PaymentLog,
    rows=[{"source_id": "run1:1", "base": "a-cso"}, {"source_id": "run1:2", "base": "a-cso"}],
    conflict_columns=["source_id"],
)
# inserted -> rows actually written; duplicates by source_id are skipped and not counted

Replaying the same batch (or two concurrent identical batches) is safe, unlike add_objects, where one duplicate fails the whole batch. All rows must have the same keys. Rows go to the database as given (no Pydantic validation or coercion -- pass already-typed values); omitted columns get their column defaults: Field(default=...) / default_factory of a plain field and server_default (SQLModel does not copy Field(default) into an explicit sa_column). Large inputs are split using the actual compiled bind-parameter count, including omitted Python defaults and SQL expressions. One row exceeding the parameter budget raises MutationError. PostgreSQL and SQLite only (MutationError on other dialects).

The caller owns the transaction. If a later chunk fails, roll back the transaction to discard earlier chunks too. The helper does not commit or perform its own rollback. Standalone wrappers manage commit/rollback automatically.

upsert_objects

Bulk upsert -- INSERT ... ON CONFLICT (...) DO UPDATE SET ...: new rows are inserted, rows hitting the unique key are updated:

import sqlmodel_object_helpers as soh
from sqlalchemy import func

written = await soh.upsert_objects(
    session, Ast,
    rows=[{"code": "a", "width": 10, "ordinal": 1}, {"code": "b", "width": 20, "ordinal": 2}],
    conflict_columns=["code"],
    update_columns=["width", "ordinal"],        # default: every row field except conflict_columns
    set_extra={"updated_at": func.now()},       # applied only on update
)
# written -> distinct table rows inserted or updated, counted by primary key from RETURNING

If a key repeats inside rows, the last row wins and the row is counted once. Repeats are decided by the database for the column type (NUMERIC rounding, citext, NULLS NOT DISTINCT), not by Python equality: distinct values such as two instants of a DST fold stay distinct, and key values do not have to be hashable (ARRAY). On PostgreSQL a chunk of several rows runs inside a SAVEPOINT (session.begin_nested() flushes pending session changes first); a chunk the database rejects for a repeated key is retried row by row, which is slower. SQLite applies repeated keys in order within one statement. Every conflict_columns field must be present in each row. update_columns must be a subset of the row fields outside conflict_columns; set_extra takes model fields (value or SQL expression) that are in neither. When nothing is left to update, use insert_objects_ignore_conflicts instead (MutationError). Column.onupdate does not fire on conflict -- pass it in set_extra. Objects already loaded in the session are not refreshed. A conditional update (WHERE) or RETURNING is not covered -- build the statement with SQLAlchemy for that. Other rules as in insert_objects_ignore_conflicts: same keys in all rows, no Pydantic validation, actual parameter-based chunking, caller-managed transaction, PostgreSQL and SQLite only.

Repeating an upsert does not create duplicate keys, but it repeats the update: expressions such as hits + 1 and func.now() run again. NULL conflict-key values follow the index: an ordinary UNIQUE keeps them distinct, NULLS NOT DISTINCT treats them as one key.

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

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.

logical_operator of get_objects, count_objects, update_objects and delete_objects takes soh.LogicalOperator.AND (default) or soh.LogicalOperator.OR; the strings "AND"/"OR" in any case still work. Any other value raises InvalidFilterError, even with suspend_error=True.

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, FilterNaiveDatetime, FilterTimedelta, FilterBool, FilterExists.

Range Filters

FilterDatetimeRange and FilterNaiveDatetimeRange parse a comma-separated date string into gt/lt operators:

import sqlmodel_object_helpers as soh

# Full range: "FROM,TO" → gt + lt
f = soh.FilterDatetimeRange.model_validate("2026-04-01,2026-05-06")
f.model_dump(exclude_none=True)
# {"gt": datetime(2026, 4, 1, tzinfo=UTC), "lt": datetime(2026, 5, 6, tzinfo=UTC)}
# SQL: WHERE field > '2026-04-01' AND field < '2026-05-06'

# Open end: "FROM," → gt only
f = soh.FilterDatetimeRange.model_validate("2026-04-01,")
# SQL: WHERE field > '2026-04-01'

# Open start: ",TO" → lt only
f = soh.FilterDatetimeRange.model_validate(",2026-05-06")
# SQL: WHERE field < '2026-05-06'

Each date part accepts ISO (2026-04-01, 2026-04-01T00:00:00Z) and display format (01.04.2026 00:00).

• FilterDatetimeRange — produces UTC-aware datetimes
• FilterNaiveDatetimeRange — produces naive (timezone-unaware) datetimes

Datetime columns and SQLModel ≥0.0.45

Since SQLModel 0.0.45 a bare datetime field (no sa_type / sa_column) becomes UTCDateTime — TIMESTAMP WITH TIME ZONE that rejects naive values and returns UTC-aware ones. soh works on both sides of this change:

• Filter such columns with FilterDatetime, FilterDatetimeRange, TimeFilter or aware datetime values. A naive value is rejected by SQLModel and surfaces as soh.DatabaseError.
• FilterNaiveDatetime / FilterNaiveDatetimeRange are for columns declared explicitly as naive: sa_type=DateTime or sa_column=Column(DateTime()).
• build_dynamic_meta reports UTCDateTime columns as datetime.
• SQLModel does not migrate existing database columns. Over an existing timestamp column a bare datetime field reads values as UTC and, when the PostgreSQL session TimeZone is not UTC, writes them shifted by its offset. To keep the current behaviour of a bare datetime field, declare its database type explicitly: sa_column=Column(DateTime()) for timestamp, sa_column=Column(DateTime(timezone=True)) for timestamptz.

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)

Dynamic Table Metadata

build_dynamic_meta builds TableMeta + list[ColumnMeta] from a SQLModel class at runtime, reading PostgreSQL pg_description (TTL-cached) with fallback to Column(comment=...) model defaults.

import sqlmodel_object_helpers as soh

# Physical columns — label, type, lookup derived automatically
table_meta, columns = await soh.build_dynamic_meta(
    session,
    EmailMessage,
    name="email_messages",
    columns=[
        "id",                        # physical column — everything derived
        "email_type_id",             # FK on *_lkp → lookup_dict auto-derived
        "updated_at",
        soh.ColumnMeta(              # virtual column — fully specified
            json_path="last_editor.user_name",
            label="Редактор",
            type=soh.ColumnType.STRING,
        ),
    ],
    role_type="operator",            # per-role label/row_link overrides
)

# Returns (TableMeta, list[ColumnMeta]) ready for GetAllPagination
return soh.GetAllPagination(
    table=table_meta,
    columns=columns,
    data=items,
    pagination=pagination_r,
)

Type derivation (SA type → ColumnType)

Physical column types are mapped automatically. Unknown types fall back to string.

Interval is a TypeDecorator over DateTime, but its own entry wins, so it stays string.

Label precedence (per physical column)

1. pg_description — DBA edit via COMMENT ON COLUMN (supports per-role || overrides)
2. Column(comment=...) — Python-side default in the model
3. ValueError — fail-loud if both are missing

Per-role overrides via || format

DBAs can set role-specific labels and row links in PostgreSQL comments:

COMMENT ON COLUMN emails.email_type_id IS 'Тип письма||operator=Категория||buh=Реквизит';
COMMENT ON TABLE  emails               IS 'Письма||row_link=/email/$id||row_link.operator=/op/email/$id';

TTL cache

pg_description queries are cached per (schema, table) with a configurable TTL (default 60s):

soh.configure_meta_cache_ttl(120)       # change TTL to 120 seconds
soh.invalidate_meta_cache()             # clear entire cache
soh.invalidate_meta_cache(schema="lead")  # clear all entries for a schema
soh.invalidate_meta_cache("lead", "emails")  # clear one entry

Standalone mode

import sqlmodel_object_helpers.standalone as soh_sa

table_meta, columns = await soh_sa.build_dynamic_meta(
    EmailMessage,
    name="email_messages",
    columns=["id", "email_type_id"],
)

Backward compatibility and migration

GetAllPagination is fully backward compatible — table and columns default to None, so existing endpoints continue to work without changes:

# Before (still works as-is)
return soh.GetAllPagination(data=items, pagination=pagination_r)

# After (meta added when ready)
return soh.GetAllPagination(
    table=table_meta,
    columns=columns,
    data=items,
    pagination=pagination_r,
)

Endpoints can be migrated one at a time. Three strategies per endpoint:

API Reference

Session Management

• soh.configure(factory) -- Register async_sessionmaker for standalone mode
• soh.auto_session() -- Async context manager: creates session, commits on success, rolls back on error
• soh.create_session_dependency(factory) -- Create async generator for FastAPI Depends

Standalone Wrappers

• import sqlmodel_object_helpers.standalone as soh_sa -- All 12 query/mutation functions plus build_dynamic_meta without session parameter

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.insert_objects_ignore_conflicts(session, model, rows, *, conflict_columns) -- Bulk insert via INSERT ... ON CONFLICT DO NOTHING, returns inserted count
• soh.upsert_objects(session, model, rows, *, conflict_columns, update_columns=None, set_extra=None) -- Bulk insert via INSERT ... ON CONFLICT DO UPDATE, returns inserted + updated count
• soh.check_for_related_records(session, model, pk) -- Pre-deletion dependency check

Dynamic Meta

• soh.build_dynamic_meta(session, model, *, name, columns, header, row_link, role_type) -- Build TableMeta + list[ColumnMeta] from model + pg_description
• soh.load_pg_comments(session, schema, table) -- Read (table_comment, {col: comment}) from pg_description (TTL-cached)
• soh.configure_meta_cache_ttl(seconds) -- Override default pg_description cache TTL (default: 60s)
• soh.invalidate_meta_cache(schema, table) -- Manually invalidate the pg_description cache (full, by-schema, or by-table)
• soh.ColumnEntry -- Type alias: str | ColumnMeta (element of columns list in build_dynamic_meta)

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.LogicalOperator -- StrEnum AND/OR for the logical_operator argument
• 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, *, suspend_error=False) -- 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.FilterNaiveDatetime, soh.FilterTimedelta, soh.FilterBool, soh.FilterExists
• soh.FilterDatetimeRange, soh.FilterNaiveDatetimeRange -- Range filters that parse "FROM,TO" strings into gt/lt operators
• soh.OrderAsc, soh.OrderDesc, soh.OrderBy
• soh.LogicalFilter
• soh.TimeFilter -- created_after, created_before, updated_after, updated_before with half-open interval [after, before)

Datetime Types

• soh.UTCDatetime -- Annotated[datetime, AfterValidator] that rejects naive datetimes and converts aware datetimes to UTC

Pagination Types

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

Table Metadata Types

• soh.TableMeta -- Table-level metadata (name, header, row_link)
• soh.ColumnMeta -- Column metadata (json_path, label, type, lookup_dict, lookup_path, bool_labels)
• soh.ColumnType -- StrEnum of column data types (string, integer, float, boolean, date, datetime)
• soh.BoolLabels -- Display labels for boolean columns (true_label, false_label)

Lookup Types

• soh.LookupMeta -- Lookup metadata (name used as cache key, matches ColumnMeta.lookup_dict)
• soh.LookupResponse[T] -- Generic wrapper for lookup endpoints (meta: LookupMeta, data: list[T])

Projection Types

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

Development

uv sync
uv run pytest tests/

PostgreSQL integration tests use the optional postgres dependency group (asyncpg). Each test creates a randomly named schema and removes that schema afterwards; it does not recreate application tables. Without SOH_TEST_POSTGRES_URL, these tests are skipped.

uv sync --group postgres
SOH_TEST_POSTGRES_URL=postgresql+asyncpg://user:password@localhost/test_db \
  uv run --group postgres pytest -q tests/test_conflict_mutations_postgres.py

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

Changelog

0.0.10 — unreleased

• insert_objects_ignore_conflicts and upsert_objects: bulk insert with explicit unique-key conflict handling, inserted/updated counts, and standalone wrappers.
• Chunk sizes account for actual SQL parameters, including omitted Python defaults, repeated positional parameters, expanding IN expressions, and update expressions. Tested with SQLAlchemy 2.0.46, 2.0.52 and 2.1.6.
• Upsert reports a missing conflict-key field as MutationError. Repeated update expressions are explicitly documented as non-idempotent.
• Upsert leaves repeated keys to the database instead of Python equality: no silent loss of distinct keys (DST fold), no DatabaseError for keys equal only after column type coercion (NUMERIC(10,2)), no TypeError for unhashable keys (ARRAY). The count is distinct rows by primary key.
• PostgreSQL tests run through the optional postgres dependency group; tested on PostgreSQL 18.4 and 18.1, the versions of the SRO Agent and ExamCRM dev databases.

0.0.9 — 2026-10-09

• Dependency lower bound is now sqlmodel>=0.0.27. SQLModel 0.0.22–0.0.26 cannot create models on Python 3.14 (Field 'id' requires a type annotation), so the previous >=0.0.22 bound was never satisfiable. Test suite passes on every SQLModel from 0.0.27 to 0.0.48.
• Removed the unused dev extra (mypy, ruff>=0.8.0) and [tool.mypy]: types are checked with pyright, dev tools live in [dependency-groups]. pip install sqlmodel-object-helpers[dev] no longer exists.

0.0.8 — 2026-10-09

• SQLModel ≥0.0.45 support — build_dynamic_meta maps an unmapped TypeDecorator by its impl, so bare datetime fields (UTCDateTime in SQLModel ≥0.0.45) stay datetime instead of falling back to string. Interval stays string. Tested on SQLModel 0.0.42 and 0.0.48. The sqlmodel>=0.0.22 lower bound is unchanged.
• Dependency is now sqlalchemy[asyncio] — the package imports sqlalchemy.ext.asyncio, and SQLAlchemy 2.1 (allowed by SQLModel 0.0.48) no longer installs greenlet by default.
• LogicalOperator — StrEnum for logical_operator. An unknown value now raises InvalidFilterError; before, anything except "AND" silently meant OR. Plain strings "AND"/"OR" keep working.
• Typing: get_object / get_objects overloads narrow the return type by columns, pagination and suspend_error; the package and tests are clean under the fleet pyright config.
• Test dependencies moved to [dependency-groups] dev — uv sync installs them.
• Docs: Datetime columns and SQLModel ≥0.0.45.

0.0.6

• build_dynamic_meta / load_pg_comments / configure_meta_cache_ttl / invalidate_meta_cache — dynamic UI metadata reader. Builds TableMeta and list[ColumnMeta] from a SQLModel class by reading PostgreSQL pg_description (TTL-cached, default 60s) with fallback to Column(comment=...) defaults from the model. Supports per-role label/row_link overrides via extended || comment format. For physical columns the label precedence is pg_description > Column(comment=...) model default. For virtual columns (paths with . traversing relationships) and full overrides — pass a fully-specified ColumnMeta instance directly in the columns list (used as-is, no derivation). Lookup lookup_dict/lookup_path are derived from FK on *_lkp tables by {schema}_{base} convention. Type derived from SA column type. DBAs can edit labels via COMMENT ON COLUMN/TABLE SQL — frontend reflects changes within cache TTL without redeploy. No sync block, no schema migrations introduced; the application's db.py is not touched.

0.0.5

• FilterDatetimeRange / FilterNaiveDatetimeRange — range filters that parse comma-separated date strings ("2026-04-01,2026-05-06") into gt/lt operators for SQL filtering
• ColumnMeta / TableMeta / ColumnType / BoolLabels — dynamic table metadata: backend describes columns, types, labels, lookups, and row navigation so the frontend renders any table without hardcoding
• GetAllPagination — now includes optional table and columns fields for delivering table metadata alongside paginated data
• LookupMeta / LookupResponse — standard {meta, data} wrapper for lookup (_lkp) endpoints, enabling unified frontend caching of dictionaries with meta.name as cache key

0.0.4

• Initial public release

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.