Deployment Overview

# Example: Orchestrate in Pixeltable, export to external systems
import pixeltable as pxt
from pixeltable.functions.video import extract_audio
from pixeltable.functions.openai import transcriptions
from pixeltable.functions.video import frame_iterator
import psycopg2
from datetime import datetime

# Setup: Define Pixeltable orchestration pipeline
pxt.create_dir('video_processing', if_exists='ignore')

videos = pxt.create_table(
    'video_processing/videos',
    {'video': pxt.Video, 'uploaded_at': pxt.Timestamp}
)

# Computed columns for orchestration
videos.add_computed_column(
    audio=extract_audio(videos.video, format='mp3')
)
videos.add_computed_column(
    transcript=transcriptions(audio=videos.audio, model='whisper-1')
)

# Optional: Add LLM-based summary
from pixeltable.functions.openai import chat_completions
videos.add_computed_column(
    summary=chat_completions(
        messages=[{'role': 'user', 'content': f"Summarize: {videos.transcript.text}"}],
        model='gpt-4o-mini'
    )
)

# Extract frames for analysis
frames = pxt.create_view(
    'video_processing/frames',
    videos,
    iterator=frame_iterator(video=videos.video, fps=1.0)
)

# Insert video for processing
videos.insert([{'video': 's3://bucket/video.mp4', 'uploaded_at': datetime.now()}])

# Export structured results to external RDBMS
conn = psycopg2.connect("postgresql://...")
cursor = conn.cursor()

for row in videos.select(videos.video, videos.transcript).collect():
    cursor.execute(
        "INSERT INTO video_metadata (video_url, transcript_json) VALUES (%s, %s)",
        (row['video'], row['transcript'])
    )
conn.commit()

# Example: FastAPI endpoints backed by Pixeltable
from pydantic import BaseModel
from fastapi import FastAPI, UploadFile
from datetime import datetime
import pixeltable as pxt

app = FastAPI()
docs_table = pxt.get_table('myapp/documents')  # Has computed columns: embedding, summary

class SearchResult(BaseModel):
    document: str
    summary: str | None
    similarity: float

@app.post("/documents/upload")
def upload_document(file: UploadFile):
    status = docs_table.insert([{
        'document': file.filename,
        'uploaded_at': datetime.now()
    }])
    return {"rows_inserted": status.num_rows}

@app.get("/documents/search")
def search_documents(query: str, limit: int = 10) -> list[SearchResult]:
    sim = docs_table.embedding.similarity(string=query)
    results = docs_table.select(
        docs_table.document,
        docs_table.summary,
        similarity=sim
    ).order_by(sim, asc=False).limit(limit).collect()

    return list(results.to_pydantic(SearchResult))

@app.get("/documents/{doc_id}")
def get_document(doc_id: int):
    result = docs_table.where(docs_table._rowid == doc_id).collect()
    return result[0] if len(result) > 0 else {"error": "Not found"}