wowsql 3.1.0

Official Python SDK for WowSQL — PostgreSQL BaaS with auth, storage, and subdomain routing

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 library

Installation

pip install wowsql-sdk

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

Parameter	Type	Default	Description
project_url	str	required	Project slug, domain, or full URL
api_key	str	required	Anonymous or service role API key
base_domain	str	wowsqlconnect.com	Base domain when project_url is a slug
secure	bool	True	Use HTTPS
timeout	int	30	Request timeout in seconds
verify_ssl	bool	True	Verify SSL certificates

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

Key Type	Prefix	Usage
Anonymous	wowsql_anon_...	Client-side, public access with RLS
Service Role	wowsql_service_...	Server-side, full access, bypasses RLS

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

Operator	Method	Description
Equal	.eq(col, val)	column = value
Not equal	.neq(col, val)	column != value
Greater than	.gt(col, val)	column > value
Greater or equal	.gte(col, val)	column >= value
Less than	.lt(col, val)	column < value
Less or equal	.lte(col, val)	column <= value
LIKE	.like(col, pattern)	case-sensitive pattern match
ILIKE	.ilike(col, pattern)	case-insensitive pattern match
Is null	.is_null(col)	column IS NULL
Is not null	.is_not_null(col)	column IS NOT NULL
In list	.in_(col, [vals])	column IN (...)
Not in list	.not_in(col, [vals])	column NOT IN (...)
Between	.between(col, lo, hi)	lo <= column <= hi

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", "[e-mail]").first()

# Exactly one record — raises WowSQLError if 0 or more than 1
user = client.table("users").eq("email", "[e-mail]").single()

Creating Records

# Single record
response = client.table("users").create({
    "email": "[e-mail]",
    "name": "New User",
})
new_id = response["id"]

# Alias
response = client.table("users").insert({"email": "[e-mail]"})

# Bulk insert
responses = client.table("users").bulk_insert([
    {"email": "[e-mail]", "name": "Alice"},
    {"email": "[e-mail]", "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="[e-mail]", password="secure-password")
session = response.session
user = response.user

# Sign in
response = auth.sign_in(email="[e-mail]", password="secure-password")

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

# Sign out
auth.logout(access_token=session.access_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="[e-mail]")
auth.reset_password(token="reset-token", new_password="new-password")

OTP and Magic Links

# OTP login
auth.send_otp(email="[e-mail]")
response = auth.verify_otp(email="[e-mail]", otp="123456")

# Magic link
auth.send_magic_link(email="[e-mail]")

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

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.

Operation	HTTP Method	PostgREST Path
GET	GET	/{table}?col=op.val
CREATE	POST	/{table} with Prefer: return=representation
UPDATE	PATCH	/{table}?id=eq.{id}
DELETE	DELETE	/{table}?id=eq.{id}
UPSERT	POST	/{table} with Prefer: resolution=merge-duplicates
COUNT	GET	/{table}?limit=0 with Prefer: count=exact

License

MIT