RafikiDB Python SDK#

sdk-python

Use when writing Python code with the RafikiDB Python SDK (rafikidb on PyPI). Covers create_client, DataBuilder filters/joins/cursor pagination, auth sessions, realtime (WebSocket/SSE), storage, env, secrets, webhooks, functions, payments, envelope and error model.

You are working with the rafikidb PyPI package (Python 3.10+, zero required dependencies; optional websockets extra for WebSocket realtime). Install: pip install rafikidb or pip install rafikidb[realtime].

Setup#

from rafikidb import create_client

db = create_client(
    project_id="5c73dab0-...",
    api_key="raf_live_...",
    base_url="https://api.rafikidb.com/api/v1",  # optional
)
Python
  • Every call returns Envelope[T] with success, message, data (None when absent) and errors: list[str].
  • Failures raise RafikiDBError with message, status, errors, and code (invalid_input, unauthorized, forbidden, not_found, conflict, quota_exceeded, internal_error, derived from status).

Data#

rows = (
    db.from_("profiles")
    .select("id, nationality")
    .eq("nationality", "Tanzania")
    .gt("age", 18)
    .in_("region", ["dar", "mba"])
    .order("created_at", ascending=False)
    .limit(25)
    .execute()
)
rows.success  # True
rows.data     # list[dict] | None
Python
  • Filters: eq, neq, gt, gte, lt, lte, like, ilike, is_null, is_not_null, in_.
  • Pagination: limit, offset, cursor(id) (keyset, for large tables).
  • Single row: .get(id); first row or None: .single(); count: .head().
  • The builder method is from_ (underscore suffix; from is reserved).

Joins (no SQL, via query engine)#

from rafikidb import JoinSpec

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

Joined columns appear as {table}.{column}. Joins route through POST /projects/{id}/query/run. JoinSpec.on takes a raw SQL clause and overrides the column pair.

Writes#

db.from_("profiles").insert({"full_name": "Asha Mwinyi", "nationality": "Tanzania"})
db.from_("profiles").insert([{...}, {...}])

db.from_("profiles").eq("id", id).update({"nationality": "Kenya"})
db.from_("profiles").eq("id", id).delete()
Python

update() / delete() require an .eq("id", ...) filter, else ValueError.

Auth (project users, not dashboard)#

db.auth.signup(email, password, full_name, phone, metadata=None)
session = db.auth.login(email, password)   # alias: signin

db.auth.otp_request("+255712345678")
db.auth.otp_verify(phone, code, full_name)

db.auth.reset_password(email)
db.auth.confirm_reset_password(email, code, new_password)

db.auth.sign_out()
db.auth.refresh(refresh_token)
db.auth.logout(refresh_token)
Python
  • Login/signup/OTP-verify auto-store the session; every request then carries the Bearer token so RLS sees the authenticated user (db.client.access_token exposes the JWT).
  • Persist by saving db.session yourself and passing it back: create_client(..., session=stored).

Realtime#

sub = db.realtime.subscribe(
    "messages",
    on_event,                                  # Callable[[RealtimeEvent], None]
    events=["INSERT", "DELETE"],               # optional filter
    transport="auto",                          # auto | websocket | sse
)
sub.close()   # stop
sub.wait()    # block until stopped (returns bool)
Python
  • auto uses WebSocket when the websockets package is installed, else SSE.
  • RealtimeEvent dataclass: id, type, table, project_id, record, old_record (None), created_at.
  • SSE reconnects every 3s; WebSocket reconnects with exponential backoff min(1.0 * 2**retry, 30.0).

Modules#

db.storage.create_bucket(name="avatars", slug="avatars", is_public=True)
db.storage.signed_upload_url(bucket_id=b["id"], object_name="user-1.png")

db.env.set("STRIPE_KEY", "sk_test_123", environment="production")
db.env.bulk_set("development", {"DEBUG": "true"})

db.secrets.set(key="API_SECRET", value="s3cr3t")
db.secrets.reveal(secret_id)

db.webhooks.create(name="order.created", url="https://myapp.com/hooks/orders", events=["order.created"])
db.webhooks.list_deliveries(webhook_id)

db.functions.create(name="hello", code="...")
db.functions.deploy(fn_id)
db.functions.invoke(fn_id, method="POST", body="{}")

db.payments.stk_push(phone="+255712345678", amount=5000, description="Order #123")
db.payments.snippe_status(reference)
db.payments.list_transactions()
Python

Gotchas#

  • Builder method is from_, filters use snake_case (in_, is_null).
  • update/delete raise ValueError without an .eq("id", ...) filter.
  • Dashboard login (/auth/login) is separate - not for app users.
  • Tests run with python -m unittest discover -s tests -q.