How to Integrate with FastAPI

The classic arrowmodel use case: your database returns Arrow data and your API needs to serve it as JSON. This guide shows how to wire arrowmodel into a FastAPI endpoint so that Arrow query results go straight to typed Pydantic response models with no intermediate dict layer.

Prerequisites: Working FastAPI project, a database that returns Arrow data (ADBC, Flight SQL, or any Arrow-PyCapsule source).

Define your response model

Start with a Pydantic model that matches the columns your query returns. Using ArrowModel as the base class gives you the convert classmethod for free:

from arrowmodel import ArrowModel


class Product(ArrowModel):
    id: int
    name: str
    price: float
    in_stock: bool

Build the endpoint

Query the database, convert the Arrow result to models, and return them. FastAPI handles the JSON serialisation of the Pydantic model list.

import adbc_driver_sqlite.dbapi as adbc_sqlite
from fastapi import FastAPI

app = FastAPI()


@app.get("/products", response_model=list[Product])
def list_products():
    with adbc_sqlite.connect("products.db") as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id, name, price, in_stock FROM products")
            table = cur.fetch_arrow_table()

    return Product.convert(table)

That is the core pattern. fetch_arrow_table() returns a pyarrow Table, convert() turns it into a list[Product], and FastAPI serialises that list to JSON.

Reuse converters across requests

ArrowModel caches the converter at class definition time, so Product.convert(table) already reuses the compiled field mapping. No extra setup needed.

If you are using ArrowModelConverter directly (because the model cannot subclass ArrowModel), create the converter once at module level:

from pydantic import BaseModel
from arrowmodel import ArrowModelConverter


class Product(BaseModel):
    id: int
    name: str
    price: float
    in_stock: bool


product_converter = ArrowModelConverter(Product)


@app.get("/products", response_model=list[Product])
def list_products():
    with adbc_sqlite.connect("products.db") as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id, name, price, in_stock FROM products")
            table = cur.fetch_arrow_table()

    return product_converter.convert(table)

Use validated mode for untrusted data

If the Arrow data comes from an external source – a third-party Flight SQL server, a user-uploaded Parquet file, or a federated query – enable validation to run Pydantic’s full pipeline:

@app.post("/products/import", response_model=list[Product])
def import_products(table: pa.Table):
    # Untrusted data: run full Pydantic validation
    return Product.convert(table, validate=True)

See How to Use Validated Mode for more on what the validated path catches and the performance trade-off.

Stream large results

For endpoints that return large datasets, use iter() with FastAPI’s StreamingResponse to avoid materialising the entire list in memory:

from fastapi.responses import StreamingResponse


@app.get("/products/export")
def export_products():
    with adbc_sqlite.connect("products.db") as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id, name, price, in_stock FROM products")
            table = cur.fetch_arrow_table()

    def generate():
        for product in Product.iter(table):
            yield product.model_dump_json() + "\n"

    return StreamingResponse(generate(), media_type="application/x-ndjson")