Python SDK - Data#

The builder mirrors PostgREST-style DX with snake_case methods.

Read#

# Select all
db.from_("profiles").select().execute()

# Select columns
db.from_("profiles").select("id, nationality").execute()

# Single row by primary key
db.from_("profiles").get(record_id)

# First matching row, or None
db.from_("profiles").eq("nationality", "Kenya").single()

# Row count only (no rows transferred)
db.from_("profiles").head()  # data: {"count": 42}

Python

from is a Python keyword, so the builder method is from_ (trailing underscore).

Filters#

Method	SQL
`.eq(col, v)`	`col = v`
`.neq(col, v)`	`col != v`
`.gt(col, v)` / `.gte(col, v)`	`col > v` / `col >= v`
`.lt(col, v)` / `.lte(col, v)`	`col < v` / `col <= v`
`.like(col, '%x%')`	`LIKE`
`.ilike(col, '%x%')`	`ILIKE` (case-insensitive)
`.is_null(col)` / `.is_not_null(col)`	`IS NULL` / `IS NOT NULL`
`.in_(col, ['a','b'])`	`IN (...)`

Filters chain with AND.

Ordering and pagination#

db.from_("profiles").order("created_at", ascending=False).execute()
db.from_("profiles").limit(50).offset(100).execute()

Python

For large tables use keyset pagination (O(1) per page vs O(n) for offset):

page = db.from_("logs").select("id").order("id").limit(100).execute()
while page.data:
    last_id = str(page.data[-1]["id"])
    # process page...
    page = (
        db.from_("logs")
        .select("id")
        .order("id")
        .limit(100)
        .cursor(last_id)
        .execute()
    )

Python

Joins#

Column-picker joins - no SQL needed:

from rafikidb import JoinSpec

rows = (
    db.from_("messages")
    .select("id, message, users.full_name as sender")
    .join(JoinSpec(table="users", from_column="user_id", to_column="id", type="LEFT JOIN"))
    .limit(20)
    .execute()
)

Python

JoinSpec(table, from_column, to_column, type) builds the ON clause
JoinSpec(table="users", on="base.id = join.user_id") raw clause (advanced)
Types: INNER JOIN, LEFT JOIN, RIGHT JOIN, FULL JOIN
Joined columns are keyed as {table}.{column} (or your alias)
Joins route through POST /projects/{id}/query/run

Write#

# Insert single or many (server fills id/created_at)
db.from_("profiles").insert({"full_name": "Asha Mwinyi", "nationality": "Tanzania"})
db.from_("profiles").insert([{"full_name": "Asha"}, {"full_name": "Juma"}])

# Update (requires .eq("id", ...))
db.from_("profiles").eq("id", record_id).update({"nationality": "Kenya"})

# Delete (requires .eq("id", ...))
db.from_("profiles").eq("id", record_id).delete()

Python

update() / delete() raise ValueError without an .eq("id", ...) filter.

Raw request#

db.request(
    f"/projects/{db.client.project_id}/data/profiles",
    method="GET",
    body=None,
)

Python

Next: Auth.