vaif-client 0.2.1
Official VAIF Studio client library for Python
Requires Python: >=3.9
Keywords: api, baas, backend, database, realtime, supabase-alternative, vaif, vaif-studio

# vaif-client

[![PyPI version](https://img.shields.io/pypi/v/vaif-client)](https://pypi.org/project/vaif-client/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)
[![Python](https://img.shields.io/pypi/pyversions/vaif-client)](https://pypi.org/project/vaif-client/)
[![Typed](https://img.shields.io/badge/typing-Pydantic%20v2-blue)](https://docs.pydantic.dev/)

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

## Installation

```bash
pip install vaif-client
```

With realtime support (WebSocket subscriptions):

```bash
pip install vaif-client[realtime]
```

With all extras:

```bash
pip install vaif-client[all]
```

## Quick Start

### Async (FastAPI, async scripts)

```python
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": "user@example.com"})
```

### Sync (Django, Flask, scripts)

```python
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

```python
# 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", "test@example.com") \
    .maybe_single()

user = result.data  # None if not found

# Insert
result = await vaif.db.from_("users") \
    .insert({"email": "new@example.com", "name": "New User"}) \
    .execute()

# Batch insert
result = await vaif.db.from_("users") \
    .insert([
        {"email": "user1@example.com", "name": "User 1"},
        {"email": "user2@example.com", "name": "User 2"},
    ]) \
    .execute()

# Upsert
result = await vaif.db.from_("users") \
    .upsert(
        {"email": "user@example.com", "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": "user@test.com"}) \
    .returning("id, email") \
    .execute()
```

### Filters

```python
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

```python
# 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

```python
# Sign up
result = await vaif.auth.sign_up(
    email="user@example.com",
    password="secure-password",
    data={"name": "John"},
)

# Sign in with password
result = await vaif.auth.sign_in_with_password(
    email="user@example.com",
    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="user@example.com")

# Verify OTP
result = await vaif.auth.verify_otp(
    email="user@example.com",
    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="user@example.com")

# Refresh session
await vaif.auth.refresh_session()

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

### Sync Auth

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

result = vaif.auth.sign_in_with_password(email="user@example.com", password="password")
user = vaif.auth.get_user()
vaif.auth.sign_out()
```

## Realtime

> Requires `pip install vaif-client[realtime]`

```python
# 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

```python
# 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

```python
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

```python
# Invoke a function
result = await vaif.functions.invoke(
    "send-email",
    body={"to": "user@example.com", "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

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

result = vaif.functions.invoke("send-email", body={"to": "user@example.com"})
```

## AI Generation

```python
# 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

```python
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

```python
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

```python
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="user@example.com",
        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

```python
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

```python
# 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

```python
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:

```python
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)](https://www.npmjs.com/package/@vaiftechnologies/vaif-client) - JavaScript/TypeScript SDK
- [@vaiftech/auth (npm)](https://www.npmjs.com/package/@vaiftech/auth) - Standalone auth client
- [@vaiftech/react (npm)](https://www.npmjs.com/package/@vaiftech/react) - React hooks

## License

MIT