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. After a service is running, FastAPI’s OpenAPI UI is at http://<host>:<port>/docs.

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 a pxt.Secret config var.

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": "..."}

background cannot combine with return_fileresponse. Flags: CLI.

Getting started

Tools

How to

Resources

Mount on your own FastAPI app

Compute without inserting

Decorator-style routes

Export to an external database

Full example

Return computed columns

Background jobs

​Mount on your own FastAPI app

​Compute without inserting

​Decorator-style routes

​Export to an external database

​Full example

​Return computed columns

​Background jobs

Mount on your own FastAPI app

Compute without inserting

Decorator-style routes

Export to an external database

Full example

Return computed columns

Background jobs