SupabaseModel

SupabaseModel

Bases: BaseModel

Base class for tables — Pydantic model + PostgREST chain entry point.

Subclass with table-creation kwargs:

• table — PostgREST table / view name. Required on concrete subclasses.
• pk — Primary key field name. Default "id".
• select — Override the auto-derived select string (escape hatch).
• query_class — Custom :class:QueryBuilder subclass for .query; inherited via MRO, so a project-wide base class can set it once.

Example

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

pet = await Pet.create(name="Whiskers", species="cat")
cats = await Pet.query.eq("species", "cat").all()

get async classmethod

get(pk_value: Any) -> Self

Fetch by primary key. Raises SupabaseORMDoesNotExist on miss.

Source code in src/supabase_orm/_async/_base.py

@classmethod
async def get(cls, pk_value: Any) -> Self:
    """Fetch by primary key. Raises ``SupabaseORMDoesNotExist`` on miss."""
    row = await cls.query.eq(cls.__pk__, pk_value).maybe_one()
    if row is None:
        raise SupabaseORMDoesNotExist(
            f"{cls.__name__}({cls.__pk__}={pk_value!r}) not found"
        )
    return row

find async classmethod

find(pk_value: Any) -> Self | None

Fetch by primary key. Returns None on miss.

Source code in src/supabase_orm/_async/_base.py

@classmethod
async def find(cls, pk_value: Any) -> Self | None:
    """Fetch by primary key. Returns ``None`` on miss."""
    return await cls.query.eq(cls.__pk__, pk_value).maybe_one()

create async classmethod

create(
    *,
    returning: Literal["representation"] = ...,
    **values: Any,
) -> Self

create(
    *, returning: Literal["minimal"], **values: Any
) -> None

create(
    *,
    returning: ReturnMode = "representation",
    **values: Any,
) -> Self | None

Insert a row in a single round-trip.

Parameters:

Name	Type	Description	Default
returning	ReturnMode	"minimal" skips the response body and returns None — useful when RLS forbids SELECT on the row, or for pure side-effect writes.	'representation'
**values	Any	Column values for the new row.	{}

Returns:

Type	Description
Self | None	The inserted row, or None when returning="minimal".

Source code in src/supabase_orm/_async/_base.py

@classmethod
async def create(
    cls, *, returning: ReturnMode = "representation", **values: Any
) -> Self | None:
    """Insert a row in a single round-trip.

    Args:
        returning: ``"minimal"`` skips the response body and returns
            ``None`` — useful when RLS forbids SELECT on the row, or
            for pure side-effect writes.
        **values: Column values for the new row.

    Returns:
        The inserted row, or ``None`` when ``returning="minimal"``.
    """
    validate_returning(returning)
    payload = {k: serialize(v) for k, v in values.items()}
    client = get_client()
    ins = await execute_logged(
        client.table(cls.__table__).insert(payload, returning=returning)
    )
    if returning == "minimal":
        return None
    if not ins.data:
        raise ValueError(f"{cls.__name__}.create returned no rows")
    if cls.__relations__:
        return await cls.get(ins.data[0][cls.__pk__])
    return cls.model_validate(ins.data[0])

bulk_create async classmethod

bulk_create(
    rows: list[dict[str, Any]],
    *,
    returning: Literal["representation"] = ...,
) -> list[Self]

bulk_create(
    rows: list[dict[str, Any]],
    *,
    returning: Literal["minimal"],
) -> None

bulk_create(
    rows: list[dict[str, Any]],
    *,
    returning: ReturnMode = "representation",
) -> list[Self] | None

Insert multiple rows in one round-trip.

Parameters:

Name	Type	Description	Default
rows	list[dict[str, Any]]	One dict of column values per row.	required
returning	ReturnMode	"minimal" skips the response body and returns None.	'representation'

Returns:

Type	Description
list[Self] | None	Inserted rows; [] for empty input; None when returning="minimal".

Source code in src/supabase_orm/_async/_base.py

@classmethod
async def bulk_create(
    cls,
    rows: list[dict[str, Any]],
    *,
    returning: ReturnMode = "representation",
) -> list[Self] | None:
    """Insert multiple rows in one round-trip.

    Args:
        rows: One dict of column values per row.
        returning: ``"minimal"`` skips the response body and returns ``None``.

    Returns:
        Inserted rows; ``[]`` for empty input; ``None`` when ``returning="minimal"``.
    """
    validate_returning(returning)
    if not rows:
        return None if returning == "minimal" else []
    payload = [{k: serialize(v) for k, v in r.items()} for r in rows]
    client = get_client()
    ins = await execute_logged(
        client.table(cls.__table__).insert(payload, returning=returning)
    )
    if returning == "minimal":
        return None
    data = ins.data or []
    if not data:
        return []
    if cls.__relations__:
        ids = [r[cls.__pk__] for r in data]
        return await cls.query.in_(cls.__pk__, ids).all()
    adapter = cls.__list_adapter__
    return (
        adapter.validate_python(data)
        if adapter
        else [cls.model_validate(r) for r in data]
    )

upsert async classmethod

upsert(
    *,
    on_conflict: "str | Column | list[str | Column] | None" = ...,
    ignore_duplicates: Literal[False] = ...,
    returning: Literal["representation"] = ...,
    **values: Any,
) -> Self

upsert(
    *,
    on_conflict: "str | Column | list[str | Column] | None" = ...,
    ignore_duplicates: Literal[True],
    returning: Literal["representation"] = ...,
    **values: Any,
) -> Self | None

upsert(
    *,
    on_conflict: "str | Column | list[str | Column] | None" = ...,
    ignore_duplicates: bool = ...,
    returning: Literal["minimal"],
    **values: Any,
) -> None

upsert(
    *,
    on_conflict: "str | Column | list[str | Column] | None" = None,
    ignore_duplicates: bool = False,
    returning: ReturnMode = "representation",
    **values: Any,
) -> Self | None

Insert or update on conflict.

Parameters:

Name	Type	Description	Default
on_conflict	'str | Column | list[str | Column] | None'	Unique-constraint column(s) used to detect duplicates. Accepts a typed :class:Column (e.g. Pet.f.email), a list of columns for composite uniques, or a raw constraint string. None falls back to the table's PK.	None
ignore_duplicates	bool	Keep the existing row unchanged on conflict; returns None since PostgREST sends no body in this case.	False
returning	ReturnMode	"minimal" skips the body and returns None.	'representation'
**values	Any	Column values for the row.	{}

Returns:

Type	Description
Self | None	The row; None for ignore_duplicates=True or returning="minimal".

Source code in src/supabase_orm/_async/_base.py

@classmethod
async def upsert(
    cls,
    *,
    on_conflict: "str | Column | list[str | Column] | None" = None,
    ignore_duplicates: bool = False,
    returning: ReturnMode = "representation",
    **values: Any,
) -> Self | None:
    """Insert or update on conflict.

    Args:
        on_conflict: Unique-constraint column(s) used to detect duplicates.
            Accepts a typed :class:`Column` (e.g. ``Pet.f.email``), a list
            of columns for composite uniques, or a raw constraint string.
            ``None`` falls back to the table's PK.
        ignore_duplicates: Keep the existing row unchanged on conflict;
            returns ``None`` since PostgREST sends no body in this case.
        returning: ``"minimal"`` skips the body and returns ``None``.
        **values: Column values for the row.

    Returns:
        The row; ``None`` for ``ignore_duplicates=True`` or ``returning="minimal"``.
    """
    validate_returning(returning)
    payload = {k: serialize(v) for k, v in values.items()}
    kw: dict[str, Any] = {
        "ignore_duplicates": ignore_duplicates,
        "returning": returning,
    }
    normalized = _coerce_on_conflict(on_conflict)
    if normalized is not None:
        kw["on_conflict"] = normalized
    resp = await execute_logged(
        get_client().table(cls.__table__).upsert(payload, **kw)
    )
    if returning == "minimal":
        return None
    if not resp.data:
        if ignore_duplicates:
            return None
        raise ValueError(f"{cls.__name__}.upsert returned no rows")
    if cls.__relations__:
        return await cls.get(resp.data[0][cls.__pk__])
    return cls.model_validate(resp.data[0])

bulk_upsert async classmethod

bulk_upsert(
    rows: list[dict[str, Any]],
    *,
    on_conflict: "str | Column | list[str | Column] | None" = ...,
    ignore_duplicates: bool = ...,
    returning: Literal["representation"] = ...,
) -> list[Self]

bulk_upsert(
    rows: list[dict[str, Any]],
    *,
    on_conflict: "str | Column | list[str | Column] | None" = ...,
    ignore_duplicates: bool = ...,
    returning: Literal["minimal"],
) -> None

bulk_upsert(
    rows: list[dict[str, Any]],
    *,
    on_conflict: "str | Column | list[str | Column] | None" = None,
    ignore_duplicates: bool = False,
    returning: ReturnMode = "representation",
) -> list[Self] | None

Bulk-upsert.

Parameters:

Name	Type	Description	Default
rows	list[dict[str, Any]]	One dict of column values per row.	required
on_conflict	'str | Column | list[str | Column] | None'	See :meth:upsert.	None
ignore_duplicates	bool	See :meth:upsert.	False
returning	ReturnMode	See :meth:upsert.	'representation'

Returns:

Type	Description
list[Self] | None	Resulting rows; [] for empty input; None when returning="minimal".

Source code in src/supabase_orm/_async/_base.py

@classmethod
async def bulk_upsert(
    cls,
    rows: list[dict[str, Any]],
    *,
    on_conflict: "str | Column | list[str | Column] | None" = None,
    ignore_duplicates: bool = False,
    returning: ReturnMode = "representation",
) -> list[Self] | None:
    """Bulk-upsert.

    Args:
        rows: One dict of column values per row.
        on_conflict: See :meth:`upsert`.
        ignore_duplicates: See :meth:`upsert`.
        returning: See :meth:`upsert`.

    Returns:
        Resulting rows; ``[]`` for empty input; ``None`` when ``returning="minimal"``.
    """
    validate_returning(returning)
    if not rows:
        return None if returning == "minimal" else []
    payload = [{k: serialize(v) for k, v in r.items()} for r in rows]
    kw: dict[str, Any] = {
        "ignore_duplicates": ignore_duplicates,
        "returning": returning,
    }
    normalized = _coerce_on_conflict(on_conflict)
    if normalized is not None:
        kw["on_conflict"] = normalized
    resp = await execute_logged(
        get_client().table(cls.__table__).upsert(payload, **kw)
    )
    if returning == "minimal":
        return None
    data = resp.data or []
    if not data:
        return []
    if cls.__relations__:
        ids = [r[cls.__pk__] for r in data]
        return await cls.query.in_(cls.__pk__, ids).all()
    adapter = cls.__list_adapter__
    return (
        adapter.validate_python(data)
        if adapter
        else [cls.model_validate(r) for r in data]
    )

get_or_create async classmethod

get_or_create(
    *, defaults: dict[str, Any] | None = None, **lookup: Any
) -> tuple[Self, bool]

Fetch the row matching lookup, or create it.

Two round-trips; not race-safe — between the lookup and the insert another writer can land a matching row. For atomic semantics, prefer :meth:upsert with a unique constraint.

Parameters:

Name	Type	Description	Default
defaults	dict[str, Any] | None	Extra column values applied only on the create branch.	None
**lookup	Any	Equality filters used to locate the existing row.	{}

Returns:

Type	Description
tuple[Self, bool]	(obj, created) — created is True on insert, False if it already existed.

Source code in src/supabase_orm/_async/_base.py

@classmethod
async def get_or_create(
    cls,
    *,
    defaults: dict[str, Any] | None = None,
    **lookup: Any,
) -> tuple[Self, bool]:
    """Fetch the row matching ``lookup``, or create it.

    Two round-trips; not race-safe — between the lookup and the insert
    another writer can land a matching row. For atomic semantics, prefer
    :meth:`upsert` with a unique constraint.

    Args:
        defaults: Extra column values applied only on the create branch.
        **lookup: Equality filters used to locate the existing row.

    Returns:
        ``(obj, created)`` — ``created`` is ``True`` on insert, ``False`` if it already existed.
    """
    q = cls.query
    for col, val in lookup.items():
        q = q.eq(col, val)
    existing = await q.maybe_one()
    if existing is not None:
        return existing, False
    return await cls.create(**{**lookup, **(defaults or {})}), True

update_or_create async classmethod

update_or_create(
    *, defaults: dict[str, Any] | None = None, **lookup: Any
) -> tuple[Self, bool]

Update the row matching lookup (with defaults), or create.

Same race caveats as :meth:get_or_create.

Parameters:

Name	Type	Description	Default
defaults	dict[str, Any] | None	Column values applied on update, or on the create branch.	None
**lookup	Any	Equality filters used to locate the existing row.	{}

Returns:

Type	Description
tuple[Self, bool]	(obj, created) — created is True on insert.

Source code in src/supabase_orm/_async/_base.py

@classmethod
async def update_or_create(
    cls,
    *,
    defaults: dict[str, Any] | None = None,
    **lookup: Any,
) -> tuple[Self, bool]:
    """Update the row matching ``lookup`` (with ``defaults``), or create.

    Same race caveats as :meth:`get_or_create`.

    Args:
        defaults: Column values applied on update, or on the create branch.
        **lookup: Equality filters used to locate the existing row.

    Returns:
        ``(obj, created)`` — ``created`` is ``True`` on insert.
    """
    q = cls.query
    for col, val in lookup.items():
        q = q.eq(col, val)
    existing = await q.maybe_one()
    if existing is not None:
        if defaults:
            await existing.update(**defaults)
        return existing, False
    return await cls.create(**{**lookup, **(defaults or {})}), True

update async

update(**values: Any) -> Self

Assign the given fields and persist in one call.

Equivalent to setting each attribute and calling :meth:save. Each assignment runs through Pydantic's validator (validate_assignment is on), so type errors surface before the round-trip.

Parameters:

Name	Type	Description	Default
**values	Any	Field=value pairs to set. Must include at least one; cannot target the primary key or a relation field.	{}

Source code in src/supabase_orm/_async/_base.py

async def update(self, **values: Any) -> Self:
    """Assign the given fields and persist in one call.

    Equivalent to setting each attribute and calling :meth:`save`. Each
    assignment runs through Pydantic's validator (``validate_assignment``
    is on), so type errors surface before the round-trip.

    Args:
        **values: Field=value pairs to set. Must include at least one;
            cannot target the primary key or a relation field.
    """
    if not values:
        raise SupabaseORMUsageError(
            "instance.update() requires at least one key=value."
        )
    cls = type(self)
    for k in values:
        if k == cls.__pk__:
            raise SupabaseORMUsageError(
                f"Cannot update primary key {cls.__pk__!r} via .update()."
            )
        if k in cls.__relations__:
            raise SupabaseORMUsageError(
                f"{k!r} is a relation, not a column on {cls.__table__!r}."
            )
    for k, v in values.items():
        setattr(self, k, v)
    return await self.save()

save async

save() -> Self

Persist dirty fields and refresh local state in one round-trip.

For flat models the UPDATE returns the new row directly. For models with relations we still need a follow-up GET to populate embeds.

Source code in src/supabase_orm/_async/_base.py

async def save(self) -> Self:
    """Persist dirty fields and refresh local state in one round-trip.

    For flat models the UPDATE returns the new row directly. For models
    with relations we still need a follow-up GET to populate embeds.
    """
    cls = type(self)
    dirty = self.__pydantic_fields_set__ - {cls.__pk__} - cls.__relations__.keys()
    if not dirty:
        return self
    payload = {f: serialize(getattr(self, f)) for f in dirty}
    client = get_client()
    pk_val = serialize(getattr(self, cls.__pk__))
    resp = await execute_logged(
        client.table(cls.__table__).update(payload).eq(cls.__pk__, pk_val)
    )
    if not resp.data:
        raise SupabaseORMDoesNotExist(
            f"{cls.__name__}({cls.__pk__}={getattr(self, cls.__pk__)!r}) "
            "not found during save"
        )
    if cls.__relations__:
        fresh = await cls.get(getattr(self, cls.__pk__))
    else:
        fresh = cls.model_validate(resp.data[0])
    self.__dict__.update(fresh.__dict__)
    object.__setattr__(self, "__pydantic_fields_set__", set())
    return self

delete async

delete() -> None

Delete this single row by primary key.

Source code in src/supabase_orm/_async/_base.py

async def delete(self) -> None:
    """Delete this single row by primary key."""
    cls = type(self)
    client = get_client()
    await execute_logged(
        client.table(cls.__table__)
        .delete(returning="minimal")
        .eq(cls.__pk__, serialize(getattr(self, cls.__pk__)))
    )