supabase-orm 0.2.1

Lightweight async ORM on top of supabase-py with Pydantic validation.

supabase-orm

A lightweight, Pydantic-native ORM on top of supabase-py — async-first, sync mirror generated.

Documentation · Install · Quick start · Sync mode · Contributing

---

Features

• Pydantic models are tables. One class is your schema, validator, and query entry point — no separate DTO layer.
• Typed query builder. Model.query.eq(...).gte(...) autocompletes; typos raise at call time, not server-side.
• Composable predicates. Pet.f.age >= 5 builds a Predicate; combine with | / & / ~, pass to .or_() / .not_().
• Declarative embeds. Annotate Annotated[Owner, Relation(...)] — the right select= string + !inner / FK hints are inferred.
• Keyset iteration. async for pet in Pet.query.iter(): — constant-time per batch, race-safe under concurrent writes, any table size.
• Async + sync. Async-first for FastAPI; a byte-for-byte sync mirror at supabase_orm.sync is generated via unasync.
• Typed RPC. Call PostgREST functions with row validation, single-row, or scalar coercion in one line.
• Per-request RLS. use_client() with a JWT-authenticated client in a FastAPI middleware — zero leakage between concurrent requests.
• Safe by default. Unfiltered bulk delete() / update() raise unless you opt in explicitly.
• Battle-tested. 500+ mock tests for the wire contract; 80+ integration tests against real Supabase.

---

Install

uv add supabase-orm
# or
pip install supabase-orm

Requires Python 3.11+, supabase-py 2.30+, pydantic 2.13+.

---

Quick start

from uuid import UUID
from typing import Annotated
from supabase_orm import SupabaseModel, Relation, lifespan


class Owner(SupabaseModel, table="owners"):
    id: UUID
    email: str
    is_active: bool


class Pet(SupabaseModel, table="pets"):
    id: UUID
    name: str
    species: str
    adopted: bool
    owner: Annotated[Owner, Relation(join="inner")]


async with lifespan(SUPABASE_URL, SUPABASE_KEY):
    # Chain-style query (sequential AND)
    cats = await Pet.query.eq("species", "cat").order_by("-created_at").limit(10).all()

    # Typed predicates (OR / NOT / boolean composition)
    rescues = await Pet.query.or_(
        Pet.f.species == "cat",
        (Pet.f.species == "dog") & (Pet.f.adopted == False),
    ).all()

    # Writes
    p = await Pet.create(name="Whiskers", species="cat", adopted=False)
    p.name = "Mr. Whiskers"
    await p.save()

    # Stream every matching row, any table size
    async for pet in Pet.query.eq("adopted", False).iter():
        await process(pet)

    # Bulk update / delete (guards block unfiltered ops)
    await Pet.query.eq("adopted", False).update(adopted=True)

Full guide: https://supabase-orm.readthedocs.io — models, predicates, embeds, lifecycle, RPC, extending.

---

Sync mode

Same model classes, same chain syntax, same predicates. Switch the import and drop await / async:

from supabase import create_client
from supabase_orm.sync import SupabaseModel, init, shutdown

class Pet(SupabaseModel, table="pets"):
    id: UUID
    name: str
    species: str

init(create_client(SUPABASE_URL, SUPABASE_KEY))

cats = Pet.query.eq("species", "cat").limit(10).all()
for p in Pet.query.eq("species", "cat").iter():
    process(p)

shutdown()  # optional — process exit drains pools anyway

The sync tree is generated from the async source — no second implementation to keep in sync.

---

Contributing

git clone https://github.com/viperadnan-git/supabase-orm
cd supabase-orm
uv sync --all-groups
uv run pytest                # mock suite (always runs)

Architecture

• src/supabase_orm/_async/ is the canonical implementation.
• src/supabase_orm/_sync/ is generated by scripts/gen_sync.py (unasync-based token rewrite + prose regex + skip-block directive).
• Tests mirror the same layout: tests/_async/ is the source, tests/_sync/ is generated.
• A pre-commit hook (nizm) auto-regenerates and stages the sync mirror whenever _async/**/*.py changes. CI also runs python scripts/gen_sync.py --check to fail on drift.

Skip-block directive

Wrap async-only test code (e.g. concurrency tests using asyncio.gather) so the sync mirror omits it:

# gen_sync: skip-block
async def test_async_only():
    ...
# gen_sync: end-skip

Running tests

uv run pytest                      # mock only (default)
uv run pytest -m integration       # live Supabase

Integration tests need a Supabase project with the test schema. See tests/integration/README.md.

Building docs locally

uv sync --group docs
uv run mkdocs serve   # http://localhost:8000

---

License

Apache License 2.0 — see LICENSE.