Hand-Written FastAPI Endpoints

If your Pixeltable app has hand-written FastAPI endpoints that call pxt.get_table(), run queries, and manually serialize results, you can replace most of them with FastAPIRouter routes. The result: fewer lines of code, automatic request/response schemas, built-in media serving, and background job support.

Related guide: Serving Tables and Queries over HTTP

Concept Mapping

Hand-Written Endpoints	`FastAPIRouter` Routes
Parse request, call `table.insert()`, serialize response	`add_insert_route(table, path, inputs, outputs)`
Parse params, run query, call `to_pydantic()` or `to_pandas()`	`add_query_route(path, query)`
Parse body, call `table.delete_where()`	`add_delete_route(table, path)`
One Pydantic model per endpoint	Auto-generated from column schemas
Manual file responses for media	Built-in `/media/...` handler
Custom task queue for background work	`background=True` on any route

Side by Side: Query Endpoint

Hand-Written
FastAPIRouter

from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter(prefix="/api/data", tags=["data"])

class DocumentItem(BaseModel):
    title: str
    document: str

class DocumentsResponse(BaseModel):
    items: list[DocumentItem]

@router.get("/documents", response_model=DocumentsResponse)
def list_documents():
    t = pxt.get_table('myapp.documents')
    results = t.select(t.title, t.document).collect()
    return DocumentsResponse(
        items=[DocumentItem(**row) for row in results.to_dicts()]
    )

import pixeltable as pxt
from pixeltable.serving import FastAPIRouter

router = FastAPIRouter(prefix="/api/data", tags=["data"])
docs = pxt.get_table('myapp/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")

What Changes

	Hand-Written	`FastAPIRouter`
New endpoint	Write handler, Pydantic model, serialization	One `add_*_route()` call
File uploads	Parse `UploadFile`, save, pass path to insert	`uploadfile_inputs=["image"]`
Media responses	Manual `FileResponse` or base64 encoding	Built-in `/media/...` serving
Background processing	Custom task queue (Celery, RQ, etc.)	`background=True` on any route
OpenAPI docs	Manual schema definitions	Auto-generated from columns
Delete	Parse body, call `delete_where()`	`add_delete_route(table, path)`

Step-by-Step Migration

1. Define queries in router files

Every read pattern becomes a @pxt.query function defined in the router file where it’s used. Each one maps to a single add_query_route call.

@pxt.query eagerly evaluates the function body at decoration time. Tables must exist before the router module is imported. Run python schema.py before starting the app.

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

router = FastAPIRouter(prefix="/api/data", tags=["data"])
docs = pxt.get_table('myapp/documents')

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

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

router.add_query_route(path="/documents", query=list_documents, method="get")
router.add_query_route(path="/search", query=search_documents)
router.add_insert_route(docs, path="/upload", uploadfile_inputs=["document"], outputs=["uuid"])

2. Replace `APIRouter` with `FastAPIRouter`

FastAPIRouter is a subclass of APIRouter, so hand-written @router.post() endpoints coexist on the same instance. Mount routers in main.py:

# main.py
from fastapi import FastAPI
from routers import data, search

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

4. Convert uploads

Hand-Written
FastAPIRouter

@router.post("/upload")
def upload(file: UploadFile):
    t = pxt.get_table('myapp.images')
    t.insert([{'image': file.filename, 'timestamp': datetime.now()}])
    return {"status": "ok"}

images_table = pxt.get_table('myapp/images')
router.add_insert_route(
    images_table, path="/upload",
    uploadfile_inputs=["image"], inputs=["timestamp"],
    outputs=["uuid", "thumbnail"],
)

The multipart form field name must match the column name.

5. Convert deletes

router.add_delete_route(docs_table, path="/delete")
# Client sends: POST {"uuid": "..."} (matches primary key)
# Response: {"num_rows": 1}

6. Delete old Pydantic models

FastAPIRouter auto-generates request/response schemas from column types. Delete hand-written Pydantic models for any endpoint that is now declarative. Keep only models for remaining hand-written endpoints.

Eliminating Insert-Then-Query with `return_rows=True`

A common hand-written pattern is inserting a row, then immediately querying to read back computed columns. Use return_rows=True instead:

Hand-Written
return_rows=True

@router.post("/query")
def agent_query(request: QueryRequest):
    table.insert([{"prompt": request.prompt}])
    result = table.where(table.prompt == request.prompt).select(
        table.answer, table.tool_output
    ).collect()
    return result[0]

from pydantic import BaseModel

class AgentResult(BaseModel):
    model_config = {"extra": "ignore"}
    answer: str | None = None
    tool_output: Any = None

@router.post("/query")
def agent_query(request: QueryRequest):
    status = table.insert(
        [{"prompt": request.prompt}], return_rows=True
    )
    return AgentResult.model_validate(status.rows[0])

status.rows contains one dict per inserted row with all columns (including computed). Use model_validate() with extra="ignore" for typed access to the subset you need. Also works with update() and batch_update().

When to Keep a Hand-Written Endpoint

Not everything fits the declarative model. Keep @router.post() when:

Multi-table operations: inserting into one table then conditionally writing to another
Conditional logic: different behavior based on intermediate results
Custom response shapes: aggregating across multiple tables into a single response
Side effects: sending emails, webhooks, or other non-Pixeltable actions

Since FastAPIRouter extends APIRouter, hand-written and declarative routes coexist on the same router instance.

Gotchas

`@pxt.query` runs at decoration time

The function body executes when Python hits the @pxt.query decorator, not when you call the function. If the tables don’t exist yet, you get an error. Run python schema.py before starting the app so tables exist when router modules are imported.

UUID parameters need `uuid.UUID` annotations

Pixeltable UUID columns require uuid.UUID objects. Use the type annotation and let Pydantic parse strings automatically:

import uuid as _uuid

@pxt.query
def get_chunks(file_uuid: _uuid.UUID):
    return view.where(view.uuid == file_uuid).select(...)

Do not call UUID() inside the query body. The parameter is an expression proxy, not a string.

Media columns serialize as URL strings

FastAPIRouter serializes pxt.Document, pxt.Image, pxt.Video columns as URL paths (e.g., /api/data/media/path/to/file.pdf). The client receives a string, not binary data.

Query routes default to POST

add_query_route defaults to method="post" (JSON body). For parameter-free list endpoints, set method="get":

router.add_query_route(path="/list", query=list_all, method="get")

Delete routes use POST, not HTTP DELETE

add_delete_route uses POST with a JSON body ({"uuid": "..."}), not the HTTP DELETE method. Update client fetch calls accordingly.

Query responses wrap results in `{ "rows": [...] }`

All add_query_route responses return {"rows": [...]}, not a flat array. Client code must unwrap .rows.

Name your embedding indexes explicitly

add_embedding_index(if_exists="ignore") can create duplicates without an explicit name. Always pass idx_name:

view.add_embedding_index(
    'text', idx_name='chunks_text_embed',
    string_embed=e5_embed, if_exists='ignore',
)

Best Practices

Start with @pxt.query for all read patterns. Each one becomes a single add_query_route.
One table per route. Let the client merge results from granular endpoints.
Use FastAPIRouter as your base router. You get media serving and background jobs for free.
Keep complex orchestration hand-written. Multi-table inserts and conditional logic stay as @router.post().
Type your @pxt.query parameters (uuid.UUID, int, float, str). Pydantic coerces the incoming JSON; Pixeltable handles the rest.
Prototype with TOML first. pxt serve with [[tool.pixeltable.service.routes]] in pyproject.toml is the fastest way to test a new route before wiring it into your app.

Next Steps

HTTP Serving Guide

TOML config, CLI, Python API, background jobs

Deployment Overview

Full backend, batch processing, and declarative serving

Production Operations

Concurrency, error handling, sync endpoints

AI Applications

End-to-end multimodal app patterns

Welcome to Pixeltable

Tools

Core Concepts

How-To

Concept Mapping

Side by Side: Query Endpoint

What Changes

Step-by-Step Migration

1. Define queries in router files

2. Replace `APIRouter` with `FastAPIRouter`

4. Convert uploads

5. Convert deletes

6. Delete old Pydantic models

Eliminating Insert-Then-Query with `return_rows=True`

When to Keep a Hand-Written Endpoint

Gotchas

`@pxt.query` runs at decoration time

UUID parameters need `uuid.UUID` annotations

Media columns serialize as URL strings

Query routes default to POST

Delete routes use POST, not HTTP DELETE

Query responses wrap results in `{ "rows": [...] }`

Name your embedding indexes explicitly

Best Practices

Next Steps

HTTP Serving Guide

Deployment Overview

Production Operations

AI Applications

​Concept Mapping

​Side by Side: Query Endpoint

​What Changes

​Step-by-Step Migration

​1. Define queries in router files

​2. Replace APIRouter with FastAPIRouter

​4. Convert uploads

​5. Convert deletes

​6. Delete old Pydantic models

​Eliminating Insert-Then-Query with return_rows=True

​When to Keep a Hand-Written Endpoint

​Gotchas

​@pxt.query runs at decoration time

​UUID parameters need uuid.UUID annotations

​Media columns serialize as URL strings

​Query routes default to POST

​Delete routes use POST, not HTTP DELETE

​Query responses wrap results in { "rows": [...] }

​Name your embedding indexes explicitly

​Best Practices

​Next Steps

HTTP Serving Guide

Deployment Overview

Production Operations

AI Applications

Concept Mapping

Side by Side: Query Endpoint

What Changes

Step-by-Step Migration

1. Define queries in router files

2. Replace `APIRouter` with `FastAPIRouter`

4. Convert uploads

5. Convert deletes

6. Delete old Pydantic models

Eliminating Insert-Then-Query with `return_rows=True`

When to Keep a Hand-Written Endpoint

Gotchas

`@pxt.query` runs at decoration time

UUID parameters need `uuid.UUID` annotations

Media columns serialize as URL strings

Query routes default to POST

Delete routes use POST, not HTTP DELETE

Query responses wrap results in `{ "rows": [...] }`

Name your embedding indexes explicitly

Best Practices

Next Steps