vaif-client 0.2.0

Official VAIF Studio client library for Python

vaif-client

The official Python SDK for VAIF Studio - the AI-powered backend platform. Async-first with a synchronous wrapper for Django, Flask, and scripts.

Installation

pip install vaif-client

With realtime support (WebSocket subscriptions):

pip install vaif-client[realtime]

With all extras:

pip install vaif-client[all]

Quick Start

Async (FastAPI, async scripts)

from vaif import create_client

async with create_client(
    project_id="your-project-id",
    api_key="your-api-key",
) as vaif:
    # Query data
    result = await vaif.db.from_("users").select("*").eq("active", True).execute()
    users = result.data

    # Upload a file
    upload = await vaif.storage.from_("avatars").upload("user.png", file_bytes)

    # Invoke a function
    response = await vaif.functions.invoke("send-email", body={"to": "[e-mail]"})

Sync (Django, Flask, scripts)

from vaif import create_sync_client

with create_sync_client(
    project_id="your-project-id",
    api_key="your-api-key",
) as vaif:
    # Same API, no await needed
    result = vaif.db.from_("users").select("*").eq("active", True).execute()
    users = result.data

    upload = vaif.storage.from_("avatars").upload("user.png", file_bytes)

Modules

Module	Description
vaif.db	Database operations with fluent query builder
vaif.auth	Authentication (email, OAuth, OTP, MFA)
vaif.realtime	WebSocket subscriptions and presence
vaif.storage	File storage with CDN
vaif.functions	Edge function invocation
vaif.ai	AI-powered code generation and chat
vaif.typegen	Python type generation from DB schema

Database

Query Builder

# Select with filters
result = await vaif.db.from_("users") \
    .select("id, name, email") \
    .eq("active", True) \
    .order("created_at", ascending=False) \
    .limit(10) \
    .execute()

for user in result.data:
    print(user["name"])

# Get single record
result = await vaif.db.from_("users") \
    .eq("id", "user-123") \
    .single()

user = result.data  # raises if not found

# Get single or None
result = await vaif.db.from_("users") \
    .eq("email", "[e-mail]") \
    .maybe_single()

user = result.data  # None if not found

# Insert
result = await vaif.db.from_("users") \
    .insert({"email": "[e-mail]", "name": "New User"}) \
    .execute()

# Batch insert
result = await vaif.db.from_("users") \
    .insert([
        {"email": "[e-mail]", "name": "User 1"},
        {"email": "[e-mail]", "name": "User 2"},
    ]) \
    .execute()

# Upsert
result = await vaif.db.from_("users") \
    .upsert(
        {"email": "[e-mail]", "name": "Updated"},
        on_conflict="email",
    ) \
    .execute()

# Update
result = await vaif.db.from_("users") \
    .update({"name": "New Name"}) \
    .eq("id", "user-123") \
    .execute()

# Delete
result = await vaif.db.from_("users") \
    .delete() \
    .eq("id", "user-123") \
    .execute()

# Return specific columns after mutation
result = await vaif.db.from_("users") \
    .insert({"email": "[e-mail]"}) \
    .returning("id, email") \
    .execute()

Filters

vaif.db.from_("posts") \
    .eq("status", "published")           # column = value
    .neq("author_id", None)              # column != value
    .gt("views", 100)                    # column > value
    .gte("rating", 4.0)                  # column >= value
    .lt("price", 50)                     # column < value
    .lte("stock", 10)                    # column <= value
    .like("title", "%Python%")           # LIKE (case-sensitive)
    .ilike("title", "%python%")          # ILIKE (case-insensitive)
    .is_("deleted_at", None)             # IS NULL
    .in_("status", ["draft", "review"])  # IN (...)
    .contains("tags", ["py", "ai"])      # @> (array contains)
    .contained_by("tags", ["py", "ai", "ml"])  # <@ (contained by)
    .overlaps("tags", ["py", "go"])      # && (array overlap)
    .text_search("body", "search query") # Full-text search
    .not_.eq("role", "admin")            # Negate any filter
    .or_("status.eq.draft,status.eq.review")   # OR conditions
    .order("created_at", ascending=False)
    .limit(20)
    .offset(40)
    .range(0, 9)                         # Shorthand for offset + limit
    .count("exact")                      # Include total count
    .execute()

Raw SQL and RPC

# Raw SQL (admin only)
result = await vaif.db.raw(
    "SELECT * FROM users WHERE created_at > $1",
    ["2024-01-01"],
)

# Stored procedure
result = await vaif.db.rpc("get_user_stats", {"user_id": "123"})

Authentication

# Sign up
result = await vaif.auth.sign_up(
    email="[e-mail]",
    password="secure-password",
    data={"name": "John"},
)

# Sign in with password
result = await vaif.auth.sign_in_with_password(
    email="[e-mail]",
    password="secure-password",
)

# Sign in with OAuth
result = await vaif.auth.sign_in_with_oauth(
    provider="google",
    redirect_to="https://myapp.com/callback",
)
print(f"Redirect to: {result.data.url}")

# Sign in with OTP
await vaif.auth.sign_in_with_otp(email="[e-mail]")

# Verify OTP
result = await vaif.auth.verify_otp(
    email="[e-mail]",
    token="123456",
    type="magiclink",
)

# Get current session and user
session = await vaif.auth.get_session()
user = await vaif.auth.get_user()

# Listen to auth state changes
vaif.auth.on_auth_state_change(lambda event, session: print(f"Auth: {event}"))

# Sign out
await vaif.auth.sign_out()

# Reset password
await vaif.auth.reset_password_for_email(email="[e-mail]")

# Refresh session
await vaif.auth.refresh_session()

# Exchange code for session (OAuth/PKCE)
result = await vaif.auth.exchange_code_for_session(code)

Sync Auth

vaif = create_sync_client(project_id="...", api_key="...")

result = vaif.auth.sign_in_with_password(email="[e-mail]", password="password")
user = vaif.auth.get_user()
vaif.auth.sign_out()

Realtime

Requires pip install vaif-client[realtime]

# Connect
await vaif.realtime.connect()

# Subscribe to database changes
channel = vaif.realtime.channel("my-channel")
channel.on(
    "postgres_changes",
    {"event": "INSERT", "schema": "public", "table": "messages"},
    lambda payload: print("New message:", payload),
)
await channel.subscribe()

# Broadcast
channel.broadcast("typing", {"user_id": "user-123"})

# Presence
channel.track({"user_id": "user-123", "status": "online"})
state = channel.presence_state()

# Cleanup
await channel.unsubscribe()
await vaif.realtime.disconnect()

Storage

# Upload file
result = await vaif.storage.from_("avatars").upload(
    "user-123/avatar.jpg",
    file_bytes,
    {"content_type": "image/jpeg", "upsert": True},
)

# Download file
result = await vaif.storage.from_("avatars").download("user-123/avatar.jpg")
file_data = result.data

# Get public URL
url = vaif.storage.from_("avatars").get_public_url("user-123/avatar.jpg")

# Create signed URL (temporary access)
result = await vaif.storage.from_("private").create_signed_url(
    "document.pdf",
    {"expires_in": 3600},
)

# List files
result = await vaif.storage.from_("avatars").list("user-123/")

# Move/copy/delete
await vaif.storage.from_("avatars").move("old.jpg", "new.jpg")
await vaif.storage.from_("avatars").copy("source.jpg", "dest.jpg")
await vaif.storage.from_("avatars").remove(["old.jpg", "another.jpg"])

# Bucket management
buckets = await vaif.storage.list_buckets()
await vaif.storage.create_bucket(name="docs", public=False)
await vaif.storage.delete_bucket("old-bucket")

Sync Storage

vaif = create_sync_client(project_id="...", api_key="...")

result = vaif.storage.from_("avatars").upload("user.jpg", file_bytes)
url = vaif.storage.from_("avatars").get_public_url("user.jpg")

Edge Functions

# Invoke a function
result = await vaif.functions.invoke(
    "send-email",
    body={"to": "[e-mail]", "subject": "Hello"},
)

if result.error:
    print(f"Error: {result.error.message}")
else:
    print(f"Result: {result.data}")

# Get function URL
url = vaif.functions.create_url("send-email")

# List deployed functions
result = await vaif.functions.list()

# Get function details
result = await vaif.functions.get("send-email")

Sync Functions

vaif = create_sync_client(project_id="...", api_key="...")

result = vaif.functions.invoke("send-email", body={"to": "[e-mail]"})

AI Generation

# Generate database schema
result = await vaif.ai.generate_schema({
    "prompt": "Create a blog with posts, authors, and comments",
    "format": "sql",
    "include_relations": True,
})
print(result.data.sql)

# Generate edge function
result = await vaif.ai.generate_function({
    "prompt": "Create a function that resizes uploaded images",
    "name": "resize-image",
})

# Generate API endpoint
result = await vaif.ai.generate_endpoint({
    "prompt": "Get user profile with recent posts",
    "method": "GET",
    "path": "/users/:id/profile",
})

# AI chat
result = await vaif.ai.chat({
    "messages": [
        {"role": "user", "content": "How do I add pagination to my query?"},
    ],
})
print(result.data.content)

# Code review
result = await vaif.ai.review(code, language="python", focus=["security"])

# Explain code
result = await vaif.ai.explain(code, language="python")

# Generate types
result = await vaif.ai.generate_types(schema_string, format="typescript")

# Generate migration
result = await vaif.ai.generate_migration(
    current_schema=old_schema,
    target_schema=new_schema,
    database="postgresql",
)

# Check credits
result = await vaif.ai.get_credits()
print(f"Balance: {result.data.balance} credits")

Sync AI

vaif = create_sync_client(project_id="...", api_key="...")

result = vaif.ai.generate_schema({"prompt": "Blog with posts and comments"})
result = vaif.ai.chat({"messages": [{"role": "user", "content": "Help me"}]})

Configuration

from vaif import create_client, RetryConfig

vaif = create_client(
    # Required
    project_id="your-project-id",
    api_key="your-api-key",

    # API endpoints
    api_url="https://api.vaif.studio",           # Custom API URL
    realtime_url="wss://realtime.vaif.studio",   # Custom WebSocket URL

    # Request options
    timeout=30000,                                # Timeout in milliseconds
    headers={"x-custom": "value"},               # Custom headers

    # Auth
    auto_refresh_token=True,                      # Auto-refresh JWT
    persist_session=True,                         # Persist to storage

    # Retry with exponential backoff
    retry=RetryConfig(
        max_retries=3,                            # Max retry attempts
        retry_delay=1.0,                          # Initial delay (seconds)
        max_retry_delay=30.0,                     # Max delay (seconds)
        backoff_multiplier=2.0,                   # Exponential multiplier
        retry_on=[429, 500, 502, 503, 504],      # Status codes to retry
        retry_on_network_error=True,              # Retry on network errors
    ),

    # Debug
    debug=False,
)

Error Handling

from vaif import (
    VaifError,
    VaifAuthError,
    VaifDatabaseError,
    VaifStorageError,
    VaifFunctionError,
    VaifAIError,
    VaifNetworkError,
    VaifTimeoutError,
    VaifRateLimitError,
    VaifValidationError,
    VaifNotFoundError,
    VaifConflictError,
)

# Pattern 1: Check error in response (recommended)
result = await vaif.db.from_("users").eq("id", "123").single()

if result.error:
    print(f"Error: {result.error.message} (code: {result.error.code})")
else:
    print(f"User: {result.data}")

# Pattern 2: Catch exceptions
try:
    result = await vaif.auth.sign_in_with_password(
        email="[e-mail]",
        password="wrong",
    )
except VaifAuthError as e:
    print(f"Auth failed: {e.message}")
except VaifRateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except VaifTimeoutError:
    print("Request timed out")
except VaifNetworkError as e:
    print(f"Network error: {e.message}")
except VaifError as e:
    print(f"VAIF error: {e.message} (code={e.code}, status={e.status})")
    print(f"Request ID: {e.request_id}")  # For support

Exception Hierarchy

VaifError (base)
  ├── VaifAuthError        (401/403)
  ├── VaifDatabaseError    (query errors)
  ├── VaifStorageError     (upload/download errors)
  ├── VaifFunctionError    (invocation errors, has function_name)
  ├── VaifAIError          (generation errors)
  ├── VaifValidationError  (400)
  ├── VaifRateLimitError   (429, has retry_after)
  ├── VaifNotFoundError    (404)
  ├── VaifConflictError    (409)
  └── VaifNetworkError
        └── VaifTimeoutError

Framework Integration

FastAPI

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from vaif import create_client, VaifClient

vaif: VaifClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global vaif
    vaif = create_client(project_id="...", api_key="...")
    yield
    await vaif.close()

app = FastAPI(lifespan=lifespan)

@app.get("/users")
async def get_users():
    result = await vaif.db.from_("users").select("*").execute()
    return result.data

Django

# settings.py
from vaif import create_sync_client

vaif = create_sync_client(
    project_id="your-project-id",
    api_key="your-api-key",
)

# views.py
from django.http import JsonResponse
from myproject.settings import vaif

def users_view(request):
    result = vaif.db.from_("users").select("*").execute()
    return JsonResponse({"users": result.data})

Flask

from flask import Flask, jsonify
from vaif import create_sync_client

app = Flask(__name__)
vaif = create_sync_client(project_id="...", api_key="...")

@app.route("/users")
def get_users():
    result = vaif.db.from_("users").select("*").execute()
    return jsonify(result.data)

Type Safety

The SDK ships with Pydantic v2 models and py.typed marker for full type checker support:

from vaif.types import User, Session, QueryResult, ApiResponse

# All responses are typed
result: QueryResult[list[dict]] = await vaif.db.from_("users").select("*").execute()

Requirements

• Python 3.9+
• httpx >= 0.25.0
• pydantic >= 2.0.0
• websockets >= 12.0 (optional, for realtime)

Related Packages

• @vaiftechnologies/vaif-client (npm) - JavaScript/TypeScript SDK
• @vaiftech/auth (npm) - Standalone auth client
• @vaiftech/react (npm) - React hooks

License

MIT