WowSQL Python SDK

Official Python client for WowSQL. All data operations communicate directly with PostgREST over HTTPS. No legacy API layers.

Requirements

• Python 3.8 or later
• requests and websocket-client

Installation

pip install wowsql

Quick Start

from wowsql import WowSQLClient

client = WowSQLClient(
    project_url="myproject",        # slug, domain, or full URL
    api_key="wowsql_anon_...",      # anon or service_role key
)

result = client.table("users").get()
print(result["data"])

Configuration

project_url formats

# Slug only
WowSQLClient(project_url="myproject", api_key="...")

# Full domain
WowSQLClient(project_url="myproject.wowsqlconnect.com", api_key="...")

# Full URL
WowSQLClient(project_url="https://myproject.wowsqlconnect.com", api_key="...")

API Keys

Database Operations

Querying Records

# Get all records
result = client.table("products").get()
data = result["data"]  # list of records
total = result["total"]  # total count

# Select specific columns
result = client.table("products").select("id", "name", "price").get()

# Filter with equality
result = client.table("products").eq("category", "electronics").get()

# Multiple filters (all ANDed by default)
result = (
    client.table("products")
    .gte("price", 100)
    .lte("price", 500)
    .eq("in_stock", True)
    .get()
)

# OR filter
result = (
    client.table("products")
    .eq("status", "active")
    .or_("status", "eq", "featured")
    .get()
)

Filter Operators

Ordering and Pagination

# Order by column
result = client.table("products").order_by("price", "desc").get()

# Multiple column ordering
result = client.table("products").order_by([
    ("category", "asc"),
    ("price", "desc"),
]).get()

# Limit and offset
result = client.table("products").limit(20).offset(40).get()

# Paginate
result = client.table("products").paginate(page=2, per_page=20)
# result["data"], result["page"], result["per_page"]
# result["total"], result["total_pages"]

Single Record Fetch

# By primary key
user = client.table("users").get_by_id("uuid-here")

# First matching record
user = client.table("users").eq("email", "user@example.com").first()

# Exactly one record — raises WowSQLError if 0 or more than 1
user = client.table("users").eq("email", "user@example.com").single()

Creating Records

# Single record
response = client.table("users").create({
    "email": "new@example.com",
    "name": "New User",
})
new_id = response["id"]

# Alias
response = client.table("users").insert({"email": "new@example.com"})

# Bulk insert
responses = client.table("users").bulk_insert([
    {"email": "a@example.com", "name": "Alice"},
    {"email": "b@example.com", "name": "Bob"},
])

Updating Records

# Update by primary key
response = client.table("users").update("user-uuid", {"name": "Updated Name"})
print(response["affected_rows"])

Deleting Records

# Delete by primary key
response = client.table("users").delete("user-uuid")
print(response["affected_rows"])

Upsert

response = client.table("users").upsert(
    {"id": "existing-uuid", "name": "Updated"},
    on_conflict="id",  # default: "id"
)

Aggregates

# Count
total = client.table("orders").eq("status", "active").count()

# Sum
revenue = client.table("orders").eq("status", "paid").sum("amount")

# Average
avg_price = client.table("products").eq("category", "electronics").avg("price")

Group By

result = (
    client.table("orders")
    .select("status", "COUNT(*) as count")
    .group_by("status")
    .get()
)

Context Manager

with WowSQLClient(project_url="myproject", api_key="...") as client:
    result = client.table("users").get()

Authentication

from wowsql.auth import ProjectAuthClient

auth = ProjectAuthClient(
    project_url="myproject",
    api_key="wowsql_anon_...",
)

Email / Password

# Sign up
response = auth.sign_up(email="user@example.com", password="secure-password")
session = response.session
user = response.user

# Sign in
response = auth.sign_in(email="user@example.com", password="secure-password")

# Get current user
user = auth.get_user(access_token=session.access_token)

# Sign out
auth.logout(access_token=session.access_token)

CAPTCHA (Turnstile)

Optional. Only required when Attack Protection → Turnstile is enabled. Existing calls without captcha_token are unchanged.

# 1. GET /auth/v1/settings → captcha.site_key
# 2. Render Cloudflare Turnstile with that site key
# 3. Pass the token:
auth.sign_up(email="user@example.com", password="secure-password", captcha_token=turnstile_token)
auth.sign_in(email="user@example.com", password="secure-password", 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)

OAuth

# Step 1 — get the redirect URL
result = auth.get_oauth_authorization_url(
    provider="google",
    redirect_uri="https://myapp.com/auth/callback",
)
# Redirect user to: result["authorization_url"]

# Step 2 — after OAuth callback, exchange code for tokens
response = auth.exchange_oauth_callback(
    provider="google",
    code=request.args["code"],
)
session = response.session
user = response.user

Token Management

# Refresh tokens
response = auth.refresh_token(refresh_token=session.refresh_token)

# Password reset
auth.forgot_password(email="user@example.com")
auth.reset_password(token="reset-token", new_password="new-password")

OTP and Magic Links

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

# Phone OTP (SMS) — provide phone instead of email
auth.send_otp(phone="+15551234567")
auth.verify_otp(phone="+15551234567", otp="123456")

# Magic link (email only)
auth.send_magic_link(email="user@example.com")

File Storage

from wowsql.storage import WowSQLStorage

storage = WowSQLStorage(
    project_url="myproject",
    api_key="wowsql_anon_...",
)

# Create bucket
storage.create_bucket("avatars", public=True)

# Upload file
with open("photo.jpg", "rb") as f:
    result = storage.upload("avatars", f, path="users/profile.jpg")

# Get public URL
url = storage.get_public_url("avatars", "users/profile.jpg")

# List files
files = storage.list_files("avatars", prefix="users/")

# Download
data = storage.download("avatars", "users/profile.jpg")

# Delete
storage.delete_file("avatars", "users/profile.jpg")

Schema Management

Requires a service role key.

from wowsql.schema import WowSQLSchema

schema = WowSQLSchema(
    project_url="myproject",
    service_key="wowsql_service_...",
)

# Create table
schema.create_table("products", [
    {"name": "id",         "type": "UUID",         "auto_increment": True},
    {"name": "name",       "type": "VARCHAR(255)",  "nullable": False},
    {"name": "price",      "type": "DECIMAL(10,2)", "nullable": False},
    {"name": "metadata",   "type": "JSONB",         "default": "'{}'"},
    {"name": "created_at", "type": "TIMESTAMPTZ",   "default": "CURRENT_TIMESTAMP"},
], primary_key="id", indexes=["name"])

# Add column
schema.add_column("products", "sku", "VARCHAR(100)")

# Drop column
schema.drop_column("products", "old_column")

# List tables
tables = schema.list_tables()

# Get table schema
info = schema.get_table_schema("products")

# Execute raw SQL (DDL only)
schema.execute_sql("CREATE INDEX ON products(name)")

# Drop table
schema.drop_table("products")

Realtime

Subscribe to INSERT / UPDATE / DELETE, broadcast ephemeral events, and track presence. Uses the same anon or service_role key as REST.

The SDK connects to:

wss://<project>.wowsqlconnect.com/realtime/v1/websocket?apikey=<wowsql_anon_... or wowsql_service_...>

Enable the table first (POST /realtime/v1/enable). Channel send / presence do not need a Postgres trigger. Payloads are capped at 64 KiB. Missing key closes 4001; invalid key closes 4003. Requires websocket-client (installed with the package).

from wowsql import WowSQLClient

client = WowSQLClient(project_url="myproject", api_key="wowsql_anon_...")

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"})

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

Error Handling

from wowsql import WowSQLError

try:
    result = client.table("orders").eq("id", "uuid").get()
except WowSQLError as err:
    print(err)             # Human-readable message
    print(err.status_code) # HTTP status code
    print(err.response)    # Raw PostgREST error dict

Architecture

All requests go directly to the PostgREST endpoint (/rest/v1). SDK filter expressions are translated into PostgREST query parameters. Pagination counts are read from the Content-Range response header.

License

MIT