HTTP serving

Install FastAPI and uvicorn first:

pip install 'pixeltable[serve]'

How the process starts is on Self-hosting. This page covers the route API once you have chosen which process serves it. A local service exposes FastAPI’s interactive docs at /docs and its OpenAPI schema at /openapi.json. pxt service update assigns a local port; look up the endpoint by catalog path and service name rather than assuming port 8000. For the README’s my_app/ingest example:

SERVICE_URL=$(pxt service list my_app --json | jq -r '.[] | select(.catalog_path == "my_app" and .name == "ingest") | .endpoint')
curl -fsS "${SERVICE_URL%/}/openapi.json" -o openapi.json
npx openapi-typescript openapi.json -o src/pixeltable.d.ts

Replace my_app and ingest when using another target or service. Each column named in a route’s outputs becomes a typed field, computed columns included. Regenerate types after changing the Python routes. For a hosted service, copy its full endpoint URL from the Cloud dashboard or find it with pxt service list pxt://org:db --json using a credential that can list services. A key scoped only to invoke a service cannot list services, but can download that service’s schema when granted access. The gateway requires a credential to download the schema:

curl -fsS -H "X-api-key: $PIXELTABLE_API_KEY" "${SERVICE_URL%/}/openapi.json" -o openapi.json
npx openapi-typescript openapi.json -o src/pixeltable.d.ts

The hosted schema declares the gateway’s X-api-key header on every route except /health; local service schemas declare no key. In the Cloud dashboard, expand API docs on the service detail page to inspect and try routes as your signed-in user, without an API key. Trying an insert, update, or delete route can change data. Routes with return_fileresponse=True advertise a binary response for any media type. Check the response’s Content-Type when handling the returned image, video, audio, or document.

Call from a TypeScript app server

Keep a Cloud API key in your application server’s environment, never in browser code. Generated types do not check raw fetch calls, so call routes through openapi-fetch with the generated paths. The same module calls a local service (no key) or a hosted one. Set PIXELTABLE_SERVICE_URL to the full service endpoint, as above, run npm install openapi-fetch server-only, and put this in src/lib/pixeltable-client.ts. With the README’s /titles compute route:

import 'server-only';
import createClient from 'openapi-fetch';
import type { paths } from '../pixeltable';

const apiKey = process.env.PIXELTABLE_API_KEY;

export const pixeltable = createClient<paths>({
  baseUrl: process.env.PIXELTABLE_SERVICE_URL,
  headers: apiKey ? { 'X-api-key': apiKey } : {},
  cache: 'no-store',
});

export class PixeltableError extends Error {
  status: number;
  body: unknown;
  constructor(status: number, body: unknown) {
    super(`Pixeltable service returned HTTP ${status}`);
    this.status = status;
    this.body = body;
  }
}

export async function computeTitle(title: string) {
  const { data, error, response } = await pixeltable.POST('/titles', { body: { title } });
  if (error !== undefined || !response.ok) throw new PixeltableError(response.status, error);
  return data; // typed from the service schema
}

Call it from a Server Component, or from your own Route Handler or Server Action; protect any that expose writes to browser users with your application’s authorization checks. Regenerate src/pixeltable.d.ts when routes change, and run your TypeScript check so request and response mismatches fail before deployment. The Cloud dashboard’s Use this endpoint TypeScript and JavaScript snippets are server-side examples. A return_fileresponse=True route needs parseAs: 'blob', or parseAs: 'stream' to pipe response.body through your Route Handler with the upstream Content-Type; without it the client parses the file as JSON and throws. Generated types describe upload fields as strings, so send an upload with fetch, a FormData body, and the same X-api-key header, and let fetch set the multipart boundary. PixeltableError.body is the parsed error response. Its detail is an object with error_code, message, and retryable (sometimes retry_after) for a Pixeltable runtime error and for an error the Cloud gateway answers itself: 401 for a missing or invalid key, 403 for a key without access to the service, 404 for an unknown service, 429 when the key is rate limited, and 503 when the service is not running. It is a string for a missing row and an array for a request-validation error, so branch on its shape, not the status. Retry a write only when detail.retryable is true, after retry_after seconds when present.

Mount on your own FastAPI app

import fastapi
import uvicorn
import pixeltable as pxt
from pixeltable.serving import FastAPIRouter

t = pxt.get_table('my_app.docs')

app = fastapi.FastAPI()
router = FastAPIRouter()
router.add_insert_route(
    t,
    path='/insert',
    inputs=['prompt'],
    outputs=['prompt', 'result'],
)
router.add_update_route(
    t,
    path='/update',
    inputs=['prompt'],
    outputs=['id', 'prompt', 'result'],
)
app.include_router(router)

uvicorn.run(app, host='0.0.0.0', port=8000)

@pxt.query evaluates the function body at decoration time. Define @pxt.query functions only after pxt schema update, because the decorator runs the function body immediately. Use a plain FastAPI @app.post() when one FastAPIRouter helper cannot express the request.

Compute without inserting

Table.compute() runs computed columns and returns the values without writing a row.

rows = t.compute([{'prompt': 'hello'}])
print(rows[0]['result'])

Same over HTTP:

router.add_compute_route(
    t,
    path='/preview',
    inputs=['prompt'],
    outputs=['prompt', 'result'],
)

Decorator-style routes

add_insert_route() builds the response model from the column schema. To return a custom JSON body, use @router.insert_route instead of add_insert_route(). The function receives outputs as keyword arguments and returns a pydantic.BaseModel.

import pydantic
from pixeltable.serving import FastAPIRouter

router = FastAPIRouter()


class GenerateResponse(pydantic.BaseModel):
    caption: str
    score: float


@router.insert_route(
    t,
    path='/generate',
    inputs=['prompt'],
    outputs=['caption', 'score'],
)
def format_insert(
    *, caption: str, score: float
) -> GenerateResponse:
    return GenerateResponse(
        caption=caption.strip(), score=round(score, 3)
    )


@router.update_route(
    t,
    path='/update',
    inputs=['prompt'],
    outputs=['id', 'caption', 'score'],
)
def format_update(
    *, id: int, caption: str, score: float
) -> GenerateResponse:
    return GenerateResponse(
        caption=caption.strip(), score=round(score, 3)
    )

At registration:

Every parameter is keyword-only and annotated.
Parameter names match outputs exactly.
Annotations match column types (nullable column: T | None). Media columns arrive as URL strings: annotate str.
Return type is a pydantic.BaseModel subclass.

background=True works the same as the non-decorator forms. Decorator routes are Python-only.

Export to an external database

SqlExport writes each successful insert or update to another SQL table. Pixeltable commits first; then the external write. If the external write fails, the request is HTTP 500. No rollback.

from pixeltable.serving import FastAPIRouter, SqlExport

router = FastAPIRouter()
router.add_insert_route(
    t,
    path='/generate',
    inputs=['prompt'],
    outputs=['prompt', 'result'],
    export_sql=SqlExport(
        db_connect='postgresql+psycopg://user:pw@host/analytics',
        table='generations',
    ),
)

The row is the response body (outputs). Media columns are URL strings. The target table must already exist. SqlExport.method:

'insert' (default): append. Replaying the request duplicates the row.
'update': match on the target primary key. Not an upsert. No match: HTTP 500. Response columns must include every target PK plus at least one non-PK.
'merge': not supported.

A Pixeltable insert with method='update' is allowed: append-only here, current-state there. export_sql= cannot combine with return_fileresponse=True. It works with background=True (the SQL write runs in the worker). SqlExport

A connection string with an embedded password is plaintext in the application file. Pull credentials from the environment, a .pgpass-style file, or pxt secret set.

Full example

# app.py
import pixeltable as pxt
from pixeltable.serving import FastAPIRouter

TableModel = pxt.model_base()


class Images(TableModel, name='images'):
    image: pxt.Image
    width: pxt.Int
    height: pxt.Int
    tag: pxt.String | None
    thumbnail = image.resize(size=(128, 128))


@pxt.query
def search_images(text: str) -> pxt.Query:
    return Images.where(Images.tag == text).select(Images.thumbnail)


images = FastAPIRouter(name='image-processing')

images.add_insert_route(
    Images,
    path='/process',
    inputs=[Images.width, Images.height],
    uploadfile_inputs=['image'],
    outputs=[Images.thumbnail],
)

images.add_insert_route(Images, path='/ingest', background=True)

images.add_update_route(
    Images,
    path='/images/update',
    inputs=[Images.tag],
    outputs=[Images.thumbnail],
)

images.add_delete_route(Images, path='/images/delete')
images.add_delete_route(
    Images, path='/images/delete-by-tag', match_columns=['tag']
)

images.add_query_route(path='/search', query=search_images)

images.add_query_route(
    path='/thumbnail',
    query=search_images,
    one_row=True,
    method='get',
    return_fileresponse=True,
)

Return computed columns

insert(), update(), and batch_update() can return computed columns without a follow-up query:

status = table.insert([row], return_rows=True)
data = status.rows[0]

status.rows is a list of dicts. For typed access, model_validate() with extra="ignore":

from pydantic import BaseModel


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


status = agent_table.insert(
    [{'prompt': user_input}], return_rows=True
)
result = AgentResult.model_validate(status.rows[0])

After .collect(), use to_pydantic(). After return_rows=True, use model_validate().

Background jobs

background=True returns a job handle immediately:

{
  "id": "abc123",
  "job_url": "http://127.0.0.1:<port>/_pxt/jobs/abc123"
}

Poll job_url (pxt service list prints the base URL):

curl http://127.0.0.1:<port>/_pxt/jobs/abc123
# {"status": "pending"}
# {"status": "done", "result": {...}}
# {"status": "error", "error": "...", "error_detail": {...}}

Poll through your application server with the same Cloud key when hosted. Cloud returns an HTTPS job_url. The polling path is in the generated schema, so polling is a typed call on the same TypeScript client: pixeltable.GET('/_pxt/jobs/{job_id}', { params: { path: { job_id: id } } }). A failed job’s error_detail has the same shape as an HTTP error’s detail, so the same retry rule applies. background cannot combine with return_fileresponse. Flags: CLI.

Getting started

Tools

How to

Resources

Call from a TypeScript app server

Mount on your own FastAPI app

Compute without inserting

Decorator-style routes

Export to an external database

Full example

Return computed columns

Background jobs

​Call from a TypeScript app server

​Mount on your own FastAPI app

​Compute without inserting

​Decorator-style routes

​Export to an external database

​Full example

​Return computed columns

​Background jobs

Call from a TypeScript app server

Mount on your own FastAPI app

Compute without inserting

Decorator-style routes

Export to an external database

Full example

Return computed columns

Background jobs