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

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

print(f"Found {len(users)} active adult users created this year")

Bulk Create with Error Handling

from taruvi import ValidationError

users_to_create = [
    {"username": "user1", "email": "user1@example.com"},
    {"username": "user2", "email": "user2@example.com"},
    {"username": "user3", "email": "user3@example.com"}
]

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 in range(1, 11):
    users = auth_client.database.query("users").page(page).page_size(100).get()
    process_users(users)

# ❌ Bad: Load all at once
users = auth_client.database.query("users").get()  # Could be 10,000+ records!

Use populate() to Avoid N+1 Queries

# ✅ Good: Eager load relationships
posts = auth_client.database.query("posts").populate("author", "comments").get()

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

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['data']['status']

    if status == 'SUCCESS':
        print(f"Success: {task_result['data']['result']}")
        break
    elif status == 'FAILURE':
        print(f"Failed: {task_result['data']['error']}")
        break

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

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

file_path = result[0]['path']
print(f"Uploaded: {file_path}")

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

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

Integration Examples

FastAPI Integration

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

app = FastAPI()

def get_client():
    client = Client(api_url="...", app_slug="...")
    return client.auth.signInWithToken(
        token="api-key",
        token_type="api_key"
    )

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

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="alice@example.com",
    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="...", app_slug="...")
auth_client = client.auth.signInWithPassword(username="...", password="...")

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

# ❌ Bad: Create new client for each operation
client1 = Client(api_url="...", app_slug="...")
auth1 = client1.auth.signInWithPassword(username="...", password="...")
users = auth1.database.query("users").get()

client2 = Client(api_url="...", app_slug="...")
auth2 = client2.auth.signInWithPassword(username="...", password="...")
result = auth2.functions.execute("my-function")

Use Async for Concurrent Operations

# ✅ Good: Concurrent async
import asyncio

client = Client(mode='async', api_url="...", app_slug="...")
auth_client = client.auth.signInWithPassword(username="...", password="...")

users, posts, comments = await asyncio.gather(
    auth_client.database.query("users").get(),
    auth_client.database.query("posts").get(),
    auth_client.database.query("comments").get()
)

# ❌ Bad: Sequential sync
users = auth_client.database.query("users").get()
posts = auth_client.database.query("posts").get()
comments = auth_client.database.query("comments").get()

Error Handling Patterns

Catch Specific Exceptions

from taruvi import NotFoundError, ValidationError, 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 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.query("users").get())

Code Organization

Use Dependency Injection

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

    def get_active_users(self):
        return self.client.database.query("users").filter("is_active", "eq", True).get()

# 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, user_id):
        return self.client.database.get("users", record_id=user_id)

    def create_user(self, data):
        return self.client.database.create("users", data)

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"

Data Service API - Database operations reference
Functions API - Function execution reference
Storage API - File operations reference
Authentication Guide - Auth methods and patterns

Database Examples​

Complex Query with Multiple Filters​

Bulk Create with Error Handling​

Pagination for Large Datasets​

Use populate() to Avoid N+1 Queries​

Function Examples​

Execute Function with Polling​

Storage Examples​

Upload and Process Image​

Integration Examples​

FastAPI Integration​

Security Best Practices​

Never Hardcode Credentials​

Use API Keys for Machine-to-Machine​

Performance Best Practices​

Reuse Clients​

Use Async for Concurrent Operations​

Error Handling Patterns​

Catch Specific Exceptions​

Implement Retry Logic​

Code Organization​

Use Dependency Injection​

Create Service Classes​

Testing​

Mock SDK Calls​

Related Documentation​