sqlmodel-object-helpers 0.0.1
Generic async query helpers for SQLModel: filtering, eager loading, pagination
Requires Python: >=3.14
Keywords: async, filter, pagination, query, sqlalchemy, sqlmodel

# sqlmodel-object-helpers

[![PyPI version](https://badge.fury.io/py/sqlmodel-object-helpers.svg)](https://pypi.org/project/sqlmodel-object-helpers/)
[![Python 3.14+](https://img.shields.io/badge/python-3.14+-blue.svg)](https://www.python.org/downloads/)
[![License: PolyForm Noncommercial](https://img.shields.io/badge/License-PolyForm%20Noncommercial-blue.svg)](https://polyformproject.org/licenses/noncommercial/1.0.0/)

Generic async query helpers for [SQLModel](https://sqlmodel.tiangolo.com/): 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` with flush-only semantics for composable transactions
- **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

```bash
pip install sqlmodel-object-helpers
```

**Note:** The import name is `sqlmodel_object_helpers`:

```python
from sqlmodel_object_helpers import get_objects, get_object, add_object
```

## Quick Start

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

from sqlmodel_object_helpers import (
    get_object, get_objects, add_object, Pagination,
)


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 add_object(session, User(name="Alice"))

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

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

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

## Configuration

Override security limits at application startup:

```python
from sqlmodel_object_helpers import settings

settings.max_per_page = 300
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).

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

# By LogicalFilter
from sqlmodel_object_helpers import LogicalFilter

user = await get_object(
    session, User,
    filters=LogicalFilter(condition={"name": {"eq": "Alice"}}),
)

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

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

### get_objects

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

```python
from sqlmodel_object_helpers import get_objects, Pagination, OrderBy, OrderAsc

# Simple filter
users = await get_objects(session, User, filters={"is_active": {"eq": True}})

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

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

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

### get_projection

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

```python
from sqlmodel_object_helpers import get_projection

rows = await 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:

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

### add_object / add_objects

```python
from sqlmodel_object_helpers import add_object, add_objects

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

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

### update_object

```python
from sqlmodel_object_helpers import update_object

user = await get_object(session, User, pk={"id": 1})
updated = await 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

```python
from sqlmodel_object_helpers import delete_object

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

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

### check_for_related_records

Pre-deletion check that inspects all ONETOMANY relationships:

```python
from sqlmodel_object_helpers import check_for_related_records

deps = await 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:

```python
from sqlmodel_object_helpers import LogicalFilter

# Simple condition
f = LogicalFilter(condition={"status_id": {"eq": 10}})

# OR
f = LogicalFilter(OR=[
    LogicalFilter(condition={"status_id": {"eq": 10}}),
    LogicalFilter(condition={"is_blocked": {"eq": True}}),
])

# Nested AND + OR
f = LogicalFilter(AND=[
    LogicalFilter(condition={"is_active": {"eq": True}}),
    LogicalFilter(OR=[
        LogicalFilter(condition={"role": {"eq": "admin"}}),
        LogicalFilter(condition={"role": {"eq": "manager"}}),
    ]),
])
```

### Flat dict filters

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

```python
# Nested form (auto-flattened)
{"application": {"applicant": {"last_name": {"eq": "test"}}}}

# Equivalent flat form (dot-notation)
{"application.applicant.last_name": {"eq": "test"}}
```

### Operators

| Operator | SQL | Example |
|----------|-----|---------|
| `eq` | `=` | `{"eq": 10}` |
| `ne` | `!=` | `{"ne": 0}` |
| `gt` | `>` | `{"gt": 5}` |
| `lt` | `<` | `{"lt": 100}` |
| `ge` | `>=` | `{"ge": 1}` |
| `le` | `<=` | `{"le": 50}` |
| `in_` | `IN (...)` | `{"in_": [1, 2, 3]}` |
| `not_in` | `NOT IN (...)` | `{"not_in": [4, 5]}` |
| `like` | `LIKE` | `{"like": "%test%"}` |
| `ilike` | `ILIKE` | `{"ilike": "%test%"}` |
| `between` | `BETWEEN` | `{"between": [1, 10]}` |
| `is` | `IS` | `{"is": null}` |
| `isnot` | `IS NOT` | `{"isnot": null}` |
| `match` | `MATCH` | `{"match": "query"}` |
| `exists` | `IS NOT NULL` / `.any()` | `{"exists": true}` |

Multiple operators can be combined on a single field: `{"age": {"ge": 18, "le": 65}}`.

### Typed Filter Models

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

```python
from sqlmodel_object_helpers import FilterInt, FilterStr, FilterBool, FilterExists

class UserFilter(BaseModel):
    age: FilterInt | None = None       # eq, ne, gt, lt, ge, le, in_
    name: FilterStr | None = None      # eq, ne, like, ilike
    is_active: FilterBool | None = None  # eq, ne
    posts: 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:

```python
# 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

```python
from sqlmodel_object_helpers import get_objects, Pagination

result = await get_objects(
    session, User,
    pagination=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

```python
from sqlmodel_object_helpers import OrderBy, OrderAsc, OrderDesc

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

result = await get_objects(session, User, order_by=order)
```

## API Reference

### Settings

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

### Query Functions

- `get_object(session, model, ...)` -- Single object by PK or filters
- `get_objects(session, model, ...)` -- Multiple objects with filtering, pagination, sorting
- `get_projection(session, model, columns, ...)` -- Column projection across joins

### Mutation Functions

- `add_object(session, instance)` -- Add single, flush + refresh
- `add_objects(session, instances)` -- Add multiple, flush + refresh each
- `update_object(session, instance, data)` -- Update fields from dict
- `delete_object(session, model, *, instance, pk)` -- Delete by reference or PK
- `check_for_related_records(session, model, pk)` -- Pre-deletion dependency check

### Filter Builders

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

### Operators

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

### Loaders

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

### Exceptions

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

### Filter Types

- `FilterInt`, `FilterStr`, `FilterDate`, `FilterDatetime`, `FilterTimedelta`, `FilterBool`, `FilterExists`
- `OrderAsc`, `OrderDesc`, `OrderBy`
- `LogicalFilter`

### Pagination Types

- `Pagination` -- Request model (page, per\_page)
- `PaginationR` -- Response model (page, per\_page, total)
- `GetAllPagination[T]` -- Generic wrapper (data: list[T], pagination: PaginationR | None)

### Projection Types

- `ColumnSpec` -- Type alias: `str | tuple[str, str]`

## Development

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

### Syncing with Remote

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

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

### Verify sync status

```bash
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](https://semver.org/) (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](LICENSE) for the full license text, or visit [polyformproject.org](https://polyformproject.org/licenses/noncommercial/1.0.0/).