Python SDK Reference

The official Python SDK for ZenDBX. Async-first architecture with full support for authentication, database operations, and storage.

zendbxv1.0.3Python 3.8+View on PyPI →

Installation

bash

pip install zendbx

Or install a specific version:

bash

pip install zendbx==1.0.3

Requirements

• Python 3.8 or higher
• aiohttp (async HTTP client)
• pydantic (data validation)

Initialize Client

ZenDBX Python SDK is async-first. Always use async/await and close the client when done.

Context Manager (Recommended)

python

import asyncio
from zendbx import ZenDBX

async def main():
    # Using context manager (recommended)
    async with ZenDBX(
        project_url="https://api.zendbx.in/p/your-project",
        anon_key="your-anon-key"
    ) as client:
        # Your code here
        user = await client.auth.get_user()
        print(user)
    # Client automatically closed

asyncio.run(main())

💡Using async with automatically closes the client and cleans up resources.

Manual Management

python

import asyncio
from zendbx import ZenDBX

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

async def main():
    # Your code here
    await client.close()  # Always close when done

asyncio.run(main())

⚠️Always call await client.close() to prevent memory leaks.

With Environment Variables

python

import os
from dotenv import load_dotenv
from zendbx import ZenDBX

load_dotenv()

async def main():
    async with ZenDBX(
        project_url=os.getenv("ZENDBX_PROJECT_URL"),
        anon_key=os.getenv("ZENDBX_ANON_KEY")
    ) as client:
        # Your code here
        pass

asyncio.run(main())

Create a .env file:

bash

ZENDBX_PROJECT_URL=https://api.zendbx.in/p/your-project
ZENDBX_ANON_KEY=your-anon-key

Parameter	Type	Required	Description
project_url	str	required	Your ZenDBX project URL (e.g., https://api.zendbx.in/p/your-project)
anon_key	str	required	Your project anon (public) key from dashboard
service_key	str	optional	Service key for admin operations (server-side only)
timeout	int	optional	Request timeout in seconds (default: 30)

Authentication

python

# Sign up a new user
response = await client.auth.sign_up(
    email="user@example.com",
    password="secure_password123",
    name="John Doe"  # Optional
)

if "access_token" in response:
    print(f"User created: {response['user']['email']}")
    print(f"Token: {response['access_token']}")
else:
    print(f"Error: {response}")

python

# Sign in existing user
response = await client.auth.sign_in(
    email="user@example.com",
    password="secure_password123"
)

if "access_token" in response:
    print("Sign in successful!")
else:
    print("Authentication failed")

Get Current User

python

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

print(f"User ID: {user['id']}")
print(f"Email: {user['email']}")
print(f"Name: {user.get('name', 'N/A')}")

Session Management (v1.0.3+)

Save and restore authentication sessions across client instances.

python

# NEW in v1.0.3: Session Management

# Save session token
response = await client.auth.sign_in(
    email="user@example.com",
    password="password123"
)
access_token = response["access_token"]

# Later, restore session in new client
async with ZenDBX(project_url="...", anon_key="...") as new_client:
    # Restore session
    new_client.auth.set_session(access_token)
    
    # Now can make authenticated requests
    user = await new_client.auth.get_user()
    print(f"Restored session for: {user['email']}")
    
    # Clear session (local only, no backend call)
    new_client.auth.clear_session()

Parameter	Type	Required	Description
set_session(access_token, refresh_token=None)	method	optional	Restore authentication session in new client
clear_session()	method	optional	Clear local session without backend call

Sign Out

python

# Sign out current user
await client.auth.sign_out()
print("Signed out successfully")

# Note: sign_out() always clears local session
# even if backend call fails (resilient logout)

Database Operations

SELECT

pythonBasic select

# Select all columns
response = await client.from_("users").select("*").execute()
print(response)

# Select specific columns
response = await client.from_("users").select("id, name, email").execute()

pythonWith filters and ordering

# With filters
response = await client.from_("users") \
    .select("*") \
    .eq("status", "active") \
    .order_by("created_at", desc=True) \
    .limit(20) \
    .execute()

Query Filters

python

# Equality
.eq("column", value)
.neq("column", value)

# Comparison
.gt("age", 18)
.gte("score", 90)
.lt("price", 100)
.lte("quantity", 50)

# String matching
.like("name", "%john%")
.ilike("email", "%@gmail.com")

# In list
.in_("status", ["active", "pending"])

# Between
.between("age", 18, 65)

# Ordering
.order_by("created_at", desc=True)  # descending
.order_by("name")  # ascending (default)

# Pagination
.limit(10)
.offset(20)

💡The Python SDK uses desc=True for descending order, and in_() (with underscore) to avoid Python keyword conflict.

INSERT

python

# Insert one row
response = await client.from_("todos").insert({
    "title": "My first todo",
    "completed": False
}).execute()

# Insert multiple rows
response = await client.from_("todos").insert([
    {"title": "Task 1", "completed": False},
    {"title": "Task 2", "completed": True}
]).execute()

UPDATE

python

# Update with filter
response = await client.from_("todos") \
    .update({"completed": True}) \
    .eq("id", "some-uuid") \
    .execute()

# Update multiple rows
response = await client.from_("todos") \
    .update({"status": "archived"}) \
    .eq("completed", True) \
    .execute()

⚠️Always chain at least one filter before .update() to avoid modifying all rows.

DELETE

python

# Delete with filter
response = await client.from_("todos") \
    .delete() \
    .eq("id", "some-uuid") \
    .execute()

# Delete multiple rows
response = await client.from_("todos") \
    .delete() \
    .eq("completed", True) \
    .execute()

⚠️Always chain at least one filter before .delete() to avoid deleting all rows.

Storage

Upload, list, and delete files from your ZenDBX buckets.

Upload File

python

# Upload a file
with open("avatar.png", "rb") as f:
    response = await client.storage \
        .from_("avatars") \
        .upload(f, "user-123.png")
    
    if "error" not in response:
        print(f"Uploaded: {response['url']}")

List Files

python

# List files in bucket
response = await client.storage \
    .from_("avatars") \
    .list()

for file in response["files"]:
    print(f"{file['name']} - {file['size']} bytes")

Delete Files

python

# Delete a file
await client.storage \
    .from_("avatars") \
    .delete("file-uuid")

# Bulk delete
await client.storage \
    .from_("avatars") \
    .bulk_delete(["uuid-1", "uuid-2"])

Error Handling

The SDK provides typed exceptions for different error scenarios.

python

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

try:
    response = await client.auth.sign_in(
        email="user@example.com",
        password="wrong_password"
    )
except ZenDBXAuthenticationError as e:
    print(f"Authentication failed: {e}")
except ZenDBXValidationError as e:
    print(f"Validation error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Parameter	Type	Required	Description
ZenDBXError	Exception	optional	Base exception class
ZenDBXAuthenticationError	Exception	optional	Raised on 401 Unauthorized
ZenDBXPermissionError	Exception	optional	Raised on 403 Forbidden
ZenDBXNotFoundError	Exception	optional	Raised on 404 Not Found
ZenDBXValidationError	Exception	optional	Raised on 400/422 validation errors
ZenDBXConflictError	Exception	optional	Raised on 409 Conflict
ZenDBXRateLimitError	Exception	optional	Raised on 429 Too Many Requests
ZenDBXTimeoutError	Exception	optional	Raised on request timeout

Web Framework Integration

Flask Example

python

from flask import Flask, session, request, jsonify
from zendbx import ZenDBX
import asyncio
import os

app = Flask(__name__)
app.secret_key = os.getenv("FLASK_SECRET_KEY")

def run_async(coro):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        return loop.run_until_complete(coro)
    finally:
        loop.close()

@app.route("/api/auth/signup", methods=["POST"])
def signup():
    data = request.get_json()
    
    async def do_signup():
        async with ZenDBX(
            project_url=os.getenv("ZENDBX_PROJECT_URL"),
            anon_key=os.getenv("ZENDBX_ANON_KEY")
        ) as client:
            response = await client.auth.sign_up(
                email=data["email"],
                password=data["password"]
            )
            
            if "access_token" in response:
                session["access_token"] = response["access_token"]
                return {"user": response["user"]}, 201
            return {"error": "Signup failed"}, 400
    
    return jsonify(*run_async(do_signup()))

@app.route("/api/todos", methods=["GET"])
def get_todos():
    if "access_token" not in session:
        return jsonify({"error": "Not authenticated"}), 401
    
    async def do_get_todos():
        async with ZenDBX(
            project_url=os.getenv("ZENDBX_PROJECT_URL"),
            anon_key=os.getenv("ZENDBX_ANON_KEY")
        ) as client:
            # Restore session
            client.auth.set_session(session["access_token"])
            
            # Get todos
            response = await client.from_("todos") \
                .select("*") \
                .order_by("created_at", desc=True) \
                .execute()
            
            return {"todos": response}, 200
    
    return jsonify(*run_async(do_get_todos()))

if __name__ == "__main__":
    app.run(debug=True)

💡Flask is synchronous, so we use asyncio.run() to run async ZenDBX operations.

FastAPI Example

python

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from zendbx import ZenDBX
import os

app = FastAPI()

class SignupRequest(BaseModel):
    email: str
    password: str
    name: str = ""

async def get_client():
    """Dependency to create ZenDBX client"""
    async with ZenDBX(
        project_url=os.getenv("ZENDBX_PROJECT_URL"),
        anon_key=os.getenv("ZENDBX_ANON_KEY")
    ) as client:
        yield client

@app.post("/auth/signup")
async def signup(
    data: SignupRequest,
    client: ZenDBX = Depends(get_client)
):
    response = await client.auth.sign_up(
        email=data.email,
        password=data.password,
        name=data.name
    )
    
    if "access_token" in response:
        return {"user": response["user"]}
    raise HTTPException(status_code=400, detail="Signup failed")

@app.get("/todos")
async def get_todos(
    access_token: str,
    client: ZenDBX = Depends(get_client)
):
    # Restore session
    client.auth.set_session(access_token)
    
    try:
        response = await client.from_("todos") \
            .select("*") \
            .order_by("created_at", desc=True) \
            .execute()
        return {"todos": response}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

💡FastAPI natively supports async, making it a perfect match for the ZenDBX Python SDK.

Best Practices

• Use async with context manager to ensure proper cleanup
• Store credentials in environment variables, never hardcode them
• Use set_session() for session persistence across requests
• Handle specific exceptions (ZenDBXAuthenticationError, etc.)
• Always await async functions
• Don't share client instances across processes or threads
• Don't forget to close clients in manual management
• Don't commit .env files to version control

Using in Jupyter Notebooks

python

# In Jupyter, the event loop is already running
# Use await directly instead of asyncio.run()

from zendbx import ZenDBX

async with ZenDBX(
    project_url="https://api.zendbx.in/p/your-project",
    anon_key="your-anon-key"
) as client:
    response = await client.auth.sign_in(
        email="user@example.com",
        password="password123"
    )
    print(response)
    
    # Query data
    todos = await client.from_("todos").select("*").execute()
    print(todos)

Additional Resources

Python SDK Reference

The official Python SDK for ZenDBX. Async-first architecture with full support for authentication, database operations, and storage.

zendbxv1.0.3Python 3.8+View on PyPI →

Installation

bash

pip install zendbx

Or install a specific version:

bash

pip install zendbx==1.0.3

Requirements

• Python 3.8 or higher
• aiohttp (async HTTP client)
• pydantic (data validation)

Initialize Client

ZenDBX Python SDK is async-first. Always use async/await and close the client when done.

Context Manager (Recommended)

python

import asyncio
from zendbx import ZenDBX

async def main():
    # Using context manager (recommended)
    async with ZenDBX(
        project_url="https://api.zendbx.in/p/your-project",
        anon_key="your-anon-key"
    ) as client:
        # Your code here
        user = await client.auth.get_user()
        print(user)
    # Client automatically closed

asyncio.run(main())

💡Using async with automatically closes the client and cleans up resources.

Manual Management

python

import asyncio
from zendbx import ZenDBX

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

async def main():
    # Your code here
    await client.close()  # Always close when done

asyncio.run(main())

⚠️Always call await client.close() to prevent memory leaks.

With Environment Variables

python

import os
from dotenv import load_dotenv
from zendbx import ZenDBX

load_dotenv()

async def main():
    async with ZenDBX(
        project_url=os.getenv("ZENDBX_PROJECT_URL"),
        anon_key=os.getenv("ZENDBX_ANON_KEY")
    ) as client:
        # Your code here
        pass

asyncio.run(main())

Create a .env file:

bash

ZENDBX_PROJECT_URL=https://api.zendbx.in/p/your-project
ZENDBX_ANON_KEY=your-anon-key

Parameter	Type	Required	Description
project_url	str	required	Your ZenDBX project URL (e.g., https://api.zendbx.in/p/your-project)
anon_key	str	required	Your project anon (public) key from dashboard
service_key	str	optional	Service key for admin operations (server-side only)
timeout	int	optional	Request timeout in seconds (default: 30)

Authentication

python

# Sign up a new user
response = await client.auth.sign_up(
    email="user@example.com",
    password="secure_password123",
    name="John Doe"  # Optional
)

if "access_token" in response:
    print(f"User created: {response['user']['email']}")
    print(f"Token: {response['access_token']}")
else:
    print(f"Error: {response}")

python

# Sign in existing user
response = await client.auth.sign_in(
    email="user@example.com",
    password="secure_password123"
)

if "access_token" in response:
    print("Sign in successful!")
else:
    print("Authentication failed")

Get Current User

python

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

print(f"User ID: {user['id']}")
print(f"Email: {user['email']}")
print(f"Name: {user.get('name', 'N/A')}")

Session Management (v1.0.3+)

Save and restore authentication sessions across client instances.

python

# NEW in v1.0.3: Session Management

# Save session token
response = await client.auth.sign_in(
    email="user@example.com",
    password="password123"
)
access_token = response["access_token"]

# Later, restore session in new client
async with ZenDBX(project_url="...", anon_key="...") as new_client:
    # Restore session
    new_client.auth.set_session(access_token)
    
    # Now can make authenticated requests
    user = await new_client.auth.get_user()
    print(f"Restored session for: {user['email']}")
    
    # Clear session (local only, no backend call)
    new_client.auth.clear_session()

Parameter	Type	Required	Description
set_session(access_token, refresh_token=None)	method	optional	Restore authentication session in new client
clear_session()	method	optional	Clear local session without backend call

Sign Out

python

# Sign out current user
await client.auth.sign_out()
print("Signed out successfully")

# Note: sign_out() always clears local session
# even if backend call fails (resilient logout)

Database Operations

SELECT

pythonBasic select

# Select all columns
response = await client.from_("users").select("*").execute()
print(response)

# Select specific columns
response = await client.from_("users").select("id, name, email").execute()

pythonWith filters and ordering

# With filters
response = await client.from_("users") \
    .select("*") \
    .eq("status", "active") \
    .order_by("created_at", desc=True) \
    .limit(20) \
    .execute()

Query Filters

python

# Equality
.eq("column", value)
.neq("column", value)

# Comparison
.gt("age", 18)
.gte("score", 90)
.lt("price", 100)
.lte("quantity", 50)

# String matching
.like("name", "%john%")
.ilike("email", "%@gmail.com")

# In list
.in_("status", ["active", "pending"])

# Between
.between("age", 18, 65)

# Ordering
.order_by("created_at", desc=True)  # descending
.order_by("name")  # ascending (default)

# Pagination
.limit(10)
.offset(20)

💡The Python SDK uses desc=True for descending order, and in_() (with underscore) to avoid Python keyword conflict.

INSERT

python

# Insert one row
response = await client.from_("todos").insert({
    "title": "My first todo",
    "completed": False
}).execute()

# Insert multiple rows
response = await client.from_("todos").insert([
    {"title": "Task 1", "completed": False},
    {"title": "Task 2", "completed": True}
]).execute()

UPDATE

python

# Update with filter
response = await client.from_("todos") \
    .update({"completed": True}) \
    .eq("id", "some-uuid") \
    .execute()

# Update multiple rows
response = await client.from_("todos") \
    .update({"status": "archived"}) \
    .eq("completed", True) \
    .execute()

⚠️Always chain at least one filter before .update() to avoid modifying all rows.

DELETE

python

# Delete with filter
response = await client.from_("todos") \
    .delete() \
    .eq("id", "some-uuid") \
    .execute()

# Delete multiple rows
response = await client.from_("todos") \
    .delete() \
    .eq("completed", True) \
    .execute()

⚠️Always chain at least one filter before .delete() to avoid deleting all rows.

Storage

Upload, list, and delete files from your ZenDBX buckets.

Upload File

python

# Upload a file
with open("avatar.png", "rb") as f:
    response = await client.storage \
        .from_("avatars") \
        .upload(f, "user-123.png")
    
    if "error" not in response:
        print(f"Uploaded: {response['url']}")

List Files

python

# List files in bucket
response = await client.storage \
    .from_("avatars") \
    .list()

for file in response["files"]:
    print(f"{file['name']} - {file['size']} bytes")

Delete Files

python

# Delete a file
await client.storage \
    .from_("avatars") \
    .delete("file-uuid")

# Bulk delete
await client.storage \
    .from_("avatars") \
    .bulk_delete(["uuid-1", "uuid-2"])

Error Handling

The SDK provides typed exceptions for different error scenarios.

python

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

try:
    response = await client.auth.sign_in(
        email="user@example.com",
        password="wrong_password"
    )
except ZenDBXAuthenticationError as e:
    print(f"Authentication failed: {e}")
except ZenDBXValidationError as e:
    print(f"Validation error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Parameter	Type	Required	Description
ZenDBXError	Exception	optional	Base exception class
ZenDBXAuthenticationError	Exception	optional	Raised on 401 Unauthorized
ZenDBXPermissionError	Exception	optional	Raised on 403 Forbidden
ZenDBXNotFoundError	Exception	optional	Raised on 404 Not Found
ZenDBXValidationError	Exception	optional	Raised on 400/422 validation errors
ZenDBXConflictError	Exception	optional	Raised on 409 Conflict
ZenDBXRateLimitError	Exception	optional	Raised on 429 Too Many Requests
ZenDBXTimeoutError	Exception	optional	Raised on request timeout

Web Framework Integration

Flask Example

python

from flask import Flask, session, request, jsonify
from zendbx import ZenDBX
import asyncio
import os

app = Flask(__name__)
app.secret_key = os.getenv("FLASK_SECRET_KEY")

def run_async(coro):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        return loop.run_until_complete(coro)
    finally:
        loop.close()

@app.route("/api/auth/signup", methods=["POST"])
def signup():
    data = request.get_json()
    
    async def do_signup():
        async with ZenDBX(
            project_url=os.getenv("ZENDBX_PROJECT_URL"),
            anon_key=os.getenv("ZENDBX_ANON_KEY")
        ) as client:
            response = await client.auth.sign_up(
                email=data["email"],
                password=data["password"]
            )
            
            if "access_token" in response:
                session["access_token"] = response["access_token"]
                return {"user": response["user"]}, 201
            return {"error": "Signup failed"}, 400
    
    return jsonify(*run_async(do_signup()))

@app.route("/api/todos", methods=["GET"])
def get_todos():
    if "access_token" not in session:
        return jsonify({"error": "Not authenticated"}), 401
    
    async def do_get_todos():
        async with ZenDBX(
            project_url=os.getenv("ZENDBX_PROJECT_URL"),
            anon_key=os.getenv("ZENDBX_ANON_KEY")
        ) as client:
            # Restore session
            client.auth.set_session(session["access_token"])
            
            # Get todos
            response = await client.from_("todos") \
                .select("*") \
                .order_by("created_at", desc=True) \
                .execute()
            
            return {"todos": response}, 200
    
    return jsonify(*run_async(do_get_todos()))

if __name__ == "__main__":
    app.run(debug=True)

💡Flask is synchronous, so we use asyncio.run() to run async ZenDBX operations.

FastAPI Example

python

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from zendbx import ZenDBX
import os

app = FastAPI()

class SignupRequest(BaseModel):
    email: str
    password: str
    name: str = ""

async def get_client():
    """Dependency to create ZenDBX client"""
    async with ZenDBX(
        project_url=os.getenv("ZENDBX_PROJECT_URL"),
        anon_key=os.getenv("ZENDBX_ANON_KEY")
    ) as client:
        yield client

@app.post("/auth/signup")
async def signup(
    data: SignupRequest,
    client: ZenDBX = Depends(get_client)
):
    response = await client.auth.sign_up(
        email=data.email,
        password=data.password,
        name=data.name
    )
    
    if "access_token" in response:
        return {"user": response["user"]}
    raise HTTPException(status_code=400, detail="Signup failed")

@app.get("/todos")
async def get_todos(
    access_token: str,
    client: ZenDBX = Depends(get_client)
):
    # Restore session
    client.auth.set_session(access_token)
    
    try:
        response = await client.from_("todos") \
            .select("*") \
            .order_by("created_at", desc=True) \
            .execute()
        return {"todos": response}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

💡FastAPI natively supports async, making it a perfect match for the ZenDBX Python SDK.

Best Practices

• Use async with context manager to ensure proper cleanup
• Store credentials in environment variables, never hardcode them
• Use set_session() for session persistence across requests
• Handle specific exceptions (ZenDBXAuthenticationError, etc.)
• Always await async functions
• Don't share client instances across processes or threads
• Don't forget to close clients in manual management
• Don't commit .env files to version control

Using in Jupyter Notebooks

python

# In Jupyter, the event loop is already running
# Use await directly instead of asyncio.run()

from zendbx import ZenDBX

async with ZenDBX(
    project_url="https://api.zendbx.in/p/your-project",
    anon_key="your-anon-key"
) as client:
    response = await client.auth.sign_in(
        email="user@example.com",
        password="password123"
    )
    print(response)
    
    # Query data
    todos = await client.from_("todos").select("*").execute()
    print(todos)

Python SDK Reference

Installation

Requirements

Initialize Client

Context Manager (Recommended)

Manual Management

With Environment Variables

Authentication

Sign Up

Sign In

Get Current User

Session Management (v1.0.3+)

Sign Out

Database Operations

SELECT

Query Filters

INSERT

UPDATE

DELETE

Storage

Upload File

List Files

Delete Files

Error Handling

Web Framework Integration

Flask Example

FastAPI Example

Best Practices

Using in Jupyter Notebooks

Additional Resources

Python SDK Reference

Installation

Requirements

Initialize Client

Context Manager (Recommended)

Manual Management

With Environment Variables

Authentication

Sign Up

Sign In

Get Current User

Session Management (v1.0.3+)

Sign Out

Database Operations

SELECT

Query Filters

INSERT

UPDATE

DELETE

Storage

Upload File

List Files

Delete Files

Error Handling

Web Framework Integration

Flask Example

FastAPI Example

Best Practices

Using in Jupyter Notebooks

Additional Resources