Skip to main content

Examples & Cookbook

Real-world code examples, recipes for common use cases, and production-ready patterns for using the Taruvi SDK effectively.

Database Examples​

Complex Query with Multiple Filters​

# Search for active adult users created this year
from datetime import datetime

year_start = datetime.now().replace(month=1, day=1).isoformat()

result = (
    auth_client.database.from_("users")
    .filter("is_active", "eq", True)
    .filter("age", "gte", 18)
    .filter("created_at", "gte", year_start)
    .sort("created_at", "desc")
    .page_size(50)
    .execute()
)

print(f"Found {result['total']} active adult users created this year")
for user in result["data"]:
    print(f"  {user['username']}: {user['email']}")

Bulk Create with Error Handling​

from taruvi import ValidationError

users_to_create = [
    {"username": "user1", "email": "[email protected]"},
    {"username": "user2", "email": "[email protected]"},
    {"username": "user3", "email": "[email protected]"}
]

try:
    new_users = auth_client.database.create("users", users_to_create)
    print(f"Created {len(new_users)} users")
except ValidationError as e:
    print(f"Validation failed: {e.details}")

Pagination for Large Datasets​

# ✅ Good: Paginate large results
for page_num in range(1, 11):
    result = auth_client.database.from_("users").page(page_num).page_size(100).execute()
    process_users(result["data"])
    if len(result["data"]) < 100:
        break  # No more pages

# ✅ Also good: Use first() for single record
user = auth_client.database.from_("users").filter("email", "eq", "[email protected]").first()

Use populate() to Avoid N+1 Queries​

# ✅ Good: Eager load relationships
result = auth_client.database.from_("posts").populate("author", "comments").execute()
posts = result["data"]

# ❌ Bad: N+1 queries
result = auth_client.database.from_("posts").execute()
for post in result["data"]:
    author = auth_client.database.get("users", post['author_id'])  # Extra query!

Search, Aggregation, and Graph Traversal​

# Full-text search
result = auth_client.database.from_("products").search("laptop").execute()

# Aggregations with group by
result = (
    auth_client.database.from_("orders")
    .aggregate("sum(total)", "count(*)")
    .group_by("status")
    .having("count(*) > 5")
    .execute()
)

# Graph/tree traversal (hierarchical data)
result = (
    auth_client.database.from_("employees")
    .format("tree")
    .include("descendants")
    .depth(3)
    .types(["manager", "dotted_line"])
    .execute()
)

# Edge (relationship) queries
result = (
    auth_client.database.from_("employees")
    .edges()
    .filter("relationship_type", "eq", "manager")
    .execute()
)

Function Examples​

Execute Function with Polling​

import time

# Execute async
result = auth_client.functions.execute(
    "process-large-dataset",
    params={"dataset_id": 123},
    is_async=True
)

task_id = result['invocation']['celery_task_id']

# Poll for result
while True:
    task_result = auth_client.functions.get_result(task_id)
    status = task_result.get('status', task_result.get('data', {}).get('status'))

    if status == 'SUCCESS':
        print(f"Success: {task_result.get('result')}")
        break
    elif status == 'FAILURE':
        print(f"Failed: {task_result.get('traceback')}")
        break

    print("Still processing...")
    time.sleep(2)

List and Inspect Functions​

# List all functions in the app
functions = auth_client.functions.list()

# Get function details
func = auth_client.functions.get("process-order")
print(f"Function: {func['name']} (active: {func['is_active']})")

# List invocations with filters
invocations = auth_client.functions.list_invocations(
    function_slug="process-order",
    status="SUCCESS",
    limit=10
)

Storage Examples​

Upload and Process Image​

# Upload image
with open("profile.jpg", "rb") as f:
    result = auth_client.storage.from_("images").upload(
        files=[("profile.jpg", f)],
        paths=["users/123/profile.jpg"],
        metadatas=[{"type": "profile_photo"}]
    )

print(f"Upload result: {result}")

# Process with function
process_result = auth_client.functions.execute(
    "resize-image",
    params={"path": "users/123/profile.jpg", "width": 200, "height": 200}
)

print(f"Processed: {process_result['data']}")

Batch Upload Multiple Files​

# Upload multiple files at once (max 10 per request)
files = [
    ("report-q1.pdf", open("q1.pdf", "rb")),
    ("report-q2.pdf", open("q2.pdf", "rb")),
    ("report-q3.pdf", open("q3.pdf", "rb")),
]

result = auth_client.storage.from_("reports").upload(
    files=files,
    paths=[
        "2024/q1-report.pdf",
        "2024/q2-report.pdf",
        "2024/q3-report.pdf",
    ],
    metadatas=[
        {"quarter": "Q1", "year": "2024"},
        {"quarter": "Q2", "year": "2024"},
        {"quarter": "Q3", "year": "2024"},
    ]
)

print(f"Uploaded: {result.get('uploaded_count')}, Failed: {result.get('failed_count')}")

List and Filter Files​

# List with filters
images = (
    auth_client.storage.from_("uploads")
    .filter(mimetype_category="image", ordering="-created_at", page_size=50)
    .list()
)

# Search by filename
docs = (
    auth_client.storage.from_("documents")
    .filter(search="invoice", visibility="private")
    .list()
)

Copy and Move Files​

# Copy within same bucket
auth_client.storage.from_("images").copy_object(
    "users/123/avatar.jpg",
    "users/123/avatar-backup.jpg"
)

# Copy to different bucket
auth_client.storage.from_("uploads").copy_object(
    "temp/report.pdf",
    "report.pdf",
    destination_bucket="archives"
)

# Move/rename a file
auth_client.storage.from_("documents").move_object(
    "drafts/proposal.pdf",
    "final/proposal-v2.pdf"
)

Update Metadata and Visibility​

# Update file metadata
auth_client.storage.from_("documents").update(
    "reports/annual.pdf",
    metadata={"status": "approved", "reviewer": "alice"},
    visibility="public"
)

Bucket Management​

# Create bucket with quotas and MIME restrictions
bucket = auth_client.storage.create_bucket(
    name="User Avatars",
    slug="user-avatars",
    visibility="public",
    file_size_limit=5242880,            # 5MB per file
    allowed_mime_types=["image/*"],
    max_size_bytes=1073741824,          # 1GB total
    max_objects=5000
)

# List buckets
buckets = auth_client.storage.list_buckets(
    search="avatar",
    visibility="public",
    page_size=20
)

# Update bucket settings
auth_client.storage.update_bucket(
    "user-avatars",
    file_size_limit=10485760,           # Increase to 10MB
    max_objects=10000
)

# Delete bucket (WARNING: deletes all files)
auth_client.storage.delete_bucket("old-bucket")

User Management Examples​

CRUD Operations​

# Get user details
user = auth_client.users.get("alice")
print(f"Email: {user['data']['email']}")

# Create a new user
new_user = auth_client.users.create({
    "username": "frank",
    "email": "[email protected]",
    "password": "secure123",
    "confirm_password": "secure123",
    "first_name": "Frank",
    "last_name": "Smith"
})

# Update user
auth_client.users.update("frank", {
    "first_name": "Franklin",
    "is_active": True
})

# Delete user
auth_client.users.delete("frank")

List and Filter Users​

# List active admins
users = auth_client.users.list(
    is_active=True,
    roles="admin",
    ordering="-created_at",
    page=1,
    page_size=20
)
for user in users["results"]:
    print(f"  {user['username']}: {user['email']}")

# Search users
users = auth_client.users.list(search="alice")

# Get user's apps
apps = auth_client.users.apps("alice")

Role Assignment​

# Assign roles to multiple users
result = auth_client.users.assign_roles(
    roles=["admin", "manager"],
    usernames=["alice", "bob"],
    expires_at="2025-12-31T23:59:59Z"  # Optional expiration
)
print(result["message"])

# Revoke roles
result = auth_client.users.revoke_roles(
    roles=["admin"],
    usernames=["bob"]
)

Policy (Authorization) Examples​

Check Resource Permissions​

# Check if user can read and write to a datatable
result = auth_client.policy.check_resources([
    {
        "resource": {"kind": "datatable", "id": "orders"},
        "actions": ["read", "write", "delete"]
    }
])

for check in result["results"]:
    for action, effect in check["actions"].items():
        print(f"  {action}: {'✅' if effect == 'EFFECT_ALLOW' else '❌'}")

Get Allowed Actions​

# Get all allowed actions for a resource
allowed = auth_client.policy.get_allowed_actions(
    {"kind": "datatable", "id": "users"}
)
print(f"Allowed actions: {allowed}")  # ['read', 'write', 'update']

Filter Allowed Resources​

# Filter to only resources where user has read+write access
all_tables = [
    {"kind": "datatable", "id": "users"},
    {"kind": "datatable", "id": "orders"},
    {"kind": "datatable", "id": "invoices"},
]
allowed = auth_client.policy.filter_allowed(all_tables, ["read", "write"])
print(f"Can read+write: {[r['id'] for r in allowed]}")

Analytics Examples​

# Execute analytics query with parameters
result = auth_client.analytics.execute(
    "monthly-revenue",
    params={
        "start_date": "2024-01-01",
        "end_date": "2024-12-31",
        "group_by": "month"
    }
)
print(result["data"])

# Query with filters
result = auth_client.analytics.execute(
    "sales-by-region",
    params={"region": "US", "product_category": "electronics"}
)

Secrets Examples​

Get and List Secrets​

# Get a single secret (uses client's app_slug automatically)
api_key = auth_client.secrets.get("STRIPE_API_KEY")
print(f"Key: {api_key['value'][:8]}...")

# Get with tag validation
secret = auth_client.secrets.get(
    "DB_PASSWORD",
    tags=["production", "database"]
)

# List secrets with filters
result = auth_client.secrets.list(
    secret_type="api_key",
    tags=["production"],
    page_size=50
)
for secret in result["data"]:
    print(f"  {secret['key']}: {secret.get('secret_type', 'N/A')}")

# Batch get multiple secrets by keys
result = auth_client.secrets.list(keys=["API_KEY", "DB_PASSWORD", "REDIS_URL"])

App & Settings Examples​

# Get app roles
roles = auth_client.app.roles()
print(f"Roles: {roles}")

# Get app settings
settings = auth_client.app.settings()
print(f"App: {settings.get('display_name')}")
print(f"Color: {settings.get('primary_color')}")

# Get site metadata
site = auth_client.settings.get()
print(f"Site: {site}")

Integration Examples​

FastAPI Integration​

from fastapi import FastAPI, Depends, HTTPException
from taruvi import Client, NotFoundError

app = FastAPI()

def get_client():
    client = Client(api_url="https://api.taruvi.cloud", app_slug="my-app")
    return client.auth.signInWithToken(
        token="api-key",
        token_type="api_key"
    )

@app.get("/users/{username}")
async def get_user(username: str, client = Depends(get_client)):
    try:
        user = client.users.get(username)
        return user
    except NotFoundError:
        raise HTTPException(status_code=404, detail="User not found")

@app.get("/data/{table}")
async def list_data(table: str, page: int = 1, client = Depends(get_client)):
    result = client.database.from_(table).page(page).page_size(20).execute()
    return {"data": result["data"], "total": result["total"]}

Security Best Practices​

Never Hardcode Credentials​

# ❌ Bad: Hardcoded credentials
client = Client(
    api_url="https://api.taruvi.cloud",
    app_slug="my-app"
)
auth_client = client.auth.signInWithPassword(
    username="[email protected]",
    password="supersecret123"
)

# ✅ Good: Environment variables
import os
from dotenv import load_dotenv

load_dotenv()
client = Client(
    api_url=os.getenv("TARUVI_API_URL"),
    app_slug=os.getenv("TARUVI_APP_SLUG")
)
auth_client = client.auth.signInWithPassword(
    username=os.getenv("TARUVI_USERNAME"),
    password=os.getenv("TARUVI_PASSWORD")
)

Use API Keys for Machine-to-Machine​

# ✅ Good: API key for background jobs
auth_client = client.auth.signInWithToken(
    token=os.getenv("TARUVI_API_KEY"),
    token_type="api_key"
)
Token Rotation

Rotate API keys and tokens regularly, especially after team member changes.

Performance Best Practices​

Reuse Clients​

# ✅ Good: Create client once
client = Client(api_url="https://api.taruvi.cloud", app_slug="my-app")
auth_client = client.auth.signInWithPassword(username="...", password="...")

# Use throughout application
result = auth_client.database.from_("users").execute()
func_result = auth_client.functions.execute("my-function")

# ❌ Bad: Create new client for each operation
client1 = Client(api_url="https://api.taruvi.cloud", app_slug="my-app")
auth1 = client1.auth.signInWithPassword(username="...", password="...")
result = auth1.database.from_("users").execute()

client2 = Client(api_url="https://api.taruvi.cloud", app_slug="my-app")
auth2 = client2.auth.signInWithPassword(username="...", password="...")
func_result = auth2.functions.execute("my-function")

Use Async for Concurrent Operations​

# ✅ Good: Concurrent async
import asyncio

client = Client(api_url="https://api.taruvi.cloud", app_slug="my-app", mode='async')
auth_client = client.auth.signInWithToken(token="your-jwt", token_type="jwt")

users, posts, comments = await asyncio.gather(
    auth_client.database.from_("users").execute(),
    auth_client.database.from_("posts").execute(),
    auth_client.database.from_("comments").execute()
)
# Each returns {"data": [...], "total": N}

Error Handling Patterns​

Catch Specific Exceptions​

from taruvi import (
    NotFoundError, ValidationError, AuthenticationError,
    NotAuthenticatedError, AuthorizationError, RateLimitError,
    ServerError, TaruviError
)

# ✅ Good: Specific exception handling
try:
    user = auth_client.database.get("users", record_id=999)
except NotFoundError:
    return {"error": "User not found"}, 404
except ValidationError as e:
    return {"error": e.message, "details": e.details}, 400
except AuthorizationError:
    return {"error": "Permission denied"}, 403
except RateLimitError:
    return {"error": "Too many requests"}, 429
except TaruviError as e:
    return {"error": "Internal error"}, 500

# ❌ Bad: Generic exception
try:
    user = auth_client.database.get("users", record_id=999)
except Exception as e:
    return {"error": str(e)}, 500  # Exposes internal details!

Implement Retry Logic​

# ✅ Good: Retry with backoff
import time
from taruvi import ServerError

def with_retry(func, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            return func()
        except ServerError:
            if attempt == max_attempts - 1:
                raise
            time.sleep(2 ** attempt)

users = with_retry(lambda: auth_client.database.from_("users").execute())

Code Organization​

Use Dependency Injection​

# ✅ Good: Dependency injection
class UserService:
    def __init__(self, client):
        self.client = client

    def get_active_users(self):
        result = self.client.database.from_("users").filter("is_active", "eq", True).execute()
        return result["data"]

# Easy to test with mock client
service = UserService(auth_client)
users = service.get_active_users()

Create Service Classes​

# ✅ Good: Organized service classes
class DatabaseService:
    def __init__(self, client):
        self.client = client

    def get_user(self, username):
        return self.client.users.get(username)

    def create_record(self, table, data):
        return self.client.database.create(table, data)

    def query(self, table):
        return self.client.database.from_(table)

class FunctionService:
    def __init__(self, client):
        self.client = client

    def send_email(self, to, subject, body):
        return self.client.functions.execute("send-email", params={
            "to": to,
            "subject": subject,
            "body": body
        })

Testing​

Mock SDK Calls​

from unittest.mock import Mock, patch

def test_get_users():
    # Mock the database query
    mock_client = Mock()
    mock_client.database.query.return_value.get.return_value = [
        {"id": 1, "username": "alice"},
        {"id": 2, "username": "bob"}
    ]

    service = UserService(mock_client)
    users = service.get_active_users()

    assert len(users) == 2
    assert users[0]["username"] == "alice"