Metadata-Version: 2.4
Name: sqldataclass
Version: 0.1.8
Summary: Memory-efficient ORM bridge: pydantic dataclasses + SQLAlchemy Core. FastAPI compatible.
Project-URL: Homepage, https://github.com/lejansenGitHub/SQLDataclass
Project-URL: Repository, https://github.com/lejansenGitHub/SQLDataclass
Project-URL: Issues, https://github.com/lejansenGitHub/SQLDataclass/issues
Author-email: Lennart Jansen <lejansen.home@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: dataclass,fastapi,memory-efficient,orm,pydantic,sqlalchemy
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: FastAPI
Classifier: Framework :: Pydantic
Classifier: Framework :: Pydantic :: 2
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: pydantic>=2.0
Requires-Dist: sqlalchemy>=2.0
Provides-Extra: dev
Requires-Dist: blueprint-linters>=0.8.0; extra == 'dev'
Requires-Dist: diff-cover>=9.0; extra == 'dev'
Requires-Dist: mypy>=1.14; extra == 'dev'
Requires-Dist: pre-commit>=4.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest-cov>=6.0; extra == 'dev'
Requires-Dist: pytest-mock>=3.14; extra == 'dev'
Requires-Dist: pytest-timeout>=2.3; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.15; extra == 'dev'
Provides-Extra: postgres
Requires-Dist: psycopg[binary]>=3.0; extra == 'postgres'
Description-Content-Type: text/markdown

# SQLDataclass

**Memory-efficient ORM bridge: pydantic dataclasses + SQLAlchemy Core. FastAPI compatible.**

Define your models once — like SQLModel — but get the memory footprint of plain dataclasses. SQLDataclass uses pydantic dataclasses (`slots=True`) under the hood, with full pydantic validation, relationships, and native FastAPI support.

## Performance

> All benchmarks run on SQLite with 10,000 rows and 20 fields per row.
> Reproducible via `src/sqldataclass/tests/performance_tests/`.

The benchmarks below compare two model types from this package:

- **`SQLDataclass`** — built on pydantic dataclasses with `slots=True`. Minimal memory overhead, no `__dict__`.
- **`SQLDataclass SQLModel`** — built on Pydantic `BaseModel`. Same SQL table mapping and convenience methods, but gives you the full BaseModel API (`model_dump`, `model_validate`, JSON schema, etc.). Imported as `from sqldataclass import SQLModel`.

Both are compared against SQLAlchemy ORM and [tiangolo's SQLModel](https://github.com/fastapi/sqlmodel).

### Object construction (20 fields, 10k objects, no DB)

| Approach | B/row | Time |
|---|---:|---:|
| dict | 578 | 22 ms |
| stdlib dataclass (`slots=True`) | 306 | 27 ms |
| pydantic dataclass (`slots=True`) | 306 | 191 ms |
| **SQLDataclass** | **322** | **66 ms** |
| SQLAlchemy ORM | 1,690 | 246 ms |
| Pydantic BaseModel | 2,914 | 64 ms |
| **SQLDataclass `SQLModel`** | **2,914** | **67 ms** |
| SQLModel (tiangolo) | 4,538 | 916 ms |

### Database loading — SQLite (10k rows, 20 fields)

| Approach | B/row | Time |
|---|---:|---:|
| Raw SQL → dict | 963 | 124 ms |
| Raw SQL → stdlib dataclass | 691 | 136 ms |
| Raw SQL → pydantic dataclass | 691 | 307 ms |
| **SQLDataclass `load_all`** | **708** | **211 ms** |
| **SQLDataclass `SQLModel` `load_all`** | **1,204** | **170 ms** |
| SQLAlchemy ORM `Session.query` | 2,098 | 167 ms |
| SQLModel (tiangolo) `session.exec` | 2,410 | 169 ms |

Times include query execution + object construction. Both `SQLDataclass` and `SQLModel` `load_all` bypass pydantic's `__init__` overhead — `SQLDataclass` uses `validate_python` (Rust fast path), `SQLModel` uses direct `__dict__` hydration (like the SA ORM does). `SQLModel` `load_all` uses **1.7x less memory than SA ORM** and **2x less than tiangolo's SQLModel** while matching their speed.

### Complex models with relationships (100 teams, 5k heroes, 20 tags, SQLite)

**Teams with heroes (one-to-many, 100 teams + 5k heroes):**

| Library | Memory | Load time | Notes |
|---|---:|---:|---|
| **SQLDataclass** | **1.2 MB** | **13 ms** | Two-query + back-ref stitching, no session |
| SQLAlchemy ORM + joinedload | 8.1 MB | 19 ms | JOIN-based, needs session |

**Heroes with team + tags (many-to-one + many-to-many, 5k heroes, 20 tags, 3 tags/hero):**

| Library | Memory | Load time | Notes |
|---|---:|---:|---|
| **SQLDataclass** | **3.7 MB** | **53 ms** | PK-cached deduplication, no session |
| SQLAlchemy ORM + eager | 12.3 MB | 70 ms | Identity map deduplication |

### When to use what

- **Simple/flat models** — SQLDataclass wins on both memory and speed (3-14x less memory)
- **One-to-many** — SQLDataclass wins (6.7x less memory, faster, with automatic back-references)
- **Many-to-many** — SQLDataclass wins (3.3x less memory, faster, PK-based deduplication)

### Summary

| Benchmark | SQLDataclass | SQLDataclass `SQLModel` | vs SQLAlchemy ORM | vs SQLModel (tiangolo) |
|---|---|---|---|---|
| **DB loading (memory)** | **3x less** | **1.7x less** | baseline | 1.1x less |
| **DB loading (speed)** | same | same | baseline | same |
| **One-to-many (memory)** | **6.7x less** | — | baseline | — |
| **Many-to-many (memory)** | **3.3x less** | — | baseline | — |
| **Object construction (memory)** | **5x less** | 1.7x more | baseline | **1.6x less** |
| **Object construction (speed)** | **4x faster** | **4x faster** | baseline | **14x faster** |

> **SQLDataclass vs SQLDataclass `SQLModel`**: For DB loading, `SQLModel` uses direct `__dict__` hydration (like the SA ORM) — **1,204 B/row**, beating both SA ORM (2,098) and tiangolo (2,410). For object construction (no DB), `SQLModel` uses 2,914 B/row due to `BaseModel.__dict__` overhead vs `SQLDataclass`'s 322 B/row (`slots=True`). Use `SQLDataclass` for maximum memory efficiency; use `SQLModel` when you need the full BaseModel API (`model_dump`, `model_validate`, JSON schema, etc.).

### Why the difference?

- **`slots=True` pydantic dataclasses** — no `__dict__`, minimal per-instance overhead
- **`validate_python` fast path** — bypasses pydantic's `__init__` wrapper, 40% faster than `cls(**row)`
- **SQLAlchemy Core, not ORM** — no session, no identity map, no state tracking overhead
- **PK-cached M2M deduplication** — same tag = same instance, matching ORM's identity map benefit
- **Two-query collections** — one-to-many loaded via `WHERE fk IN (...)`, not expensive JOINs
- **Back-reference stitching** — `hero.team = parent` set directly, zero extra queries

## Install

```bash
pip install sqldataclass
```

For PostgreSQL upsert support:

```bash
pip install sqldataclass[postgres]
```

## Quick start

### Define your model

One class. That's it. No separate schema and domain model.

```python
from sqldataclass import SQLDataclass, Field

class Hero(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None
```

This creates:
- A **pydantic dataclass** with `slots=True` (validation, FastAPI compat, minimal memory)
- A **SQLAlchemy Table** (for DDL and queries, never instantiated)

### Setup

```python
from sqlalchemy import create_engine

engine = create_engine("sqlite:///app.db")
SQLDataclass.metadata.create_all(engine)

# Bind the engine once — conn becomes optional everywhere
SQLDataclass.bind(engine)
```

### Insert data

```python
hero = Hero(name="Spider-Man", secret_name="Peter Parker")
hero.insert()

# Bulk insert
Hero.insert_many(objects=[
    Hero(name="Iron Man", secret_name="Tony Stark", age=45),
    Hero(name="Thor", secret_name="Thor Odinson", age=1500),
])
```

### Query data

```python
# Load all
heroes = Hero.load_all()

# Filter
heroes = Hero.load_all(where=Hero.c.age > 100)

# Load one
hero = Hero.load_one(where=Hero.c.name == "Spider-Man")

# Order
heroes = Hero.load_all(where=Hero.c.age > 25, order_by=Hero.c.name)
```

### Explicit connections (when you need transaction control)

```python
with engine.begin() as conn:
    hero1.insert(conn)
    hero2.insert(conn)
    # both commit together, or both rollback on error
```

### Working with existing databases

If your tables already exist (managed by Alembic, a DBA, or another service), skip `metadata.create_all()` — just define models that match the existing schema:

```python
from sqldataclass import SQLDataclass, Field

# Match your existing table and column names exactly
class User(SQLDataclass, table=True):
    __tablename__ = "users"          # must match the existing table name
    id: int = Field(primary_key=True)
    email: str
    name: str
    is_active: bool = True

engine = create_engine("postgresql+psycopg2://user:pass@host/mydb")
SQLDataclass.bind(engine)

# No create_all() needed — read and write directly
users = User.load_all(where=User.c.is_active == True)
User.update({"is_active": False}, where=User.c.id == 42)
```

You don't need to define every column — only the ones you read or write. Columns not in your model are simply ignored.

### Migrations with Alembic

SQLDataclass uses standard SQLAlchemy `MetaData`, so [Alembic](https://alembic.sqlalchemy.org/) works out of the box for schema migrations.

**Setup:**

```bash
pip install alembic
alembic init migrations
```

**Edit `migrations/env.py`** — point Alembic at your SQLDataclass metadata:

```python
from sqldataclass import SQLDataclass

# Import your models so they register their tables
from myapp.models import Hero, Team  # noqa: F401

target_metadata = SQLDataclass.metadata
```

**Generate and run migrations:**

```bash
# Auto-generate migration from model changes
alembic revision --autogenerate -m "add heroes table"

# Apply migration
alembic upgrade head
```

**Example: adding a column to an existing model**

Start with:

```python
class Hero(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
```

Later, add a field:

```python
class Hero(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    power: str = ""           # new column
    age: int | None = None    # new nullable column
```

Then generate and apply the migration:

```bash
alembic revision --autogenerate -m "add power and age to hero"
alembic upgrade head
```

Alembic auto-detects the diff and generates:

```python
def upgrade():
    op.add_column('hero', sa.Column('power', sa.String(), nullable=False))
    op.add_column('hero', sa.Column('age', sa.Integer(), nullable=True))

def downgrade():
    op.drop_column('hero', 'age')
    op.drop_column('hero', 'power')
```

This works for any schema change: adding/removing columns, renaming tables, changing types, adding indexes, etc.

Use `create_all()` for development/testing and Alembic for production deployments.

## Relationships

SQLDataclass supports all common relationship patterns, loaded eagerly via SQLAlchemy Core — no ORM session required.

### Many-to-one

```python
from sqldataclass import SQLDataclass, Field, Relationship

class Team(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

class Hero(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    team: Team | None = Relationship()  # auto-JOINed on load

hero = Hero.load_one(where=Hero.c.name == "Spider-Man")
print(hero.team.name)  # "Avengers"
```

The FK column (`team_id`) is created automatically from the relationship declaration. If you need to control the column name or set the FK without loading the full object, declare it explicitly:

```python
class Hero(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    team_id: int = Field(foreign_key="team.id")  # explicit FK
    team: Team | None = Relationship()
```

### Cascading insert

When inserting a model with many-to-one relationships, related objects are inserted automatically. Unpersisted related objects (PK is `None`) are inserted first, and their generated PK is copied into the FK column — no manual ordering required:

```python
avengers = Team(name="Avengers")
hero = Hero(name="Spider-Man", team=avengers)
hero.insert()  # inserts avengers first, then hero with team_id set

print(avengers.id)  # 1 — PK populated by the DB
print(hero.team_id)  # 1 — FK set automatically
```

Already-persisted related objects (PK is not `None`) are not re-inserted — only the FK is copied:

```python
avengers = Team(name="Avengers")
avengers.insert()  # insert separately

hero1 = Hero(name="Iron Man", team=avengers)
hero1.insert()  # avengers already has a PK, so only hero1 is inserted

hero2 = Hero(name="Thor", team=avengers)
hero2.insert()  # same — avengers is not inserted again
```

Cascading insert works recursively for nested relationships and has zero ongoing memory cost — it's a one-shot tree walk at insert time, not a session or identity map.

### One-to-many

```python
class Team(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    heroes: list[Hero] = Relationship(back_populates="team")

team = Team.load_one(where=Team.c.name == "Avengers")
print([h.name for h in team.heroes])  # ["Iron Man", "Thor"]
```

One-to-many uses a **two-query strategy** (not JOIN-then-group) for memory efficiency — one query for parents, one `WHERE fk IN (...)` for all children.

### Many-to-many

```python
class HeroTeamLink(SQLDataclass, table=True):
    hero_id: int = Field(primary_key=True, foreign_key="hero.id")
    team_id: int = Field(primary_key=True, foreign_key="team.id")

class Hero(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    teams: list[Team] = Relationship(link_model=HeroTeamLink)

class Team(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    heroes: list[Hero] = Relationship(link_model=HeroTeamLink)

hero = Hero.load_one(where=Hero.c.name == "Wolverine")
print([t.name for t in hero.teams])  # ["Avengers", "X-Men"]
```

### Discriminated unions

For polymorphic data stored in separate tables:

```python
from typing import Literal

class NormalData(SQLDataclass, table=True):
    id: int = Field(primary_key=True, foreign_key="participant.id")
    behavior: Literal["normal"] = "normal"
    p_max: float = 0.0

class BatteryData(SQLDataclass, table=True):
    id: int = Field(primary_key=True, foreign_key="participant.id")
    behavior: Literal["battery"] = "battery"
    capacity: float = 0.0

class Participant(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    behavior: str  # discriminator column
    data: NormalData | BatteryData = Relationship(discriminator="behavior")

p = Participant.load_one(where=Participant.c.name == "Alice")
print(type(p.data).__name__)  # "NormalData"
print(p.data.p_max)           # 100.0
```

### Single-table inheritance

Store multiple subtypes in one table with a discriminator column. Child classes can add their own fields (auto-appended as nullable columns):

```python
class Vehicle(SQLDataclass, table=True):
    __discriminator__ = "type"  # enables single-table inheritance
    id: int | None = Field(default=None, primary_key=True)
    type: str = ""
    name: str = ""

class Car(Vehicle):                    # just inherit — no extra keywords
    doors: int | None = None           # auto-added to Vehicle's table

class Truck(Vehicle):
    payload: float | None = None       # auto-added to Vehicle's table
```

```python
# Insert — discriminator auto-set from class name
Car(name="Civic", doors=4).insert()    # type="car"
Truck(name="F-150", payload=1000).insert()  # type="truck"

# Subtype queries — auto-filtered
cars = Car.load_all()                  # only cars
trucks = Truck.load_all()             # only trucks

# Polymorphic query — returns correct subtypes
all_vehicles = Vehicle.load_all()      # [Car(...), Truck(...), ...]
type(all_vehicles[0])                  # <class 'Car'>

# Scoped update/delete
Car.update({"doors": 2}, where=Car.c.name == "Civic")
Truck.delete()                         # only deletes trucks
```

Override the default discriminator value with `__discriminator_value__`:

```python
class Motorcycle(Vehicle):
    __discriminator_value__ = "moto"   # instead of default "motorcycle"
    wheel_count: int | None = None
```

## Use with FastAPI

Pydantic dataclasses are first-class citizens in FastAPI — no conversion needed:

```python
from fastapi import FastAPI

app = FastAPI()

@app.get("/heroes", response_model=list[Hero])
def get_heroes():
    return Hero.load_all()

@app.get("/heroes/{hero_id}", response_model=Hero)
def get_hero(hero_id: int):
    return Hero.load_one(where=Hero.c.id == hero_id)
```

### Data-only models (API schemas)

Models without `table=True` are pure pydantic dataclasses — useful for request bodies:

```python
class HeroCreate(SQLDataclass):
    name: str
    secret_name: str
    age: int | None = None

@app.post("/heroes", response_model=Hero)
def create_hero(data: HeroCreate):
    hero = Hero(name=data.name, secret_name=data.secret_name, age=data.age)
    hero.insert()
    return hero
```

## Field options

`Field()` accepts both pydantic and SQLAlchemy parameters:

```python
class User(SQLDataclass, table=True):
    id: int | None = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True, max_length=255)
    name: str = Field(min_length=1, max_length=100)
    age: int = Field(ge=0, le=200)
    team_id: int | None = Field(default=None, foreign_key="teams.id")
    # Not stored in DB — only exists on the Python object
    display_name: str = Field(default="", column=False)
    is_cached: bool = Field(default=False, column=False)
```

| Parameter | Type | Description |
|---|---|---|
| `primary_key` | `bool` | Mark as primary key |
| `index` | `bool` | Create database index |
| `unique` | `bool` | Add unique constraint |
| `foreign_key` | `str` | Foreign key reference (e.g. `"users.id"`) |
| `nullable` | `bool` | Override nullable inference |
| `sa_type` | `TypeEngine` | Override SQLAlchemy column type |
| `default` | `Any` | Default value |
| `ge`, `le`, `gt`, `lt` | `float` | Pydantic numeric validators |
| `min_length`, `max_length` | `int` | Pydantic string validators |
| `pattern` | `str` | Pydantic regex pattern |
| `column` | `bool` | `False` = field exists on Python object but not in DB |

## Relationship options

`Relationship()` marks a field as loaded from a related table — not stored as a column.

| Parameter | Type | Description |
|---|---|---|
| `back_populates` | `str` | Inverse field name on the child model |
| `link_model` | `type` | Link table class for many-to-many |
| `discriminator` | `str` | Column name for discriminated unions |
| `order_by` | `str` | Column name to sort collection children by |
| `default` | `Any` | Default value (`None` for many-to-one, `[]` for collections) |

## Custom type annotations

SQLDataclass doesn't bundle domain-specific type annotations (e.g. numpy), but you can define your own in your project and use them seamlessly with pydantic's `Annotated` types:

```python
# your_project/annotations.py
from functools import partial
from typing import Annotated

import numpy as np
import numpy.typing as npt
from pydantic import BeforeValidator, PlainSerializer


def _to_np_array(dtype, x):
    return np.asarray(x, dtype=dtype)


class Np:
    """Numpy type annotations with auto-serialization."""

    float64 = Annotated[
        np.float64,
        PlainSerializer(float, return_type=float, when_used="always"),
        BeforeValidator(np.float64),
    ]
    int64 = Annotated[
        np.int64,
        PlainSerializer(int, return_type=int, when_used="always"),
        BeforeValidator(np.int64),
    ]

    class Array:
        float64 = Annotated[
            npt.NDArray[np.float64],
            PlainSerializer(lambda x: x.tolist(), return_type=list[float], when_used="always"),
            BeforeValidator(partial(_to_np_array, np.float64)),
        ]
```

Then use them in your models:

```python
from sqldataclass import SQLDataclass
from your_project.annotations import Np

class Measurement(SQLDataclass):
    score: Np.float64
    readings: Np.Array.float64

m = Measurement(score=9.5, readings=[1.0, 2.0, 3.0])
m.dump()  # {"score": 9.5, "readings": [1.0, 2.0, 3.0]}
```

This pattern works for any custom type — numpy, pandas, domain objects, etc. Pydantic's `Annotated` + `BeforeValidator`/`PlainSerializer` handles the conversion automatically.

## API reference

### Model methods

All methods accept an optional `conn` parameter. If omitted, a connection is auto-created from the bound engine (see `SQLDataclass.bind(engine)`).

| Method | Type | Description |
|---|---|---|
| `SQLDataclass.bind(engine)` | classmethod | Bind engine — makes `conn` optional everywhere |
| `Model.select()` | classmethod | Build a `SELECT` query |
| `Model.load_all(conn=, where=, order_by=)` | classmethod | Load all matching rows with relationships |
| `Model.load_one(conn=, where=)` | classmethod | Load one row or `None` |
| `Model.insert_many(conn=, objects=)` | classmethod | Bulk insert |
| `Model.update(values, conn=, where=)` | classmethod | Update matching rows, returns count |
| `Model.delete(conn=, where=)` | classmethod | Delete matching rows, returns count |
| `instance.insert(conn=)` | instance | Insert this row |
| `instance.to_dict(exclude_keys=)` | instance | Flat dict for SQL |
| `instance.upsert(conn=, index_elements=)` | instance | PostgreSQL upsert |
| `Model.c` | attribute | Column access for WHERE clauses |
| `Model.metadata` | attribute | SQLAlchemy MetaData |

### Low-level bridge API

For advanced use cases, the underlying bridge functions are also available:

| Function | Description |
|---|---|
| `load_all(conn, query, cls)` | Execute query, construct instances inline |
| `fetch_all(conn, query)` | Execute query, return `list[dict]` |
| `fetch_one(conn, query)` | Execute query, return single `dict` or `None` |
| `insert_row(conn, table_class, values)` | Insert a single row |
| `insert_many(conn, table_class, rows)` | Insert multiple rows |
| `flatten_for_table(obj, exclude_keys=)` | Flatten dataclass to dict |
| `nest_fields(data, field_name, keys)` | Reshape flat dict for nested models |
| `format_discriminated(data, cls, ...)` | Reshape flat row for discriminated unions |

## Design philosophy

1. **One class, one definition** — no separate SQL schema and domain model
2. **Memory-first** — pydantic dataclasses with `slots=True` match stdlib dataclass footprint
3. **SQLAlchemy Core, not ORM** — explicit queries, no hidden state tracking
4. **Relationships without a session** — eager loading via JOINs and two-query strategy
5. **FastAPI native** — pydantic dataclasses work as response models out of the box
6. **Escape hatches** — low-level bridge API available when you need full control

## Acknowledgements

SQLDataclass was born from combining two lines of work:

**[SQLModel](https://github.com/fastapi/sqlmodel)** by Sebastián Ramírez and its contributors provided the inspiration for the single-class API — one model definition that serves as both the database schema and the pydantic data model. SQLDataclass recreates this developer experience while targeting lower memory consumption by building on pydantic dataclasses and SQLAlchemy Core instead of the full ORM.

## Requirements

- Python 3.11+
- pydantic >= 2.0
- sqlalchemy >= 2.0

## License

MIT
