Official Python SDK for WowSQL — PostgreSQL BaaS with auth, storage, and subdomain routing
Project description
🚀 WowSQL Python SDK
Official Python client for WowSQL - PostgreSQL Backend-as-a-Service with built-in Storage.
Installation
pip install wowsql
Quick Start
Database Operations
from wowsql import WowSQLClient
# Initialize client
client = WowSQLClient(
project_url="https://your-project.wowsql.com",
api_key="your-api-key" # Get from dashboard
)
# Select data
response = client.table("users").select("*").limit(10).execute()
print(response.data) # [{'id': 1, 'name': 'John', ...}, ...]
# Insert data
result = client.table("users").insert({
"name": "Jane Doe",
"email": "jane@example.com",
"age": 25
}).execute()
# Update data
result = client.table("users").update({
"name": "Jane Smith"
}).eq("id", 1).execute()
# Delete data
result = client.table("users").delete().eq("id", 1).execute()
Storage Operations (NEW in 0.2.0!)
from wowsql import WowSQLStorage
# Initialize storage client
storage = WowSQLStorage(
project_url="https://your-project.wowsql.com",
api_key="your-api-key"
)
# Upload file from local path
storage.upload_from_path("local-file.pdf", "uploads/document.pdf")
# Upload file object
with open("local-file.pdf", "rb") as f:
storage.upload_file(f, "uploads/document.pdf")
# Get presigned URL for file
url_data = storage.get_file_url("uploads/document.pdf")
print(url_data["file_url"])
# List files
files = storage.list_files(prefix="uploads/")
for file in files:
print(f"{file.key}: {file.size_mb:.2f} MB")
# Delete file
storage.delete_file("uploads/document.pdf")
# Check storage quota
quota = storage.get_quota()
print(f"Used: {quota.used_gb:.2f} GB")
print(f"Available: {quota.available_gb:.2f} GB")
Project Authentication (NEW)
from wowsql import ProjectAuthClient
auth = ProjectAuthClient(
project_url="https://your-project.wowsql.com",
api_key="your-anon-key" # Use anon key for client-side, service key for server-side
)
Sign Up Users
result = auth.sign_up(
email="user@example.com",
password="SuperSecret123",
full_name="End User",
user_metadata={"referrer": "landing"}
)
print(result.user.email, result.session.access_token)
Sign In & Persist Sessions
session = auth.sign_in(
email="user@example.com",
password="SuperSecret123"
).session
auth.set_session(
access_token=session.access_token,
refresh_token=session.refresh_token
)
current_user = auth.get_user()
print(current_user.id, current_user.email_verified)
OAuth Authentication
# Step 1: Get authorization URL
oauth = auth.get_oauth_authorization_url(
provider="github",
redirect_uri="https://app.example.com/auth/callback"
)
print("Send the user to:", oauth["authorization_url"])
# Step 2: After user authorizes, exchange code for tokens
# (In your callback handler)
result = auth.exchange_oauth_callback(
provider="github",
code="authorization_code_from_callback",
redirect_uri="https://app.example.com/auth/callback"
)
print(f"Logged in as: {result.user.email}")
print(f"Access token: {result.session.access_token}")
Password Reset
# Request password reset
result = auth.forgot_password(email="user@example.com")
print(result["message"]) # "If that email exists, a password reset link has been sent"
# Reset password (after user clicks email link)
result = auth.reset_password(
token="reset_token_from_email",
new_password="NewSecurePassword123"
)
print(result["message"])
Features
Database Features
- ✅ Full CRUD operations (Create, Read, Update, Delete)
- ✅ Advanced filtering (eq, neq, gt, gte, lt, lte, like, is_null, in, not_in, between, not_between)
- ✅ Logical operators (AND, OR)
- ✅ GROUP BY and aggregate functions (COUNT, SUM, AVG, MAX, MIN)
- ✅ HAVING clause for filtering aggregated results
- ✅ Multiple ORDER BY columns
- ✅ Date/time functions in SELECT and filters
- ✅ Expressions in SELECT (e.g., "COUNT(*)", "DATE(created_at) as date")
- ✅ Pagination (limit, offset)
- ✅ Sorting (order_by)
- ✅ Get record by ID
- ✅ Table schema introspection
- ✅ Automatic rate limit handling
- ✅ Built-in error handling
- ✅ Context manager support
Storage Features (NEW!)
- ✅ PostgreSQL-native storage client
- ✅ File upload from path or file object with automatic quota validation
- ✅ Presigned URL generation for file access
- ✅ File listing with metadata
- ✅ File deletion
- ✅ Storage quota management
- ✅ Multi-region support
- ✅ Client-side limit enforcement
- ✅ Storage provisioning and management
Usage Examples
Select Queries
from wowsql import WowSQLClient
client = WowSQLClient(
project_url="https://your-project.wowsql.com",
api_key="your-api-key"
)
# Select all columns
users = client.table("users").select("*").execute()
# Select specific columns
users = client.table("users").select("id", "name", "email").execute()
# With filters
active_users = client.table("users") \
.select("*") \
.eq("status", "active") \
.gt("age", 18) \
.execute()
# With ordering
recent_users = client.table("users") \
.select("*") \
.order_by("created_at", desc=True) \
.limit(10) \
.execute()
# With pagination
page_1 = client.table("users").select("*").limit(20).offset(0).execute()
page_2 = client.table("users").select("*").limit(20).offset(20).execute()
# Pattern matching
gmail_users = client.table("users") \
.select("*") \
.like("email", "%@gmail.com") \
.execute()
# IN operator
categories = client.table("products") \
.select("*") \
.filter("category", "in", ["electronics", "books", "clothing"]) \
.execute()
# BETWEEN operator
price_range = client.table("products") \
.select("*") \
.filter("price", "between", [10, 100]) \
.execute()
# OR conditions
results = client.table("products") \
.filter("category", "eq", "electronics", "AND") \
.or("price", "gt", 1000) \
.execute()
# GROUP BY with aggregates
stats = client.table("orders") \
.select("DATE(created_at) as date", "COUNT(*) as count", "SUM(total) as revenue") \
.filter("status", "eq", "completed") \
.group_by("DATE(created_at)") \
.having("COUNT(*)", "gt", 10) \
.order_by([("date", "desc"), ("count", "asc")]) \
.limit(100) \
.execute()
# Date/time functions
recent_orders = client.table("orders") \
.select("*") \
.filter("created_at", "gte", "DATE_SUB(NOW(), INTERVAL 30 DAY)") \
.execute()
Insert Data
# Insert single row
result = client.table("users").insert({
"name": "John Doe",
"email": "john@example.com",
"age": 30
}).execute()
# Insert multiple rows
result = client.table("users").insert([
{"name": "Alice", "email": "alice@example.com"},
{"name": "Bob", "email": "bob@example.com"}
]).execute()
Update Data
# Update by ID
result = client.table("users").update({
"name": "Jane Smith",
"age": 26
}).eq("id", 1).execute()
# Update with conditions
result = client.table("users").update({
"status": "inactive"
}).lt("last_login", "2024-01-01").execute()
Delete Data
# Delete by ID
result = client.table("users").delete().eq("id", 1).execute()
# Delete with conditions
result = client.table("users").delete() \
.eq("status", "deleted") \
.lt("created_at", "2023-01-01") \
.execute()
Advanced Query Features
GROUP BY and Aggregates
GROUP BY supports both simple column names and SQL expressions with functions. All expressions are validated for security.
Basic GROUP BY
# Group by single column
result = client.table("products") \
.select("category", "COUNT(*) as count", "AVG(price) as avg_price") \
.group_by("category") \
.get()
# Group by multiple columns
result = client.table("sales") \
.select("region", "category", "SUM(amount) as total") \
.group_by("region", "category") \
.get()
GROUP BY with Date/Time Functions
# Group by date (extract date part)
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()
# Group by year
result = client.table("orders") \
.select("YEAR(created_at) as year", "COUNT(*) as orders") \
.group_by("YEAR(created_at)") \
.get()
# Group by year and month
result = client.table("orders") \
.select("YEAR(created_at) as year", "MONTH(created_at) as month", "SUM(total) as revenue") \
.group_by("YEAR(created_at)", "MONTH(created_at)") \
.order_by("year", "desc") \
.order_by("month", "desc") \
.get()
# Group by week
result = client.table("orders") \
.select("WEEK(created_at) as week", "COUNT(*) as orders") \
.group_by("WEEK(created_at)") \
.get()
# Group by quarter
result = client.table("orders") \
.select("QUARTER(created_at) as quarter", "SUM(total) as revenue") \
.group_by("QUARTER(created_at)") \
.get()
GROUP BY with String Functions
# Group by first letter of name
result = client.table("users") \
.select("LEFT(name, 1) as first_letter", "COUNT(*) as count") \
.group_by("LEFT(name, 1)") \
.get()
# Group by uppercase category
result = client.table("products") \
.select("UPPER(category) as category_upper", "COUNT(*) as count") \
.group_by("UPPER(category)") \
.get()
GROUP BY with Mathematical Functions
# Group by rounded price ranges
result = client.table("products") \
.select("ROUND(price, -1) as price_range", "COUNT(*) as count") \
.group_by("ROUND(price, -1)") \
.get()
# Group by price tier (using FLOOR)
result = client.table("products") \
.select("FLOOR(price / 10) * 10 as price_tier", "COUNT(*) as count") \
.group_by("FLOOR(price / 10) * 10") \
.get()
Supported Functions in GROUP BY
The following functions are allowed in GROUP BY expressions:
Date/Time Functions:
DATE(),YEAR(),MONTH(),DAY(),DAYOFMONTH(),DAYOFWEEK(),DAYOFYEAR()WEEK(),QUARTER(),HOUR(),MINUTE(),SECOND()DATE_FORMAT(),TIME(),DATE_ADD(),DATE_SUB()DATEDIFF(),TIMEDIFF(),TIMESTAMPDIFF()NOW(),CURRENT_TIMESTAMP(),CURDATE(),CURRENT_DATE()CURTIME(),CURRENT_TIME(),UNIX_TIMESTAMP()
String Functions:
CONCAT(),CONCAT_WS(),SUBSTRING(),SUBSTR(),LEFT(),RIGHT()LENGTH(),CHAR_LENGTH(),UPPER(),LOWER(),TRIM(),LTRIM(),RTRIM()REPLACE(),LOCATE(),POSITION()
Mathematical Functions:
ABS(),ROUND(),CEIL(),CEILING(),FLOOR(),POW(),POWER(),SQRT(),MOD(),RAND()
Note: All expressions are validated for security. Only whitelisted functions are allowed.
HAVING Clause
HAVING is used to filter aggregated results after GROUP BY. It supports aggregate functions and comparison operators.
Basic HAVING
# Filter aggregated results
result = client.table("products") \
.select("category", "COUNT(*) as count") \
.group_by("category") \
.having("COUNT(*)", "gt", 10) \
.get()
# Multiple HAVING conditions (AND logic)
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()
HAVING with Aggregate Functions
# Filter by average
result = client.table("products") \
.select("category", "AVG(price) as avg_price", "COUNT(*) as count") \
.group_by("category") \
.having("AVG(price)", "gt", 100) \
.having("COUNT(*)", "gte", 5) \
.get()
# Filter by sum
result = client.table("orders") \
.select("customer_id", "SUM(total) as total_spent") \
.group_by("customer_id") \
.having("SUM(total)", "gt", 1000) \
.get()
# Filter by max/min
result = client.table("products") \
.select("category", "MAX(price) as max_price", "MIN(price) as min_price") \
.group_by("category") \
.having("MAX(price)", "gt", 500) \
.get()
Supported HAVING Operators
eq- Equal toneq- Not equal togt- Greater thangte- Greater than or equal tolt- Less thanlte- Less than or equal to
Supported Aggregate Functions in HAVING
COUNT(*)orCOUNT(column)- Count of rowsSUM(column)- Sum of valuesAVG(column)- Average of valuesMAX(column)- Maximum valueMIN(column)- Minimum valueGROUP_CONCAT(column)- Concatenated valuesSTDDEV(column),STDDEV_POP(column),STDDEV_SAMP(column)- Standard deviationVARIANCE(column),VAR_POP(column),VAR_SAMP(column)- Variance
Multiple ORDER BY
# Order by multiple columns
result = client.table("products") \
.select("*") \
.order_by([("category", "asc"), ("price", "desc"), ("created_at", "desc")]) \
.get()
# Using OrderByItem objects
from wowsql.types import OrderByItem
result = client.table("products") \
.order_by([
OrderByItem(column="category", direction="asc"),
OrderByItem(column="price", direction="desc")
]) \
.get()
Date/Time Functions
# Filter by date range using functions
result = client.table("orders") \
.select("*") \
.filter("created_at", "gte", "DATE_SUB(NOW(), INTERVAL 7 DAY)") \
.get()
# Group by date
result = client.table("orders") \
.select("DATE(created_at) as date", "COUNT(*) as count") \
.group_by("DATE(created_at)") \
.get()
# Filter by year/month
result = client.table("orders") \
.select("*") \
.filter("YEAR(created_at)", "eq", 2024) \
.filter("MONTH(created_at)", "eq", 1) \
.get()
Filter Operators
# Equal
.eq("status", "active")
# Not equal
.neq("status", "deleted")
# Greater than
.gt("age", 18)
# Greater than or equal
.gte("age", 18)
# Less than
.lt("age", 65)
# Less than or equal
.lte("age", 65)
# Pattern matching (SQL LIKE)
.like("email", "%@gmail.com")
# Is null
.is_null("deleted_at")
# IN operator (value must be a list)
.filter("category", "in", ["electronics", "books", "clothing"])
# NOT IN operator
.filter("status", "not_in", ["deleted", "archived"])
# BETWEEN operator (value must be a list/tuple of 2 values)
.filter("price", "between", [10, 100])
# NOT BETWEEN operator
.filter("age", "not_between", [18, 65])
# IS NOT NULL
.is_not_null("email")
# OR logical operator
.filter("category", "eq", "electronics", "AND")
.filter("price", "gt", 1000, "OR") # OR condition
Storage Operations
from wowsql import WowSQLStorage, StorageLimitExceededError
storage = WowSQLStorage(
project_url="https://your-project.wowsql.com",
api_key="your-api-key"
)
# Upload file from local path
storage.upload_from_path(
"document.pdf",
"uploads/2024/document.pdf"
)
# Upload file object
with open("image.jpg", "rb") as f:
storage.upload_file(f, "images/photo.jpg")
# List files with prefix
files = storage.list_files(prefix="uploads/2024/", max_keys=100)
for file in files:
print(f"{file.key}: {file.size_mb:.2f} MB")
# Get presigned URL (valid for 1 hour by default)
url_data = storage.get_file_url("uploads/document.pdf", expires_in=3600)
print(url_data["file_url"]) # https://s3.amazonaws.com/...
# Get presigned URL for upload
upload_url = storage.get_presigned_url(
"new-file.pdf",
expires_in=3600,
operation="put_object"
)
# Delete file
storage.delete_file("uploads/old-file.pdf")
# Check quota before upload
quota = storage.get_quota()
print(f"Available: {quota.available_gb:.2f} GB")
# Get storage information
info = storage.get_storage_info()
print(f"Bucket: {info['bucket_name']}")
print(f"Total objects: {info['total_objects']}")
# Handle storage limit errors
try:
with open("huge-file.zip", "rb") as f:
storage.upload_file(f, "uploads/huge-file.zip")
except StorageLimitExceededError as e:
print(f"Storage limit exceeded: {e}")
print("Please upgrade your plan or delete old files")
Context Manager
# Automatically closes connection
with WowSQLClient(project_url="...", api_key="...") as client:
users = client.table("users").select("*").execute()
print(users.data)
# Connection closed here
# Works with storage too
with WowSQLStorage(project_url="...", api_key="...") as storage:
files = storage.list_files()
print(f"Total files: {len(files)}")
Error Handling
from wowsql import (
WowSQLClient,
WowSQLError,
StorageError,
StorageLimitExceededError
)
client = WowSQLClient(project_url="...", api_key="...")
try:
users = client.table("users").select("*").execute()
except WowSQLError as e:
print(f"Database error: {e}")
storage = WowSQLStorage(project_url="...", api_key="...")
try:
storage.upload("file.pdf", "uploads/file.pdf")
except StorageLimitExceededError as e:
print(f"Storage full: {e}")
except StorageError as e:
print(f"Storage error: {e}")
Utility Methods
# List all tables
tables = client.list_tables()
print(tables) # ['users', 'posts', 'comments']
# Get table schema
schema = client.get_table_schema("users")
print(schema) # {'table': 'users', 'columns': [...], 'primary_key': 'id'}
# Get record by ID
user = client.table("users").get_by_id(1)
print(user) # {'id': 1, 'name': 'John', ...}
Response Object
All database queries return a response object:
response = client.table("users").select("*").limit(10).execute()
# Access data
print(response.data) # [{'id': 1, ...}, {'id': 2, ...}]
# Access count
print(response.count) # 10
# Check for errors
if response.error:
print(response.error)
Configuration
Timeouts
# Custom timeout (default: 30 seconds)
client = WowSQLClient(
project_url="...",
api_key="...",
timeout=60 # 60 seconds
)
# Storage timeout (default: 60 seconds for large files)
storage = WowSQLStorage(
project_url="...",
api_key="...",
timeout=120 # 2 minutes
)
Auto Quota Check
# Disable automatic quota checking before uploads
storage = WowSQLStorage(
project_url="...",
api_key="...",
auto_check_quota=False
)
# Manually check quota
quota = storage.get_quota()
file_size_gb = file_size / (1024**3) # Convert bytes to GB
if quota.available_gb > file_size_gb:
with open("file.pdf", "rb") as f:
storage.upload_file(f, "uploads/file.pdf", check_quota=False)
API Keys
WOWSQL uses different API keys for different operations. Understanding which key to use is crucial for proper authentication.
Key Types Overview
🔑 Unified Authentication
✨ One Project = One Set of Keys for ALL Operations
WowSQL uses unified authentication - the same API keys work for both database operations AND authentication operations.
| Operation Type | Recommended Key | Alternative Key | Used By |
|---|---|---|---|
| Database Operations (CRUD) | Service Role Key (wowsql_service_...) |
Anonymous Key (wowsql_anon_...) |
WowSQLClient |
| Authentication Operations (OAuth, sign-in) | Anonymous Key (wowsql_anon_...) |
Service Role Key (wowsql_service_...) |
ProjectAuthClient |
Where to Find Your Keys
All keys are found in: WOWSQL Dashboard → Settings → API Keys or Authentication → PROJECT KEYS
-
Anonymous Key (
wowsql_anon_...) ✨ Unified Key- Location: "Anonymous Key (Public)"
- Used for:
- ✅ Client-side auth operations (signup, login, OAuth)
- ✅ Public/client-side database operations with limited permissions
- Safe to expose in frontend code (browser, mobile apps)
-
Service Role Key (
wowsql_service_...) ✨ Unified Key- Location: "Service Role Key (keep secret)"
- Used for:
- ✅ Server-side auth operations (admin, full access)
- ✅ Server-side database operations (full access, bypass RLS)
- NEVER expose in frontend code - server-side only!
Database Operations
Use Service Role Key or Anonymous Key for database operations:
from wowsql import WowSQLClient
# Using Service Role Key (recommended for server-side, full access)
client = WowSQLClient(
project_url="https://your-project.wowsql.com",
api_key="wowsql_service_your-service-key-here" # Service Role Key
)
# Using Anonymous Key (for public/client-side access with limited permissions)
client = WowSQLClient(
project_url="https://your-project.wowsql.com",
api_key="wowsql_anon_your-anon-key-here" # Anonymous Key
)
# Query data
users = client.table("users").get()
Authentication Operations
✨ UNIFIED AUTHENTICATION: Use the same keys as database operations!
from wowsql import ProjectAuthClient
# Using Anonymous Key (recommended for client-side auth operations)
auth = ProjectAuthClient(
project_url="https://your-project.wowsql.com",
api_key="wowsql_anon_your-anon-key-here" # Same key as database operations!
)
# Using Service Role Key (for server-side auth operations)
auth = ProjectAuthClient(
project_url="https://your-project.wowsql.com",
api_key="wowsql_service_your-service-key-here" # Same key as database operations!
)
# OAuth authentication
oauth_url = auth.get_oauth_authorization_url(
provider="github",
redirect_uri="https://app.example.com/auth/callback"
)
Note: The public_api_key parameter is deprecated but still works for backward compatibility. Use api_key instead.
Environment Variables
Best practice: Use environment variables for API keys:
import os
from wowsql import WowSQLClient, ProjectAuthClient
# UNIFIED AUTHENTICATION: Same keys for both operations!
# Database operations - Service Role Key
db_client = WowSQLClient(
project_url=os.getenv("WOWSQL_PROJECT_URL"),
api_key=os.getenv("WOWSQL_SERVICE_ROLE_KEY") # or WOWSQL_ANON_KEY
)
# Authentication operations - Use the SAME key!
auth_client = ProjectAuthClient(
project_url=os.getenv("WOWSQL_PROJECT_URL"),
api_key=os.getenv("WOWSQL_ANON_KEY") # Same key for client-side auth
# Or use WOWSQL_SERVICE_ROLE_KEY for server-side auth
)
Key Usage Summary
✨ UNIFIED AUTHENTICATION:
WowSQLClient→ Uses Service Role Key or Anonymous Key for database operationsProjectAuthClient→ Uses Anonymous Key (client-side) or Service Role Key (server-side) for authentication operations- Same keys work for both database AND authentication operations! 🎉
- Anonymous Key (
wowsql_anon_...) → Client-side operations (auth + database) - Service Role Key (
wowsql_service_...) → Server-side operations (auth + database) - Anonymous Key is optional and provides limited permissions for public database access
Security Best Practices
- Never expose Service Role Key in client-side code or public repositories
- Use Public API Key for client-side authentication flows
- Use Anonymous Key for public database access with limited permissions
- Store keys in environment variables, never hardcode them
- Rotate keys regularly if compromised
Troubleshooting
Error: "Invalid API key for project"
- Ensure you're using the correct key type for the operation
- Database operations require Service Role Key or Anonymous Key
- Authentication operations require Anonymous Key (client-side) or Service Role Key (server-side)
- Verify the key is copied correctly (no extra spaces)
Error: "Authentication failed"
- Check that you're using the correct key: Anonymous Key for client-side, Service Role Key for server-side
- Verify the project URL matches your dashboard
- Ensure the key hasn't been revoked or expired
Examples
Blog Application
from wowsql import WowSQLClient
client = WowSQLClient(project_url="...", api_key="...")
# Create a new post
post = client.table("posts").insert({
"title": "Hello World",
"content": "My first blog post",
"author_id": 1,
"published": True
}).execute()
# Get published posts
posts = client.table("posts") \
.select("id", "title", "content", "created_at") \
.eq("published", True) \
.order_by("created_at", desc=True) \
.limit(10) \
.execute()
# Get post by ID
post = client.table("posts").get_by_id(1)
# Get post with comments
post = client.table("posts").select("*").eq("id", 1).execute()
comments = client.table("comments").select("*").eq("post_id", 1).execute()
File Upload Application
from wowsql import WowSQLClient, WowSQLStorage
client = WowSQLClient(project_url="...", api_key="...")
storage = WowSQLStorage(project_url="...", api_key="...")
# Upload user avatar
user_id = 123
avatar_path = f"avatars/{user_id}.jpg"
storage.upload_from_path("avatar.jpg", avatar_path)
# Save avatar URL in database
url_data = storage.get_file_url(avatar_path)
client.table("users").update(user_id, {
"avatar_url": url_data["file_url"]
})
# List user's files
user_files = storage.list_files(prefix=f"users/{user_id}/")
print(f"User has {len(user_files)} files")
Requirements
- Python 3.8+
- requests>=2.31.0
Development
# Clone repository
git clone https://github.com/wowsql/wowsql.git
cd wowsql/sdk/python
# Install in development mode
pip install -e ".[dev]"
# Run tests
pytest
# Run examples
python examples/basic_usage.py
python examples/storage_usage.py
Contributing
Contributions are welcome! Please open an issue or submit a pull request.
License
MIT License - see LICENSE file for details.
Links
- 📚 Documentation
- 🌐 Website
- 💬 Discord
- 🐛 Issues
🔧 Schema Management (NEW in v0.5.0!)
Programmatically manage your database schema with the WowSQLSchema client.
⚠️ IMPORTANT: Schema operations require a Service Role Key (
service_*). Anonymous keys will return a 403 Forbidden error.
Quick Start
from wowsql import WowSQLSchema
# Initialize schema client with SERVICE ROLE KEY
schema = WowSQLSchema(
project_url="https://your-project.wowsql.com",
service_key="service_xyz789..." # ⚠️ Backend only! Never expose!
)
Create Table
# Create a new table
schema.create_table(
table_name="products",
columns=[
{"name": "id", "type": "INT", "auto_increment": True},
{"name": "name", "type": "VARCHAR(255)", "not_null": True},
{"name": "price", "type": "DECIMAL(10,2)", "not_null": True},
{"name": "category", "type": "VARCHAR(100)"},
{"name": "created_at", "type": "TIMESTAMP", "default": "CURRENT_TIMESTAMP"}
],
primary_key="id",
indexes=[
{"name": "idx_category", "columns": ["category"]},
{"name": "idx_price", "columns": ["price"]}
]
)
print("Table 'products' created successfully!")
Alter Table
# Add a new column
schema.alter_table(
table_name="products",
add_columns=[
{"name": "stock_quantity", "type": "INT", "default": "0"}
]
)
# Modify an existing column
schema.alter_table(
table_name="products",
modify_columns=[
{"name": "price", "type": "DECIMAL(12,2)"} # Increase precision
]
)
# Drop a column
schema.alter_table(
table_name="products",
drop_columns=["category"]
)
# Rename a column
schema.alter_table(
table_name="products",
rename_columns=[
{"old_name": "name", "new_name": "product_name"}
]
)
Drop Table
# Drop a table
schema.drop_table("old_table")
# Drop with CASCADE (removes dependent objects)
schema.drop_table("products", cascade=True)
Execute Raw SQL
# Execute custom schema SQL
schema.execute_sql("""
CREATE INDEX idx_product_name
ON products(product_name);
""")
# Add a foreign key constraint
schema.execute_sql("""
ALTER TABLE orders
ADD CONSTRAINT fk_product
FOREIGN KEY (product_id)
REFERENCES products(id);
""")
Security & Best Practices
✅ DO:
- Use service role keys only in backend/server code
- Store service keys in environment variables
- Use anonymous keys for client-side data operations
- Test schema changes in development first
❌ DON'T:
- Never expose service role keys in frontend code
- Never commit service keys to version control
- Don't use anonymous keys for schema operations (will fail)
Example: Backend Migration Script
import os
from wowsql import WowSQLSchema
def run_migration():
schema = WowSQLSchema(
project_url=os.getenv("WOWSQL_PROJECT_URL"),
service_key=os.getenv("WOWSQL_SERVICE_KEY") # From env var
)
# Create users table
schema.create_table(
table_name="users",
columns=[
{"name": "id", "type": "INT", "auto_increment": True},
{"name": "email", "type": "VARCHAR(255)", "unique": True, "not_null": True},
{"name": "name", "type": "VARCHAR(255)", "not_null": True},
{"name": "created_at", "type": "TIMESTAMP", "default": "CURRENT_TIMESTAMP"}
],
primary_key="id",
indexes=[{"name": "idx_email", "columns": ["email"]}]
)
print("Migration completed!")
if __name__ == "__main__":
run_migration()
Error Handling
from wowsql import WowSQLSchema, SchemaPermissionError
try:
schema = WowSQLSchema(
project_url="https://your-project.wowsql.com",
service_key="wowsql_service_..."
)
schema.create_table("test", [{"name": "id", "type": "SERIAL"}])
except SchemaPermissionError as e:
print(f"Permission denied: {e}")
print("Make sure you're using a SERVICE ROLE KEY, not an anonymous key!")
except Exception as e:
print(f"Error: {e}")
Support
- Email: support@wowsql.com
- Discord: https://discord.gg/wowsql
- Documentation: https://wowsql.com/docs
Made with ❤️ by the WowSQL Team
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file wowsql-3.0.0.tar.gz.
File metadata
- Download URL: wowsql-3.0.0.tar.gz
- Upload date:
- Size: 46.0 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.9.13
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
3d5c57d5d4fc6185342afd952a667245b0e42b4f5488b263f95ffde743117ec0
|
|
| MD5 |
a4cfb334f0ac16c515232dbe9b1a5040
|
|
| BLAKE2b-256 |
e558c9d4a793844472676ea5c52960e3b902e09bf7263f40862e13e9dfcd88d0
|
File details
Details for the file wowsql-3.0.0-py3-none-any.whl.
File metadata
- Download URL: wowsql-3.0.0-py3-none-any.whl
- Upload date:
- Size: 33.5 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.9.13
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
0b2cbb5f068dbb25eba9e393b16dbd3ac5e1b50d83c161ce29912810cf5e6674
|
|
| MD5 |
19c5e4eb8a2fcaf545e12916010d72e2
|
|
| BLAKE2b-256 |
f23fc2ae64d80144d2478920176aacf5046f5f633feb6fb13b30451831de42c2
|