Reference architecture

User-uploaded media pipeline reference architecture

User-uploaded media is an HTTP route, a typed row, optional exact-hash and near-duplicate checks, models that run on insert, and an index that stays current. The glue stack is FastAPI plus S3 plus a queue plus a vector database. One TableModel file can cover the same slots.

Updated 2026-10-28 · Matching docs cookbook

An upload pipeline has a different failure mode than a batch corpus job. Two requests can race. POST /images does not run a near-duplicate check unless you call that check yourself. The how-to How do I deduplicate a large image dataset states that limit in the code comments.

This architecture covers images, and the same HTTP pattern for video and audio. Image similarity search over HTTP is How do I store images with embeddings and serve similarity search. Audio is How do I transcribe audio uploads and make them searchable. Video insert is on the video search architecture.

User-uploaded media: HTTP, a typed row, then models and an index

Role of each component

Upload HTTP: Multipart insert. FastAPIRouter add_insert_route with uploadfile_inputs. background=True returns a job id for long transcriptions.
Typed media column: Image, Video, or Audio. A path string is not enough for later models. Locally the bytes live under PIXELTABLE_HOME. Cloud hosts the same app file.
Exact-hash check: A UDF that hashes a normalized RGBA PNG. Look up the digest before CLIP so identical pixels never need a vector comparison. There is no built-in image hash function.
Near-duplicate check: CLIP similarity above a threshold you tune. Call it before insert. A computed column on Images that queried Images would be self-referencing.
Enrichment: CLIP, Whisper, YOLOX, or an LLM caption as computed columns. Insert runs the affected expressions. Unchanged rows stay cached.
Search HTTP: A @pxt.query exposed with add_query_route. Image-to-image search uses uploadfile_inputs on the query route; GET cannot take uploads.

Honest alternatives per slot

Keep the named engine when it is already the contract. Pixeltable fills the slot when you want that job on the same table as the files.

Named alternatives for User-uploaded media pipeline reference architecture
Slot	Role	Pixeltable	Named alternatives
Object store	Bytes for uploads	Media columns; PIXELTABLE_HOME locally; Cloud for the hosted file	Amazon S3. The bucket is already the CDN origin or the archive Cloudflare R2. S3-compatible storage with no egress fee on that vendor MinIO. An S3 API inside your network
API / queue	Accept the upload and run work off the request	FastAPIRouter insert; background=True for long jobs	FastAPI + SQS + Lambda. Those services are already the contract Celery + Redis. You already run workers next to a custom API
Vector index	Search and near-duplicate ranking	EmbeddingIndex on the media or transcript column	Qdrant. You already upsert vectors from a worker pgvector. Postgres already holds the rest of the app Pinecone. A hosted ANN namespace is the serving contract
Moderation / virus scan	Reject unsafe or malicious files	Not built in. Run a scanner in front of insert, or a UDF you own	ClamAV or a cloud malware API. You must scan bytes before they are stored A vision moderation API. You must score NSFW or policy labels; Pixeltable can call it as a column

Where Pixeltable fits

Pixeltable is the upload app in this architecture: typed columns, FastAPIRouter insert and search, a hash UDF, and CLIP on insert. Keep S3 plus SQS plus Qdrant when those services are already the contract. POST /images does not call keep_if_new; two uploads at once can both pass. Pixeltable Cloud hosts the same app file. The how-tos for image search and image dedup are the step-by-step versions of this sketch.

import hashlib
from io import BytesIO

import pixeltable as pxt
from pixeltable.functions.huggingface import clip
from pixeltable.serving import FastAPIRouter

TableModel = pxt.model_base()
clip_embed = clip.using(model_id='openai/clip-vit-base-patch32')

def rgba_png_sha256(img) -> str:
    buf = BytesIO()
    img.convert('RGBA').save(buf, format='PNG')
    return hashlib.sha256(buf.getvalue()).hexdigest()

@pxt.udf
def image_sha256(img: pxt.Image) -> str:
    return rgba_png_sha256(img)

class Images(TableModel, name='images'):
    caption: pxt.String | None
    image: pxt.Image
    file_sha256 = image_sha256(image)
    __indexes__ = [
        pxt.EmbeddingIndex(image, embedding=clip_embed),
    ]

@pxt.query
def search_by_text(query: str, limit: int = 5):
    sim = Images.image.similarity(string=query)
    return (
        Images.order_by(sim, asc=False)
        .limit(limit)
        .select(Images.image, Images.caption, sim)
    )

@pxt.query
def search_by_image(img: pxt.Image, limit: int = 5):
    sim = Images.image.similarity(image=img)
    return (
        Images.order_by(sim, asc=False)
        .limit(limit)
        .select(Images.image, Images.caption, sim)
    )

def keep_if_new(candidate, caption: str | None = None, threshold: float = 0.95) -> None:
    images = pxt.get_table('uploads.images')
    digest = rgba_png_sha256(candidate)
    exact = images.where(images.file_sha256 == digest).select(images.file_sha256).collect()
    if len(exact) > 0:
        return
    sim = images.image.similarity(image=candidate)
    hits = (
        images.where(sim > threshold)
        .order_by(sim, asc=False)
        .limit(10)
        .select(images.file_sha256, sim)
        .collect()
    )
    if len(hits) == 0:
        images.insert([{'image': candidate, 'caption': caption}])

api = FastAPIRouter(name='uploads')
api.add_insert_route(
    Images,
    path='/images',
    inputs=[Images.caption],
    uploadfile_inputs=[Images.image],
)
api.add_query_route(path='/search', query=search_by_text)
api.add_query_route(
    path='/search-image',
    query=search_by_image,
    uploadfile_inputs=['img'],
)

# POST /images does not call keep_if_new. Call it yourself after schema update.
# pxt init
# pxt schema update app.py uploads -f
# pxt service update app.py uploads -f

To run a sketch, install pixeltable[serve] and the packages its model functions import (such as sentence-transformers, openai, or openai-whisper), export the provider key a model call reads (such as OPENAI_API_KEY), save the file as app.py, and run the commands in its last lines. The tables start empty: insert a row, or POST to an insert route, then select the computed columns. Get started covers the CLI.

Questions

Does the insert route reject duplicates?

No. add_insert_route writes the row. keep_if_new is a function you call before insert. Two uploads at once can both pass and both land. The image-dedup how-to states that limit.

Where should malware scanning run?

In front of insert, or as a UDF you own. Pixeltable does not scan uploads. A ClamAV sidecar or a cloud malware API is the usual slot.

Can the same pattern serve video or audio?

Yes. Swap the column type and the models. Video search and audio transcription architectures on this hub use uploadfile_inputs the same way. background=True fits a long Whisper job.