Python

pip install requests

"""Minimal Baselyra client. Nothing here is Baselyra-specific beyond the paths."""
from __future__ import annotations

from typing import Any
import requests


class BaselyraError(Exception):
    def __init__(self, code: str, message: str, status: int) -> None:
        super().__init__(f"{status} {code}: {message}")
        self.code = code
        self.message = message
        self.status = status


class Baselyra:
    def __init__(self, url: str, key: str, access_token: str | None = None) -> None:
        self.url = url.rstrip("/")
        self.key = key
        self.access_token = access_token
        self.session = requests.Session()

    @property
    def _headers(self) -> dict[str, str]:
        return {
            "apikey": self.key,
            "authorization": f"Bearer {self.access_token or self.key}",
            "content-type": "application/json",
        }

    def _request(self, method: str, path: str, *, json: Any = None,
                 params: dict[str, str] | None = None,
                 headers: dict[str, str] | None = None) -> requests.Response:
        res = self.session.request(
            method,
            f"{self.url}{path}",
            json=json,
            params=params,
            headers={**self._headers, **(headers or {})},
            timeout=30,
        )
        if res.status_code >= 400:
            try:
                error = res.json().get("error", {})
            except ValueError:
                error = {}
            raise BaselyraError(
                error.get("code", "http_error"),
                error.get("message", res.reason),
                res.status_code,
            )
        return res

    # ---- auth ----

    def sign_in(self, email: str, password: str) -> dict[str, Any]:
        session = self._request(
            "POST", "/auth/v1/token", params={"grant_type": "password"},
            json={"email": email, "password": password},
        ).json()
        self.access_token = session["access_token"]
        return session

    def refresh(self, refresh_token: str) -> dict[str, Any]:
        """A refresh token is single-use: store BOTH values it returns."""
        session = self._request(
            "POST", "/auth/v1/token", params={"grant_type": "refresh_token"},
            json={"refresh_token": refresh_token},
        ).json()
        self.access_token = session["access_token"]
        return session

    # ---- data ----

    def select(self, table: str, **params: str) -> list[dict[str, Any]]:
        return self._request("GET", f"/rest/v1/{table}", params=params).json()

    def count(self, table: str, **params: str) -> int:
        res = self._request(
            "GET", f"/rest/v1/{table}", params={**params, "limit": "1"},
            headers={"Prefer": "count=exact"},
        )
        total = res.headers.get("content-range", "*/0").split("/")[-1]
        return 0 if total == "*" else int(total)

    def insert(self, table: str, row: dict[str, Any] | list[dict[str, Any]]) -> list[dict[str, Any]]:
        return self._request(
            "POST", f"/rest/v1/{table}", json=row,
            headers={"Prefer": "return=representation"},
        ).json()

    def upsert(self, table: str, rows: list[dict[str, Any]], on_conflict: str) -> list[dict[str, Any]]:
        return self._request(
            "POST", f"/rest/v1/{table}", json=rows,
            params={"on_conflict": on_conflict},
            headers={"Prefer": "return=representation,resolution=merge-duplicates"},
        ).json()

    def update(self, table: str, filters: dict[str, str], patch: dict[str, Any]) -> list[dict[str, Any]]:
        if not filters:
            raise ValueError("a filter is required: an unfiltered PATCH rewrites the table")
        return self._request(
            "PATCH", f"/rest/v1/{table}", params=filters, json=patch,
            headers={"Prefer": "return=representation"},
        ).json()

    def delete(self, table: str, filters: dict[str, str]) -> None:
        if not filters:
            raise ValueError("a filter is required: an unfiltered DELETE empties the table")
        self._request("DELETE", f"/rest/v1/{table}", params=filters)

    def rpc(self, fn: str, **args: Any) -> Any:
        return self._request("POST", f"/rest/v1/rpc/{fn}", json=args).json()

    def paginate(self, table: str, page_size: int = 1000, **params: str):
        """Yield every row a filter matches, one page at a time."""
        offset = 0
        while True:
            page = self.select(table, limit=str(page_size), offset=str(offset), **params)
            if not page:
                return
            yield from page
            if len(page) < page_size:
                return
            offset += page_size

    # ---- storage ----

    def upload(self, bucket: str, key: str, data: bytes, content_type: str) -> dict[str, Any]:
        res = self.session.post(
            f"{self.url}/storage/v1/object/{bucket}/{key}",
            data=data,
            headers={
                "authorization": f"Bearer {self.access_token or self.key}",
                "content-type": content_type,
            },
            timeout=120,
        )
        res.raise_for_status()
        return res.json()

    def signed_url(self, bucket: str, key: str, expires_in: int = 3600) -> str:
        return self._request(
            "POST", f"/storage/v1/object/sign/{bucket}/{key}",
            json={"expiresIn": expires_in},
        ).json()["url"]

import os
from baselyra import Baselyra

# The service key bypasses RLS, so this sees everything.
bl = Baselyra("https://api.example.com", os.environ["BASELYRA_SERVICE_KEY"])

for user in bl.paginate("profiles", select="id,email", plan="eq.free"):
    print(user["email"])

bl.upsert(
    "metrics",
    [{"day": "2026-08-23", "signups": 412}],
    on_conflict="day",
)

bl = Baselyra("https://api.example.com", ANON_KEY)
bl.sign_in("ada@example.com", "correct-horse-battery")
print(bl.select("notes", select="id,title"))     # only Ada's, decided by RLS

total = bl.count("orders", status="eq.paid")
print(f"{total} paid orders")

# limit is clamped to 10000 by the server, so page rather than asking for everything
for order in bl.paginate("orders", page_size=1000, status="eq.paid", order="created_at.asc"):
    process(order)

rows = bl.rpc("search_articles", term="postgres", max_results=5)

from decimal import Decimal

row = bl.select("invoices", select="amount", id="eq.1")[0]
row["amount"]                 # '1250.00' — a str, not a float
Decimal(row["amount"])        # what to do with it