Python Quick Start

WoWSQL Python SDK — Get started in minutes

Get your API key: In your WoWSQL project dashboard, go to Settings → API to find your anon key (wowsql_anon_...) and service role key (wowsql_service_...). The anon key is safe for client-side use (respects Row Level Security). The service role key bypasses RLS — use server-side only.

Installation

Official package: View on PyPI →

pip install wowsql
# pin this release: pip install "wowsql==3.10.0"

Initialize Client

from wowsql import WowSQLClient

# Initialize client — connects to PostgREST at /rest/v1/
client = WowSQLClient(
    project_url="myproject",  # → https://myproject.wowsqlconnect.com
    api_key="wowsql_anon_your_key_here"
)

# Or with full URL
client = WowSQLClient(
    project_url="https://myproject.wowsqlconnect.com",
    api_key="wowsql_anon_your_key_here"
)

# Using context manager (recommended)
with WowSQLClient(project_url="myproject", api_key="wowsql_anon_your_key_here") as client:
    users = client.table("users").select("*").get()

How it works under the hood: The SDK translates your query-builder calls into standard PostgREST HTTP requests. For example, client.table("users").select("*").eq("status","active").limit(10).get() becomes GET /rest/v1/users?select=*&status=eq.active&limit=10 with the apikey header.

Query Data

Basic Queries

# Get all records - CORRECT: Always use .select() before .get() or .execute()
users = client.table("users").select("*").get()

# With filters - CORRECT: Use convenience methods (.gte()) or filter() method
adults = client.table("users") \
    .select("*") \
    .gte("age", 18) \
    .get()

# Or using filter() method
adults = client.table("users") \
    .select("*") \
    .filter("age", "gte", 18) \
    .get()

# Select specific columns - CORRECT
names = client.table("users") \
    .select("id", "name") \
    .get()

Advanced Queries

# Multiple filters with sorting
active_users = client.table("users") \
    .filter({"column": "status", "operator": "eq", "value": "active"}) \
    .filter({"column": "age", "operator": "gt", "value": 18}) \
    .order("created_at", "desc") \
    .limit(20) \
    .offset(0) \
    .get()

print(f"Found {active_users['count']} users")
for user in active_users["data"]:
    print(user["name"])

Filter Helper Methods

# Using helper methods (more readable)
users = client.table("users") \
    .eq("status", "active") \
    .gt("age", 18) \
    .like("email", "%@gmail.com") \
    .is_not_null("email_verified_at") \
    .in_("role", ["admin", "moderator"]) \
    .between("created_at", "2024-01-01", "2024-12-31") \
    .get()

# OR conditions
users = client.table("users") \
    .eq("status", "active") \
    .or_("role", "eq", "admin") \
    .get()

GROUP BY and Aggregates

Basic GROUP BY

# Group by category with counts
result = client.table("products") \
    .select("category", "COUNT(*) as count", "AVG(price) as avg_price") \
    .group_by("category") \
    .get()

# Group by date
result = client.table("orders") \
    .select("DATE(created_at) as date", "COUNT(*) as orders", "SUM(total) as revenue") \
    .group_by("DATE(created_at)") \
    .order_by("date", "desc") \
    .get()

# Multiple GROUP BY columns
result = client.table("sales") \
    .select("region", "category", "SUM(amount) as total") \
    .group_by("region", "category") \
    .get()

HAVING Clause

# Filter aggregated results
result = client.table("products") \
    .select("category", "COUNT(*) as count") \
    .group_by("category") \
    .having("COUNT(*)", "gt", 10) \
    .get()

# Multiple HAVING conditions
result = client.table("orders") \
    .select("DATE(created_at) as date", "SUM(total) as revenue") \
    .group_by("DATE(created_at)") \
    .having("SUM(total)", "gt", 1000) \
    .having("COUNT(*)", "gte", 5) \
    .get()

Multiple ORDER BY

# Order by multiple columns
result = client.table("products") \
    .select("*") \
    .order_by([("category", "asc"), ("price", "desc"), ("created_at", "desc")]) \
    .get()

CRUD Operations

Create Record

result = client.table("users").create({
    "name": "John Doe",
    "email": "john@example.com",
    "age": 30
})

print(f"Created user with ID: {result['id']}")

Update Record

client.table("users").update(123, {
    "email": "newemail@example.com"
})

print("User updated successfully")

Delete Record

client.table("users").delete(123)
print("User deleted successfully")

Get Single Record by ID

user = client.table("users").get_by_id(123)
print(f"User: {user['name']}")

Get First Record

# Get first matching record
user = client.table("users") \
    .filter({"column": "status", "operator": "eq", "value": "active"}) \
    .first()

if user:
    print(f"First active user: {user['name']}")

Filter Operators

Available operators for filtering data:

Authentication

Auth endpoints are at /auth/v1/ on your project URL. Use the same API key.

Sign Up & Sign In

from wowsql import ProjectAuthClient

auth = ProjectAuthClient(
    project_url="myproject",   # → https://myproject.wowsqlconnect.com
    api_key="wowsql_anon_your_key_here"
)

# Sign up → POST /auth/v1/signup
result = auth.sign_up(
    email="user@example.com",
    password="SecurePassword123",
    full_name="John Doe"
)
print(f"Access token: {result.session.access_token}")

# Sign in → POST /auth/v1/login (returns a session, or a 2FA challenge)
login = auth.sign_in(
    email="user@example.com",
    password="SecurePassword123"
)
if login.requires_2fa:
    # TOTP is on for this user: finish with verify_two_factor_login (see Two-Factor Authentication)
    print(f"2FA required, method: {login.method}")
else:
    print(f"Access token: {login.session.access_token}")

Direct REST Call (Auth)

import requests

PROJECT_URL = "https://myproject.wowsqlconnect.com"
API_KEY = "wowsql_anon_your_key_here"

# Sign up directly via REST
resp = requests.post(f"{PROJECT_URL}/auth/v1/signup", json={
    "email": "user@example.com",
    "password": "SecurePassword123",
    "full_name": "John Doe"
}, headers={"apikey": API_KEY})

# Sign in directly via REST
resp = requests.post(f"{PROJECT_URL}/auth/v1/login", json={
    "email": "user@example.com",
    "password": "SecurePassword123"
}, headers={"apikey": API_KEY})
session = resp.json()  # { "access_token": "...", "refresh_token": "...", "user": {...} }

CAPTCHA (Turnstile)

Optional. Only required when dashboard Attack Protection → Turnstile is enabled. Create a Cloudflare Turnstile widget for your app domain and paste the site key and secret in Attack Protection (WoWSQL does not share a platform widget). GET /auth/v1/settings returns your captcha.site_key. Render Turnstile with that key, then pass the token. Omit it when captcha is off — existing calls are unchanged.

auth.sign_up(email="user@example.com", password="SecurePassword123", captcha_token=turnstile_token)
auth.sign_in(email="user@example.com", password="SecurePassword123", captcha_token=turnstile_token)
auth.forgot_password(email="user@example.com", captcha_token=turnstile_token)
auth.send_otp(email="user@example.com", captcha_token=turnstile_token)
auth.send_magic_link(email="user@example.com", captcha_token=turnstile_token)
auth.resend_verification(email="user@example.com", captcha_token=turnstile_token)

Password Reset

# Request password reset
auth.forgot_password(email="user@example.com")

# Reset password with token from email
auth.reset_password(
    token="reset_token_from_email",
    new_password="NewSecurePassword123"
)

OTP Authentication

Provide exactly one of email or phone. Email OTP needs Auth → Emails. Phone OTP needs Auth → SMS / Phone OTP (Twilio, Fast2SMS, or MSG91). purpose: login, signup, or password_reset.

# Email OTP
auth.send_otp(email="user@example.com", purpose="login")
result = auth.verify_otp(email="user@example.com", otp="123456", purpose="login")

# Phone OTP (SMS) — E.164
auth.send_otp(phone="+15551234567", purpose="login")
result = auth.verify_otp(phone="+15551234567", otp="123456", purpose="login")
print(f"Logged in: {result.user.id if result.user else result}")

Magic Link & Email Verification

# Send magic link
auth.send_magic_link(email="user@example.com", purpose="login")

# Verify email with token
auth.verify_email(token="verification_token_from_email")

# Resend verification email
auth.resend_verification(email="user@example.com")

OAuth Authentication

Setup checklist — do this once per provider, per project:
1. Dashboard → Project → Auth → Providers: enable the provider, paste Client ID and Client Secret (no leading/trailing spaces).
2. Copy the Redirect URI shown: https://<slug>.wowsqlconnect.com/auth/v1/oauth/<provider>/callback and register it in your provider's developer console (Google: Cloud Console → Credentials → OAuth 2.0 Client → Authorised redirect URIs).
3. The callback URL is handled entirely by WoWSQL — your app does not send an API key on it. WoWSQL exchanges the code and redirects to your frontend_redirect_uri with ?access_token=&refresh_token=.

# ── Step 1: Server-side — get the authorization URL ─────────────────────────
# Call from your server (Flask route, FastAPI endpoint, etc.).
# Never expose your API key in the browser.
result = auth.get_oauth_authorization_url(
    provider="google",
    frontend_redirect_uri="https://yourapp.com/auth/callback"
)
# Redirect the browser to:
print(result["authorization_url"])
# Google calls: https://<slug>.wowsqlconnect.com/auth/v1/oauth/google/callback
# WowSQL exchanges the code and redirects to your frontend_redirect_uri.


# ── Step 2: Frontend — read tokens from the redirect ────────────────────────
# WowSQL redirects to: https://yourapp.com/auth/callback?access_token=...&refresh_token=...
# Parse them from the URL query string in your frontend JS / template.


# ── (Advanced) Manual code exchange — only if you handle the callback yourself
result = auth.exchange_oauth_callback(
    provider="google",
    code="authorization_code_from_callback",
    redirect_uri="https://yourapp.com/auth/callback"  # optional; the server exchanges the code with the project callback URL
)
print(f"OAuth login successful: {result.user.id if result.user else result.session.access_token}")

Session and Token Refresh

Token refresh calls POST /auth/v1/refresh-token with the stored refresh token and stores the new token pair. From 3.10.0 every SDK uses this route; before 3.10.0 the TypeScript and Java SDKs called routes that do not exist, and Flutter and Rust had no refresh method.

# Current user → GET /auth/v1/me (uses the stored access token)
me = auth.get_user()

# Refresh tokens → POST /auth/v1/refresh-token (uses the stored refresh token)
refreshed = auth.refresh_token()
print(refreshed.session.access_token)

# Read or restore the stored session
tokens = auth.get_session()   # {"access_token": ..., "refresh_token": ...}
auth.set_session(access_token="saved_access_token", refresh_token="saved_refresh_token")

# Sign out → POST /auth/v1/logout
auth.logout()

Two-Factor Authentication (TOTP)

Added in 3.10.0. When a user has TOTP two-factor authentication enabled, sign-in does not return a session. It returns a challenge instead: requires_2fa is true and a short-lived pending_token is set (SDK field names are in the example). Finish the sign-in with the verify call, using a 6-digit code from the authenticator app or one of the user's backup codes. Status, setup, enable, disable and recovery-code calls run as the signed-in user: the SDK sends your API key and the stored access token. Enabling 2FA and rotating recovery codes return new backup codes once; show them to the user and do not store them in plain text.

# 1. Enrol (user is signed in)
setup = auth.setup_totp()                 # POST /auth/v1/2fa/setup-totp
# Show setup["qr_code_url"] (a PNG data URI when available) or setup["otpauth_uri"]
enabled = auth.enable_totp("123456")      # POST /auth/v1/2fa/verify-and-enable
print(enabled["backup_codes"])            # shown once

# 2. Sign in when 2FA is on
login = auth.sign_in(email="user@example.com", password="SecurePassword123")
if login.requires_2fa:
    # POST /auth/v1/2fa/verify-login
    done = auth.verify_two_factor_login(login.pending_token, code="123456")
    # or: backup_code="one-of-your-backup-codes"
    print(done.session.access_token)

# 3. Manage
status = auth.get_two_factor_status()               # {"enabled": ..., "method": ...}
rotated = auth.regenerate_recovery_codes("123456")  # {"backup_codes": [...], "recovery_codes": [...]}
auth.disable_totp(code="123456")                    # or backup_code="..."

Settings, Anonymous Sign-in, Identity Linking and Email Change

Added in 3.10.0. Anonymous sign-in and identity linking must be enabled in the project Auth settings (anonymous_auth_enabled, manual_linking_enabled); otherwise the server rejects the call. Linking and email change run as the signed-in user.

# Public auth settings → GET /auth/v1/settings
settings = auth.get_settings()   # allow_signup, anonymous_auth_enabled, captcha, ...

# Anonymous sign-in → POST /auth/v1/sign-in-anonymous
anon = auth.sign_in_anonymous()

# Link another OAuth identity to the signed-in user → POST /auth/v1/oauth/{provider}/link
auth.link_oauth_identity("github", code="authorization_code_from_callback",
                         redirect_uri="https://yourapp.com/auth/callback")

# Change email → POST /auth/v1/update-email
auth.update_email("new@example.com")

Error Handling

Wrap SDK calls in try/except to handle API and validation errors.

from wowsql import WowSQLClient, WowSQLError

try:
    users = client.table("users").select("*").get()
    print(f"Success: {len(users['data'])} users")
except WowSQLError as e:
    if e.status_code == 401:
        print("Authentication failed — check your API key")
    elif e.status_code == 404:
        print("Table or resource not found")
    else:
        print(f"Error ({e.status_code}): {e.message}")

Storage Operations

Initialize Storage

The SDK storage client calls /api/v1/storage/projects/{project_slug}/… on your project host (buckets, files, stats); the REST routes below are served at /storage/v1/.

from wowsql import WowSQLStorage

# The SDK calls /api/v1/storage/projects/{slug}/... on your project URL
storage = WowSQLStorage(
    project_url="myproject",   # → https://myproject.wowsqlconnect.com
    api_key="wowsql_anon_your_key_here"
)

File Operations

# Upload: upload(bucket_name, file_data, path=None, file_name=None)
# path is the object path; its folder part ("uploads") is sent as the folder query parameter.
with open("path/to/document.pdf", "rb") as f:
    result = storage.upload("documents", f, path="uploads/document.pdf")
print(f"Uploaded: {result.path} ({result.size} bytes)")

# Upload a local file: upload_from_path(file_path, bucket_name="default", path=None)
storage.upload_from_path("path/to/document.pdf", "documents", path="uploads/document.pdf")

# Public URL (public buckets): /storage/v1/object/public/{bucket}/{path}
url = storage.get_public_url("documents", "uploads/document.pdf")
print(f"Public URL: {url}")

# Expiring signed URLs (get_file_url / get_presigned_url in earlier versions of this page) are not part of the SDK in 3.10.0.
# Use get_public_url for public buckets, or download() with your API key for private ones.

# Download bytes (GET .../buckets/{bucket}/files/{path}/download), or save straight to disk
data = storage.download("documents", "uploads/document.pdf")
storage.download_to_file("documents", "uploads/document.pdf", "/tmp/document.pdf")

# List files: list_files(bucket_name, prefix=None, limit=100, offset=0) → GET .../buckets/{bucket}/files
files = storage.list_files("documents", prefix="uploads/")
for file in files:
    print(f"{file.path}: {file.size_mb:.2f} MB")

# Delete file: delete_file(bucket_name, file_path) → DELETE .../buckets/{bucket}/files/{path}
storage.delete_file("documents", "uploads/document.pdf")

# Check usage (GET .../stats). get_quota() returns the same fields.
stats = storage.get_stats()
print(f"{stats.total_files} files, {stats.total_size_gb:.2f} GB used")

Direct REST Call (Storage)

import requests

PROJECT_URL = "https://myproject.wowsqlconnect.com"
API_KEY = "wowsql_anon_your_key_here"

# Upload via REST
with open("photo.png", "rb") as f:
    resp = requests.post(
        f"{PROJECT_URL}/storage/v1/object/avatars/photo.png",
        files={"file": f},
        headers={"apikey": API_KEY}
    )

# Download via REST
resp = requests.get(
    f"{PROJECT_URL}/storage/v1/object/avatars/photo.png",
    headers={"apikey": API_KEY}
)

# Public file (no auth needed if bucket is public)
# GET https://myproject.wowsqlconnect.com/storage/v1/object/public/{bucket}/{path}

Bucket Management

# Create bucket: create_bucket(name, public=False, file_size_limit=None, allowed_mime_types=None) → POST .../buckets
storage.create_bucket("avatars", public=True)

# List buckets → GET .../buckets
buckets = storage.list_buckets()

# Update → PATCH .../buckets/{name}
storage.update_bucket("avatars", public=False)

# Delete bucket → DELETE .../buckets/{name}
storage.delete_bucket("avatars")

Realtime

Subscribe to INSERT / UPDATE / DELETE, broadcast ephemeral events, and track presence. Uses the same anon or service_role key as REST. Requires websocket-client (installed with the package).

WebSocket URL: wss://<project>.wowsqlconnect.com/realtime/v1/websocket?apikey=<wowsql_anon_... or wowsql_service_...>. Browsers cannot set WebSocket headers, so auth is always the query param. Enable a table first with POST /realtime/v1/enable and body {"schema_name":"public","table_name":"messages"}. Channel send / presence do not need a Postgres trigger. Payloads are capped at 64 KiB. Missing key closes 4001; invalid key closes 4003 (do not reconnect). Multi-replica deployments need REDIS_URL.

from wowsql import WowSQLClient

client = WowSQLClient(project_url="myproject", api_key="wowsql_anon_your_key_here")

def on_change(change):
    print(change["event"], change.get("new") or change.get("old"))

sub = client.realtime.subscribe("messages", on_change, schema="public", event="*")

channel = client.realtime.channel("chat")
channel.on("broadcast", {"event": "typing"}, lambda msg: print("typing", msg["payload"]))
channel.on("presence", {}, lambda msg: print("online", channel.presence_state()))
channel.subscribe()
channel.track({"user": "alice"})
channel.send(event="typing", payload={"user": "alice"})

sub["unsubscribe"]()
channel.unsubscribe()
client.close()

Schema DDL (WowSQLSchema)

Requires service role key: Schema operations use wowsql_service_... (not the anon key). POST /api/v2/schema/tables requires a primary_key column with type UUID. Use auto_increment: True for DEFAULT gen_random_uuid(). Raw DDL via POST /api/v2/schema/execute cannot use SERIAL PRIMARY KEY. Existing tables are unchanged.

from wowsql.schema import WowSQLSchema

schema = WowSQLSchema(
    project_url="myproject",   # → https://myproject.wowsqlconnect.com
    service_key="wowsql_service_your_key_here"
)

# Create table → POST /api/v2/schema/tables
schema.create_table(
    "items",
    [
        {"name": "id", "type": "UUID", "auto_increment": True},
        {"name": "title", "type": "TEXT"},
        {"name": "price", "type": "NUMERIC"},
        {"name": "created_at", "type": "TIMESTAMPTZ", "default": "NOW()"},
    ],
    primary_key="id",
)

# Execute raw DDL → POST /api/v2/schema/execute
schema.execute("""
    ALTER TABLE items ADD COLUMN description TEXT;
    CREATE INDEX idx_items_title ON items (title);
""")

Direct REST Call (PostgREST)

You can also call the PostgREST API directly. WoWSQL uses standard PostgREST — all filtering, ordering, and pagination is done via query parameters:

import requests

PROJECT_URL = "https://myproject.wowsqlconnect.com"
API_KEY = "wowsql_anon_your_key_here"

# SELECT with filters, ordering, and pagination
resp = requests.get(f"{PROJECT_URL}/rest/v1/users", params={
    "select": "id,email,name",
    "status": "eq.active",
    "order": "created_at.desc",
    "limit": 10
}, headers={"apikey": API_KEY})
data = resp.json()

# INSERT a new row
resp = requests.post(f"{PROJECT_URL}/rest/v1/users", json={
    "email": "john@example.com",
    "name": "John Doe",
    "status": "active"
}, headers={
    "apikey": API_KEY,
    "Content-Type": "application/json",
    "Prefer": "return=representation"
})

# UPDATE a row
resp = requests.patch(f"{PROJECT_URL}/rest/v1/users", params={
    "id": "eq.123"
}, json={"name": "Updated Name"}, headers={
    "apikey": API_KEY,
    "Content-Type": "application/json",
    "Prefer": "return=representation"
})

# DELETE a row
resp = requests.delete(f"{PROJECT_URL}/rest/v1/users", params={
    "id": "eq.123"
}, headers={"apikey": API_KEY})