Infrastructure Setup

​

Code Organization

• Tables, views, computed columns, indexes, and agent-internal @pxt.query functions
• Flat module with if_exists='ignore' for idempotency (no setup() wrapper, no _initialized flag)
• Run once before starting workers: python schema.py

• Call pxt.get_table() directly to get table handles
• Define router-facing @pxt.query functions next to the routes that use them
• No import schema needed; tables already exist from the init step

• Externalizes model IDs, API keys, thresholds, connection strings
• Uses environment variables (.env + python-dotenv) or secrets management
• Never hardcodes secrets

# schema.py — creates tables (flat, idempotent, no queries for routers)
import pixeltable as pxt
import config

pxt.create_dir(config.APP_NAMESPACE, if_exists='ignore')

docs = pxt.create_table(
    f'{config.APP_NAMESPACE}/documents',
    {'document': pxt.Document, 'metadata': pxt.Json, 'timestamp': pxt.Timestamp},
    if_exists='ignore',
)

# ---

# routers/data.py — queries live next to the routes that use them
import pixeltable as pxt
from pixeltable.serving import FastAPIRouter
import config

router = FastAPIRouter(prefix="/api/data", tags=["data"])
docs = pxt.get_table(f'{config.APP_NAMESPACE}/documents')

@pxt.query
def list_documents():
    return docs.select(docs.title, docs.document).order_by(docs.title)

router.add_query_route(path="/documents", query=list_documents, method="get")

​

Project Structure

project/
├── config.py              # Environment variables, model IDs, API keys
├── functions.py           # Custom UDFs (imported as modules)
├── schema.py              # Schema definition (tables, views, indexes)
├── main.py                # FastAPI app, mounts routers
├── routers/
│   ├── data.py            # CRUD routes + queries for data pipeline
│   ├── search.py          # Search routes + queries
│   └── agent.py           # Agent routes (declarative + hand-written)
├── pyproject.toml         # Dependencies and pxt serve config
└── .env                   # Secrets (gitignored)

import os

ENV = os.getenv('ENVIRONMENT', 'dev')
APP_NAMESPACE = f'{ENV}_myapp'

# Model Configuration
EMBEDDING_MODEL = os.getenv('EMBEDDING_MODEL', 'intfloat/e5-large-v2')
OPENAI_MODEL = os.getenv('OPENAI_MODEL', 'gpt-4o-mini')

# Storage
MEDIA_STORAGE_BUCKET = os.getenv('MEDIA_STORAGE_BUCKET')

# Prompts
RAG_SYSTEM_PROMPT = """You are a helpful assistant. Use the provided context to answer questions."""

import pixeltable as pxt

@pxt.udf
def format_prompt(context: list, question: str) -> str:
    """Format RAG prompt with context."""
    context_str = "\n".join([doc['text'] for doc in context])
    return f"Context:\n{context_str}\n\nQuestion: {question}"

@pxt.udf(resource_pool='request-rate:my_service')
async def call_custom_model(prompt: str) -> dict:
    """Call self-hosted model endpoint."""
    # Your custom logic here
    return {"response": "..."}

import pixeltable as pxt
from pixeltable.functions.huggingface import sentence_transformer
import config

pxt.create_dir(config.APP_NAMESPACE, if_exists='ignore')

docs = pxt.create_table(
    f'{config.APP_NAMESPACE}/documents',
    {'document': pxt.Document, 'metadata': pxt.Json, 'timestamp': pxt.Timestamp},
    if_exists='ignore',
)

docs.add_computed_column(
    embedding=sentence_transformer(docs.document, model_id=config.EMBEDDING_MODEL),
    if_exists='ignore',
)

docs.add_embedding_index('embedding', idx_name='docs_embed', metric='cosine', if_exists='ignore')

import pixeltable as pxt
from pixeltable.serving import FastAPIRouter
import config

router = FastAPIRouter(prefix="/api/search", tags=["search"])
docs = pxt.get_table(f'{config.APP_NAMESPACE}/documents')

@pxt.query
def search_documents(query_text: str, limit: int = 5):
    sim = docs.embedding.similarity(string=query_text)
    return docs.order_by(sim, asc=False).limit(limit).select(docs.document, sim)

router.add_query_route(path="/documents", query=search_documents)

from fastapi import FastAPI
from routers import data, search

app = FastAPI()
app.include_router(data.router)
app.include_router(search.router)

​

Storage Architecture

• PostgreSQL stores only file paths/URLs, never raw media data.
• Inserted local files: path stored, original file remains in place.
• Inserted URLs: URL stored, file downloaded to File Cache on first access.
• Generated media (computed columns): saved to Media Store (default: local, configurable to S3/GCS/Azure per-column).
• File Cache size: configure via file_cache_size_g in ~/.pixeltable/config.toml. See configuration guide

# ~/.pixeltable/config.toml
file_cache_size_g = 50  # 50 GB cache

​

References, Not Copies

• No data duplication — you don’t pay for storage twice.
• Schema changes don’t require re-upload — add a column, not a migration script.
• Works with existing storage — point Pixeltable at your current buckets.

• Pixeltable storage can be ephemeral (re-computable).
• Processing results exported to external RDBMS via export_sql and media to blob storage via destination.
• Reference input media from S3/GCS/Azure URIs.

• Pixeltable IS the RDBMS (embedded PostgreSQL, not replaceable).
• Requires persistent volume at ~/.pixeltable (pgdata, media, file_cache).
• Media Store configurable to S3/GCS/Azure buckets for generated files.

• Same persistent storage as Full Backend.
• API routes declared in pyproject.toml, no hand-written endpoint code.

PIXELTABLE_INPUT_MEDIA_DEST=s3://your-bucket/input    # or gs:// or az://
PIXELTABLE_OUTPUT_MEDIA_DEST=s3://your-bucket/output

​

Dependency Management

[project]
name = "my-pixeltable-app"
requires-python = ">=3.10"
dependencies = [
    "pixeltable>=0.6.2",
    "fastapi[standard]>=0.115.0",
    "python-dotenv>=1.0.1",
    "sentence-transformers>=3.3.0",  # If using embedding indexes
]

• Use pyproject.toml with uv or pip for dependency management
• Include integration packages (e.g., openai, sentence-transformers)
• Test updates in staging before production

​

Data Interoperability

• CSV, Excel, JSON: pxt.io.import_csv(), pxt.io.import_excel(), pxt.io.import_json()
• Parquet: pxt.io.import_parquet()
• SQL databases: pxt.io.import_sql() — see SQL import/export cookbook
• Pandas DataFrames: table.insert(df) or pxt.create_table(source=df)
• Hugging Face Datasets: pxt.io.import_huggingface_dataset()

• CSV: pxt.io.export_csv(table, path) for tabular data
• JSON: pxt.io.export_json(table, path) for structured data
• Parquet: pxt.io.export_parquet(table, path) for data warehousing
• SQL databases: pxt.io.export_sql(table, ...) — see SQL import/export cookbook
• Iceberg: pxt.io.export_iceberg(table, ...) — see analytics export cookbook
• LanceDB: pxt.io.export_lancedb(table, db_uri, table_name) for vector databases
• PyTorch: table.to_pytorch_dataset() for ML training pipelines
• COCO: table.to_coco_dataset() for computer vision
• Pandas: table.collect().to_pandas() for analysis

# Export query results to Parquet
import pixeltable as pxt

docs_table = pxt.get_table('myapp/documents')
results = docs_table.where(docs_table.timestamp > '2024-01-01')
pxt.io.export_parquet(results, '/data/exports/recent_docs.parquet')