RPC helpers

rpc async

rpc(name: str, model: type[T], **params: Any) -> list[T]

Call a Postgres function returning setof. Validates each row.

Parameters:

Name	Type	Description	Default
name	str	SQL function name (must exist in the configured schema).	required
model	type[T]	Pydantic model used to validate each returned row.	required
**params	Any	Named arguments — must match the function's parameter names.	{}

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

async def rpc(name: str, model: type[T], **params: Any) -> list[T]:
    """Call a Postgres function returning ``setof``. Validates each row.

    Args:
        name: SQL function name (must exist in the configured schema).
        model: Pydantic model used to validate each returned row.
        **params: Named arguments — must match the function's parameter names.
    """
    resp = await execute_logged(get_client().rpc(name, _serialize_params(params)))
    rows = resp.data or []
    if not isinstance(rows, list):
        rows = [rows]
    return _list_adapter(model).validate_python(rows)

rpc_one async

rpc_one(name: str, model: type[T], **params: Any) -> T

Like :func:rpc but expects exactly one row; raises otherwise.

Parameters:

Name	Type	Description	Default
name	str	SQL function name.	required
model	type[T]	Pydantic model used to validate the row.	required
**params	Any	Named arguments matching the function signature.	{}

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

async def rpc_one(name: str, model: type[T], **params: Any) -> T:
    """Like :func:`rpc` but expects exactly one row; raises otherwise.

    Args:
        name: SQL function name.
        model: Pydantic model used to validate the row.
        **params: Named arguments matching the function signature.
    """
    rows = await rpc(name, model, **params)
    if not rows:
        raise ValueError(f"rpc({name!r}) returned no rows")
    return rows[0]

rpc_maybe_one async

rpc_maybe_one(
    name: str, model: type[T], **params: Any
) -> T | None

Like :func:rpc but returns the first row, or None if empty.

Parameters:

Name	Type	Description	Default
name	str	SQL function name.	required
model	type[T]	Pydantic model used to validate the row.	required
**params	Any	Named arguments matching the function signature.	{}

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

async def rpc_maybe_one(name: str, model: type[T], **params: Any) -> T | None:
    """Like :func:`rpc` but returns the first row, or ``None`` if empty.

    Args:
        name: SQL function name.
        model: Pydantic model used to validate the row.
        **params: Named arguments matching the function signature.
    """
    rows = await rpc(name, model, **params)
    return rows[0] if rows else None

rpc_scalar async

rpc_scalar(
    name: str, result_type: type[S], **params: Any
) -> S

Call a function returning a scalar (int, str, bool, ...).

Parameters:

Name	Type	Description	Default
name	str	SQL function name.	required
result_type	type[S]	The scalar Python type to coerce the result into.	required
**params	Any	Named arguments matching the function signature.	{}

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

async def rpc_scalar(name: str, result_type: type[S], **params: Any) -> S:
    """Call a function returning a scalar (int, str, bool, ...).

    Args:
        name: SQL function name.
        result_type: The scalar Python type to coerce the result into.
        **params: Named arguments matching the function signature.
    """
    resp = await execute_logged(get_client().rpc(name, _serialize_params(params)))
    return _scalar_adapter(result_type).validate_python(resp.data)