Skip to main content

Taruvi SDK Guide for APP Mode Functions

Complete reference for using the Taruvi SDK within APP mode serverless functions.

Function Signature​

APP mode functions must use this exact signature:

def main(params, user_data, sdk_client):
    """
    APP mode function entry point.

    Args:
        params: User parameters + params['__function__'] metadata
        user_data: Dict with id, username, email, full_name
        sdk_client: Pre-authenticated Taruvi SDK client

    Returns:
        Any JSON-serializable data
    """
    pass

Parameters Explained​

params - Dictionary containing:

  • User-provided parameters from function invocation
  • params['__function__'] - Function metadata (name, slug, execution_mode, etc.)

user_data - Dictionary containing authenticated user info:

{
    "id": 123,
    "username": "alice",
    "email": "alice@example.com",
    "full_name": "Alice Smith"
}

sdk_client - Pre-authenticated Taruvi SDK client

  • Already authenticated with current user context
  • Ready to use - no need to call auth methods
  • Provides access to all platform features

Database Module​

CRUD operations on DataTables.

Methods​

sdk_client.database.query(table_name)           # Returns QueryBuilder (chainable)
sdk_client.database.get(table_name, record_id)  # Get single record
sdk_client.database.create(table_name, data)    # Create record
sdk_client.database.update(table_name, record_id, data)  # Update record
sdk_client.database.delete(table_name, record_id)        # Delete record

QueryBuilder Chaining​

.filter(field, operator, value)  # Operators: eq, ne, gt, gte, lt, lte, in, contains
.sort(field, direction)          # direction: "asc" or "desc"
.page_size(limit)                # Limit results
.execute()                       # Returns list[dict]

Examples​

Simple Query:

def main(params, user_data, sdk_client):
    users = sdk_client.database.query("users").execute()
    return {"users": users}

Query with Filters:

def main(params, user_data, sdk_client):
    active_users = sdk_client.database.query("users")\
        .filter("status", "eq", "active")\
        .filter("age", "gte", 18)\
        .sort("created_at", "desc")\
        .page_size(10)\
        .execute()

    return {"users": active_users, "total": len(active_users)}

Get Single Record:

def main(params, user_data, sdk_client):
    user_id = params.get("user_id")
    user = sdk_client.database.get("users", user_id)
    return {"user": user}

Create Record:

def main(params, user_data, sdk_client):
    new_user = sdk_client.database.create("users", {
        "username": params.get("username"),
        "email": params.get("email"),
        "status": "active"
    })
    return {"created": True, "user": new_user}

Update Record:

def main(params, user_data, sdk_client):
    user_id = params.get("user_id")
    updated_user = sdk_client.database.update("users", user_id, {
        "status": "inactive"
    })
    return {"updated": True, "user": updated_user}

Delete Record:

def main(params, user_data, sdk_client):
    user_id = params.get("user_id")
    sdk_client.database.delete("users", user_id)
    return {"deleted": True, "user_id": user_id}

Functions Module​

Execute other serverless functions and retrieve results.

Methods​

sdk_client.functions.execute(function_slug, params, is_async=False, timeout=None)
sdk_client.functions.get_result(task_id)
sdk_client.functions.list(limit=None, offset=None)
sdk_client.functions.get(function_slug)
sdk_client.functions.list_invocations(function_slug, status=None, limit=None, offset=None)

Examples​

Execute Sync Function:

def main(params, user_data, sdk_client):
    result = sdk_client.functions.execute(
        "send-email",
        params={
            "to": "user@example.com",
            "subject": "Welcome"
        }
    )
    return {"email_sent": True, "result": result}

Execute Async Function:

def main(params, user_data, sdk_client):
    result = sdk_client.functions.execute(
        "process-large-dataset",
        params={"dataset_id": 123},
        is_async=True
    )

    task_id = result['invocation']['celery_task_id']
    return {"task_started": True, "task_id": task_id}

Get Async Result:

def main(params, user_data, sdk_client):
    task_id = params.get("task_id")
    result = sdk_client.functions.get_result(task_id)

    return {
        "status": result['data']['status'],
        "result": result['data'].get('result')
    }

List Functions:

def main(params, user_data, sdk_client):
    functions = sdk_client.functions.list(limit=10)
    return {"functions": functions['data']}

Secrets Module​

Access secrets securely without hardcoding credentials.

Methods​

sdk_client.secrets.get_secret(key, app=None, tags=None)
sdk_client.secrets.get_secrets(keys, app=None)
sdk_client.secrets.list_secrets(search=None, app=None, tags=None, secret_type=None, page=1, page_size=20)
sdk_client.secrets.update(key, value)

Examples​

Get Single Secret:

def main(params, user_data, sdk_client):
    api_key = sdk_client.secrets.get_secret("SENDGRID_API_KEY")

    # Use the secret
    send_email_with_api_key(api_key["value"])

    return {"sent": True}

Get Multiple Secrets:

def main(params, user_data, sdk_client):
    secrets = sdk_client.secrets.get_secrets([
        "SENDGRID_API_KEY",
        "STRIPE_SECRET_KEY",
        "AWS_ACCESS_KEY"
    ])

    sendgrid_key = secrets["SENDGRID_API_KEY"]["value"]
    stripe_key = secrets["STRIPE_SECRET_KEY"]["value"]

    return {"keys_loaded": True}

List Secrets:

def main(params, user_data, sdk_client):
    secrets = sdk_client.secrets.list_secrets(
        search="API",
        tags=["production"]
    )
    return {"secrets": secrets['data']}

⚠️ Critical Rules​

  • Never hardcode secrets in function code
  • Always use get_secret() to retrieve credentials
  • Handle missing secrets gracefully with try/except
  • Never log secret values - only log keys

Analytics Module​

Execute saved analytics queries with parameters.

Methods​

sdk_client.analytics.execute(query_slug, params=None)

Returns​

{
    "data": [...],           # Query results
    "total": 100,            # Row count
    "execution_key": "..."   # Execution tracking key
}

Examples​

Simple Query Execution:

def main(params, user_data, sdk_client):
    result = sdk_client.analytics.execute("monthly-revenue")
    return {"revenue": result["data"]}

Query with Parameters:

def main(params, user_data, sdk_client):
    result = sdk_client.analytics.execute(
        "user-signups",
        params={
            "start_date": "2024-01-01",
            "end_date": "2024-12-31",
            "group_by": "month"
        }
    )
    return {
        "signups": result["data"],
        "total": result["total"]
    }

Query with Dynamic Filters:

def main(params, user_data, sdk_client):
    region = params.get("region", "US")
    category = params.get("category", "all")

    result = sdk_client.analytics.execute(
        "sales-by-region",
        params={
            "region": region,
            "product_category": category
        }
    )
    return result["data"]

Notes​

  • Queries must be created first (use MCP analytics tools or API)
  • Supports Jinja2 templating in query definitions
  • Uses secrets for database credentials
  • Results are cached based on execution_key

Storage Module​

Upload, download, and manage files in storage buckets.

Methods​

sdk_client.storage.list_buckets()
sdk_client.storage.create_bucket(name, description=None)
sdk_client.storage.get_bucket(name)
sdk_client.storage.update_bucket(name, data)
sdk_client.storage.delete_bucket(name)

# Bucket operations (chainable)
sdk_client.storage.from_(bucket_name).upload(files, paths)
sdk_client.storage.from_(bucket_name).download(path)
sdk_client.storage.from_(bucket_name).list()
sdk_client.storage.from_(bucket_name).delete(paths)

Examples​

Upload File:

def main(params, user_data, sdk_client):
    with open("report.pdf", "rb") as f:
        result = sdk_client.storage.from_("documents").upload(
            files=[("report.pdf", f)],
            paths=["uploads/report.pdf"]
        )

    return {"uploaded": True, "path": result[0]["path"]}

Download File:

def main(params, user_data, sdk_client):
    file_path = params.get("file_path")
    file_data = sdk_client.storage.from_("documents").download(file_path)

    # Process file_data
    return {"downloaded": True, "size": len(file_data)}

List Files:

def main(params, user_data, sdk_client):
    files = sdk_client.storage.from_("documents").list()
    return {"files": files, "total": len(files)}

Delete Files:

def main(params, user_data, sdk_client):
    paths = params.get("paths", [])
    sdk_client.storage.from_("documents").delete(paths)
    return {"deleted": True, "count": len(paths)}

Create Bucket:

def main(params, user_data, sdk_client):
    bucket = sdk_client.storage.create_bucket(
        name="user-uploads",
        description="User uploaded documents"
    )
    return {"created": True, "bucket": bucket}

Users Module​

Manage users and retrieve user information.

Methods​

sdk_client.users.list_users(search=None, is_active=None, is_staff=None, roles=None, page=1, page_size=20)
sdk_client.users.get_user(username)
sdk_client.users.create_user(username, email, password, ...)
sdk_client.users.update_user(username, data)
sdk_client.users.delete_user(username)

Examples​

Get User:

def main(params, user_data, sdk_client):
    username = params.get("username")
    user = sdk_client.users.get_user(username)
    return {"user": user}

List Users:

def main(params, user_data, sdk_client):
    users = sdk_client.users.list_users(
        search="alice",
        is_active=True,
        page=1,
        page_size=10
    )
    return {"users": users['data'], "total": users['total']}

Create User:

def main(params, user_data, sdk_client):
    new_user = sdk_client.users.create_user(
        username=params.get("username"),
        email=params.get("email"),
        password=params.get("password"),
        first_name=params.get("first_name"),
        last_name=params.get("last_name")
    )
    return {"created": True, "user": new_user}

Update User:

def main(params, user_data, sdk_client):
    username = params.get("username")
    updated_user = sdk_client.users.update_user(username, {
        "first_name": params.get("first_name"),
        "is_active": True
    })
    return {"updated": True, "user": updated_user}

Policy Module​

Check authorization permissions and filter allowed resources.

Methods​

sdk_client.policy.check_resources(resources, principal, aux_data=None)
sdk_client.policy.filter_allowed(principal, resources, actions)

Examples​

Check Permissions:

def main(params, user_data, sdk_client):
    result = sdk_client.policy.check_resources(
        resources=[{
            "resource": {
                "kind": "datatable",
                "id": "users"
            },
            "actions": ["read", "write"]
        }],
        principal={
            "id": user_data["id"],
            "roles": ["user"],
            "attr": {"department": "engineering"}
        }
    )

    can_read = result["results"][0]["actions"]["read"] == "EFFECT_ALLOW"
    can_write = result["results"][0]["actions"]["write"] == "EFFECT_ALLOW"

    return {"can_read": can_read, "can_write": can_write}

Filter Allowed Resources:

def main(params, user_data, sdk_client):
    all_tables = [
        {"kind": "datatable", "id": "users"},
        {"kind": "datatable", "id": "orders"},
        {"kind": "datatable", "id": "admin_logs"}
    ]

    allowed = sdk_client.policy.filter_allowed(
        principal={"id": user_data["id"], "roles": ["user"]},
        resources=all_tables,
        actions=["read", "write"]
    )

    return {"allowed_tables": allowed}

Auth Module​

ℹ️ Usually not needed in APP mode functions

The sdk_client is already authenticated with the current user context.

Methods​

sdk_client.auth.signInWithPassword(username, password)
sdk_client.auth.signInWithToken(token, token_type)  # token_type: 'jwt', 'api_key', 'session_token'
sdk_client.auth.refreshToken(refresh_token)
sdk_client.auth.signOut()
sdk_client.auth.is_authenticated  # Property

In most cases, you don't need auth methods because sdk_client is already authenticated:

def main(params, user_data, sdk_client):
    # ✅ sdk_client is already authenticated as the calling user
    # ✅ Just use it directly
    users = sdk_client.database.query("users").execute()
    return users

Advanced Usage (If Needed)​

If you need to authenticate as a different user or service account:

def main(params, user_data, sdk_client):
    # Get service account API key from secrets
    service_key = sdk_client.secrets.get_secret("SERVICE_ACCOUNT_KEY")

    # Authenticate as service account
    service_client = sdk_client.auth.signInWithToken(
        token=service_key["value"],
        token_type='api_key'
    )

    # Use service_client for privileged operations
    admin_data = service_client.database.query("admin_logs").execute()

    return {"admin_data": admin_data}

External SDK Usage​

Auth methods are primarily for external SDK usage (scripts, CLIs, standalone apps):

# External Python script (NOT an APP mode function)
from taruvi import Client

client = Client(api_url='https://api.taruvi.cloud', app_slug='my-app')

# Authenticate
auth_client = client.auth.signInWithPassword(
    username='user@example.com',
    password='password'
)

# Use authenticated client
users = auth_client.database.query("users").execute()

Complete Examples​

Email Notification Function​

def main(params, user_data, sdk_client):
    # Get parameters
    recipient_id = params.get("user_id")
    message = params.get("message")

    # Fetch recipient from database
    recipient = sdk_client.database.get("users", recipient_id)

    # Get SendGrid API key from secrets
    sendgrid = sdk_client.secrets.get_secret("SENDGRID_API_KEY")

    # Call email function
    result = sdk_client.functions.execute(
        "send-email",
        params={
            "to": recipient["email"],
            "subject": "Notification",
            "body": message,
            "api_key": sendgrid["value"]
        }
    )

    return {
        "sent": True,
        "recipient": recipient["email"],
        "message_id": result.get("message_id")
    }

Data Processing Pipeline​

def main(params, user_data, sdk_client):
    # Get input data
    dataset_id = params.get("dataset_id")

    # Fetch dataset
    dataset = sdk_client.database.get("datasets", dataset_id)

    # Check permissions
    can_process = sdk_client.policy.check_resources(
        resources=[{
            "resource": {"kind": "dataset", "id": dataset_id},
            "actions": ["process"]
        }],
        principal={"id": user_data["id"], "roles": ["analyst"]}
    )

    if can_process["results"][0]["actions"]["process"] != "EFFECT_ALLOW":
        return {"error": "Permission denied"}

    # Execute processing function asynchronously
    result = sdk_client.functions.execute(
        "process-dataset",
        params={"dataset_id": dataset_id},
        is_async=True
    )

    # Upload results to storage
    task_id = result['invocation']['celery_task_id']

    return {
        "processing": True,
        "task_id": task_id,
        "dataset": dataset["name"]
    }

Report Generation Function​

def main(params, user_data, sdk_client):
    # Get date range from params
    start_date = params.get("start_date")
    end_date = params.get("end_date")

    # Execute analytics query
    revenue_data = sdk_client.analytics.execute(
        "monthly-revenue",
        params={
            "start_date": start_date,
            "end_date": end_date,
            "group_by": "month"
        }
    )

    # Get user signups
    signup_data = sdk_client.analytics.execute(
        "user-signups",
        params={
            "start_date": start_date,
            "end_date": end_date
        }
    )

    # Generate report content
    report_content = f"""
    Revenue Report
    Period: {start_date} to {end_date}

    Total Revenue: ${sum(r['revenue'] for r in revenue_data['data'])}
    New Signups: {signup_data['total']}
    """

    # Upload report to storage
    with open("/tmp/report.txt", "w") as f:
        f.write(report_content)

    with open("/tmp/report.txt", "rb") as f:
        sdk_client.storage.from_("reports").upload(
            files=[("report.txt", f)],
            paths=[f"reports/{start_date}_to_{end_date}.txt"]
        )

    return {
        "report_generated": True,
        "total_revenue": sum(r['revenue'] for r in revenue_data['data']),
        "new_signups": signup_data['total']
    }

Critical Rules​

❌ Don't Do This​

  • Don't create new Client() - Use provided sdk_client
  • Don't call auth methods - sdk_client is already authenticated
  • Don't hardcode secrets - Always use secrets.get_secret()
  • Don't skip error handling - User input can be anything
  • Don't assume records exist - Always check before operations

✅ Do This​

  • Validate all params - Check for required fields
  • Handle exceptions gracefully - Use try/except blocks
  • Return meaningful data - Include success flags and error messages
  • Use secrets for credentials - Never hardcode API keys
  • Check permissions - Use policy module when needed
  • Log important events - For debugging and audit trails

Error Handling Pattern​

def main(params, user_data, sdk_client):
    try:
        # Validate params
        user_id = params.get("user_id")
        if not user_id:
            return {"success": False, "error": "user_id is required"}

        # Fetch user
        user = sdk_client.database.get("users", user_id)

        # Get secret
        api_key = sdk_client.secrets.get_secret("API_KEY")

        # Perform operation
        result = perform_operation(user, api_key["value"])

        return {"success": True, "result": result}

    except Exception as e:
        return {
            "success": False,
            "error": str(e),
            "error_type": type(e).__name__
        }