ZenDBX Python SDK

The official Python SDK for ZenDBX - Your Backend as a Service platform.

Features

• 🔥 Unified Client - Single entry point for all ZenDBX features
• 🚀 Async First - Built on asyncio for high performance
• 🔍 Query Builder - Fluent, type-safe database queries
• 🔐 Authentication - Complete auth system (signup, login, OAuth)
• 📦 Storage - File upload/download with streaming support
• ⚡ Real-time - WebSocket subscriptions (coming soon)
• 🛡️ Type Safe - Full type hints and IDE autocomplete
• 📝 Well Documented - Comprehensive docs and examples

Installation

pip install zendbx

Quick Start

from zendbx import ZenDBX
import asyncio

# Initialize client
client = ZenDBX(
    project_url="https://api.zendbx.in/p/my-project",
    anon_key="your-anon-key"
)

async def main():
    # Sign up
    user = await client.auth.sign_up(
        email="user@example.com",
        password="secure-password"
    )
    
    # Insert data
    todo = await client.table("todos").insert({
        "title": "Ship ZenDBX",
        "completed": False
    }).execute()
    
    # Query data
    todos = await client.table("todos")\
        .select("*")\
        .where(completed=False)\
        .order_by("-created_at")\
        .limit(10)\
        .execute()
    
    print(todos)

asyncio.run(main())

Authentication

# Sign up
user = await client.auth.sign_up(
    email="user@example.com",
    password="password123"
)

# Log in
session = await client.auth.sign_in(
    email="user@example.com",
    password="password123"
)

# Get current user
user = await client.auth.get_user()

# Log out
await client.auth.sign_out()

# OAuth (Google, GitHub)
url = client.auth.get_oauth_url(provider="google")
# After callback:
session = await client.auth.exchange_code(code)

Database Operations

Query Builder

# Select
users = await client.table("users")\
    .select("id", "email", "created_at")\
    .where(is_active=True)\
    .order_by("-created_at")\
    .limit(20)\
    .execute()

# Insert
user = await client.table("users").insert({
    "email": "new@example.com",
    "full_name": "John Doe"
}).execute()

# Update
await client.table("users")\
    .update({"is_active": False})\
    .where(id=user_id)\
    .execute()

# Delete
await client.table("users")\
    .delete()\
    .where(id=user_id)\
    .execute()

# Count
count = await client.table("users").count().execute()

# Upsert
await client.table("settings").upsert({
    "user_id": 123,
    "theme": "dark"
}, on_conflict="user_id").execute()

Parameterized Queries

# Safe parameterized queries
result = await client.query(
    "SELECT * FROM users WHERE email = $1 AND is_active = $2",
    "user@example.com",
    True
)

Storage

# Upload file
with open("photo.jpg", "rb") as f:
    file = await client.storage.upload(
        bucket="avatars",
        path="user-123/avatar.jpg",
        file=f
    )

# Download file
data = await client.storage.download(
    bucket="avatars",
    path="user-123/avatar.jpg"
)

# Get public URL
url = client.storage.get_public_url(
    bucket="avatars",
    path="user-123/avatar.jpg"
)

# Delete file
await client.storage.delete(
    bucket="avatars",
    path="user-123/avatar.jpg"
)

# List files
files = await client.storage.list(bucket="avatars", prefix="user-123/")

Error Handling

from zendbx.exceptions import (
    ZenDBXAuthenticationError,
    ZenDBXPermissionError,
    ZenDBXValidationError,
    ZenDBXStorageError
)

try:
    user = await client.auth.sign_in(email="...", password="...")
except ZenDBXAuthenticationError as e:
    print(f"Auth failed: {e.message}")
    print(f"Status: {e.status_code}")
except ZenDBXValidationError as e:
    print(f"Validation error: {e.message}")

Advanced Usage

Custom Configuration

client = ZenDBX(
    project_url="https://api.zendbx.in/p/my-project",
    anon_key="your-anon-key",
    service_key="your-service-key",  # Optional, for server-side
    options={
        "timeout": 30,
        "max_retries": 3,
        "auto_refresh_token": True
    }
)

Connection Reuse

# The client automatically reuses connections
# across requests for optimal performance

async with ZenDBX(...) as client:
    # All operations use the same connection pool
    await client.table("users").select().execute()
    await client.table("posts").select().execute()

Documentation

• API Reference
• Authentication Guide
• Database Guide
• Storage Guide
• Examples

Requirements

• Python 3.8+
• aiohttp
• pydantic

Contributing

Contributions are welcome! Please read our Contributing Guide.

License

MIT License - see LICENSE file for details.

Support

• 📧 Email: support@zendbx.in
• 💬 Discord: Join our community
• 📚 Docs: docs.zendbx.in
• 🐛 Issues: GitHub Issues