All samples

Research sample · Python 3.14

Production FastAPI and pgvector RAG API

Research question

How do I build a production RAG API with Python FastAPI, PostgreSQL, and pgvector in 2026? Audit this implementation for async SQLAlchemy correctness, hybrid search and reciprocal rank fusion, HNSW filtered-search recall, tenant isolation and PostgreSQL row-level security, embedding versioning, timeouts, connection pooling, caching, observability, evaluation, and graceful shutdown. Identify exact defects and modernization opportunities, then provide corrected examples and authoritative sources.

Topics covered: FastAPI RAG, PostgreSQL pgvector, HNSW filtering, tenant isolation

Hat Trick research

Python 3.14 · 9,202 characters

Research complete

Question

How do I build a production RAG API with Python FastAPI, PostgreSQL, and pgvector in 2026? Audit this implementation for async SQLAlchemy correctness, hybrid search and reciprocal rank fusion, HNSW filtered-search recall, tenant isolation and PostgreSQL row-level security, embedding versioning, timeouts, connection pooling, caching, observability, evaluation, and graceful shutdown. Identify exact defects and modernization opportunities, then provide corrected examples and authoritative sources.

Submitted code · production-rag-api.pyPython
from __future__ import annotations

import asyncio
import hashlib
import json
import os
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Annotated, Protocol
from uuid import UUID

from fastapi import Depends, FastAPI, Header, HTTPException, Request
from pydantic import BaseModel, Field
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)


class Embedder(Protocol):
    dimensions: int
    model: str

    async def embed(self, texts: Sequence[str]) -> list[list[float]]: ...


class HttpEmbedder:
    model = "text-embedding-3-small"
    dimensions = 512
View all 306 lines
from __future__ import annotations

import asyncio
import hashlib
import json
import os
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Annotated, Protocol
from uuid import UUID

from fastapi import Depends, FastAPI, Header, HTTPException, Request
from pydantic import BaseModel, Field
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)


class Embedder(Protocol):
    dimensions: int
    model: str

    async def embed(self, texts: Sequence[str]) -> list[list[float]]: ...


class HttpEmbedder:
    model = "text-embedding-3-small"
    dimensions = 512

    async def embed(self, texts: Sequence[str]) -> list[list[float]]:
        # Replace with a bounded provider client that records exact token usage.
        await asyncio.sleep(0)
        return [[0.0] * self.dimensions for _ in texts]


@dataclass(frozen=True)
class Settings:
    database_url: str
    query_timeout_seconds: float = 8.0
    result_limit: int = 12

    @classmethod
    def from_env(cls) -> "Settings":
        url = os.environ.get("DATABASE_URL", "").strip()
        if not url:
            raise RuntimeError("DATABASE_URL is required")
        return cls(database_url=url)


class SearchRequest(BaseModel):
    query: str = Field(min_length=3, max_length=1_000)
    repository_id: UUID
    limit: int = Field(default=8, ge=1, le=20)


class SearchHit(BaseModel):
    chunk_id: UUID
    path: str
    start_line: int
    end_line: int
    content: str
    semantic_score: float
    lexical_score: float
    fused_score: float


class SearchResponse(BaseModel):
    query_hash: str
    embedding_model: str
    embedding_dimensions: int
    hits: list[SearchHit]


@dataclass(frozen=True)
class Principal:
    user_id: UUID
    workspace_id: UUID


async def principal(
    x_user_id: Annotated[str, Header()],
    x_workspace_id: Annotated[str, Header()],
) -> Principal:
    # A production gateway verifies a signed access token before setting claims.
    try:
        return Principal(UUID(x_user_id), UUID(x_workspace_id))
    except ValueError as error:
        raise HTTPException(401, "Invalid authenticated identity") from error


class RagStore:
    def __init__(
        self,
        sessions: async_sessionmaker[AsyncSession],
        embedder: Embedder,
    ) -> None:
        self.sessions = sessions
        self.embedder = embedder

    async def hybrid_search(
        self,
        *,
        principal: Principal,
        request: SearchRequest,
    ) -> SearchResponse:
        embedding = (await self.embedder.embed([request.query]))[0]
        if len(embedding) != self.embedder.dimensions:
            raise RuntimeError("Embedding dimension mismatch")

        async with self.sessions.begin() as session:
            # RLS policies read these transaction-local settings. The repository
            # predicate remains explicit as defense in depth and for planning.
            await session.execute(
                text("select set_config('app.workspace_id', :value, true)"),
                {"value": str(principal.workspace_id)},
            )
            await session.execute(
                text("select set_config('app.user_id', :value, true)"),
                {"value": str(principal.user_id)},
            )
            await session.execute(text("set local hnsw.iterative_scan = strict_order"))
            rows = await session.execute(
                HYBRID_SEARCH_SQL,
                {
                    "workspace_id": principal.workspace_id,
                    "repository_id": request.repository_id,
                    "query": request.query,
                    "embedding": json.dumps(embedding),
                    "candidate_limit": max(50, request.limit * 8),
                    "result_limit": request.limit,
                },
            )

        hits = [SearchHit.model_validate(dict(row)) for row in rows.mappings()]
        query_hash = hashlib.sha256(
            f"{principal.workspace_id}\0{request.repository_id}\0{request.query}".encode()
        ).hexdigest()
        return SearchResponse(
            query_hash=query_hash,
            embedding_model=self.embedder.model,
            embedding_dimensions=self.embedder.dimensions,
            hits=hits,
        )


HYBRID_SEARCH_SQL = text(
    """
    with semantic as materialized (
      select
        c.id,
        row_number() over (order by c.embedding <=> cast(:embedding as vector)) as rank,
        1 - (c.embedding <=> cast(:embedding as vector)) as score
      from repository_chunks c
      where c.workspace_id = :workspace_id
        and c.repository_id = :repository_id
        and c.embedding is not null
      order by c.embedding <=> cast(:embedding as vector)
      limit :candidate_limit
    ),
    lexical as materialized (
      select
        c.id,
        row_number() over (
          order by ts_rank_cd(c.search_document, websearch_to_tsquery('english', :query)) desc
        ) as rank,
        ts_rank_cd(c.search_document, websearch_to_tsquery('english', :query)) as score
      from repository_chunks c
      where c.workspace_id = :workspace_id
        and c.repository_id = :repository_id
        and c.search_document @@ websearch_to_tsquery('english', :query)
      order by score desc
      limit :candidate_limit
    ),
    fused as (
      select
        coalesce(s.id, l.id) as id,
        coalesce(s.score, 0) as semantic_score,
        coalesce(l.score, 0) as lexical_score,
        coalesce(1.0 / (60 + s.rank), 0) +
          coalesce(1.0 / (60 + l.rank), 0) as fused_score
      from semantic s
      full outer join lexical l on l.id = s.id
    )
    select
      c.id as chunk_id,
      c.path,
      c.start_line,
      c.end_line,
      c.content,
      f.semantic_score,
      f.lexical_score,
      f.fused_score
    from fused f
    join repository_chunks c on c.id = f.id
    where c.workspace_id = :workspace_id
      and c.repository_id = :repository_id
    order by f.fused_score desc, c.path, c.start_line
    limit :result_limit
    """
)


def create_engine(settings: Settings) -> AsyncEngine:
    return create_async_engine(
        settings.database_url,
        pool_pre_ping=True,
        pool_size=10,
        max_overflow=10,
        pool_recycle=1_800,
    )


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    settings = Settings.from_env()
    engine = create_engine(settings)
    app.state.settings = settings
    app.state.engine = engine
    app.state.sessions = async_sessionmaker(engine, expire_on_commit=False)
    app.state.embedder = HttpEmbedder()
    try:
        async with engine.connect() as connection:
            await connection.execute(text("select 1"))
        yield
    finally:
        await engine.dispose()


app = FastAPI(
    title="Tenant-isolated hybrid RAG API",
    version="1.0.0",
    lifespan=lifespan,
)


def store(request: Request) -> RagStore:
    return RagStore(request.app.state.sessions, request.app.state.embedder)


@app.get("/healthz")
async def health(request: Request) -> dict[str, object]:
    async with request.app.state.engine.connect() as connection:
        await connection.execute(text("select 1"))
    return {"ok": True, "embedding_model": request.app.state.embedder.model}


@app.post("/v1/search", response_model=SearchResponse)
async def search(
    body: SearchRequest,
    identity: Annotated[Principal, Depends(principal)],
    rag: Annotated[RagStore, Depends(store)],
    request: Request,
) -> SearchResponse:
    timeout = request.app.state.settings.query_timeout_seconds
    try:
        async with asyncio.timeout(timeout):
            return await rag.hybrid_search(principal=identity, request=body)
    except TimeoutError as error:
        raise HTTPException(504, "Search deadline exceeded") from error


SCHEMA_SQL = """
create extension if not exists vector;

create table repository_chunks (
  id uuid primary key,
  workspace_id uuid not null,
  repository_id uuid not null,
  commit_sha text not null,
  path text not null,
  start_line integer not null,
  end_line integer not null,
  content text not null,
  content_hash text not null,
  embedding_model text not null,
  embedding_dimensions integer not null,
  embedding vector(512),
  search_document tsvector generated always as (
    setweight(to_tsvector('english', path), 'A') ||
    setweight(to_tsvector('english', content), 'B')
  ) stored,
  unique (workspace_id, repository_id, commit_sha, path, content_hash)
);

alter table repository_chunks enable row level security;
alter table repository_chunks force row level security;

create policy workspace_chunks on repository_chunks
using (
  workspace_id = nullif(current_setting('app.workspace_id', true), '')::uuid
);

create index repository_chunks_hnsw
on repository_chunks using hnsw (embedding vector_cosine_ops);

create index repository_chunks_search
on repository_chunks using gin (search_document);

create index repository_chunks_tenant_repo
on repository_chunks (workspace_id, repository_id, path);
"""

Result

Python3 priority findings6 displayed sources

Tenant identity is trusted before it is authenticated

The API has useful transaction-local RLS context, parameterized hybrid search, deadlines, and explicit engine disposal. Direct exposure is still unsafe because identity comes from unsigned headers, while retrieval ignores the embedding model and dimensions already stored in the schema.

Priority findings

03
  1. 01

    Derive tenant identity from a verified token

    Fix now

    The API accepts caller-controlled UUID headers as its principal. Verify signature, issuer, audience, expiry, and workspace membership before establishing RLS context.

    production-rag-api.py:91

    return Principal(UUID(x_user_id), UUID(x_workspace_id))
    FASTAPI-JWT · POSTGRES-RLS
  2. 02

    Filter semantic candidates by embedding generation

    Fix now

    The schema stores model and dimensions, but the query searches every non-null embedding. Constrain candidates to the active model and dimensions before changing models or vector widths.

    production-rag-api.py:161

    and c.embedding is not null
    OPENAI-EMBED · PGVECTOR
  3. 03

    Move startup DDL into versioned migrations

    Fix next

    Schema ownership inside the application lifecycle complicates deployment, rollback, and independent testing. Keep runtime startup focused on resource readiness.

    production-rag-api.py:267

    SCHEMA_SQL = """
    ALEMBIC · FASTAPI-LIFESPAN

Establish a verified principal dependency

Keep authentication at a reusable FastAPI boundary and pass only verified identity into retrieval and RLS context.

Modernized techniquePython
async def principal(authorization: str = Header()) -> Principal:
    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer":
        raise HTTPException(401, "Bearer token required")

    claims = await verify_signed_token(token)
    return Principal(
        user_id=UUID(claims["sub"]),
        workspace_id=UUID(claims["workspace_id"]),
    )

Authoritative sources

FASTAPI-JWTOAuth2 with JWT tokensPOSTGRES-RLSRow security policiesPGVECTORFiltering and iterative vector scansOPENAI-EMBEDCreate embeddingsALEMBICAlembic tutorialFASTAPI-LIFESPANLifespan events

What this production RAG API audit covers

This Python example combines FastAPI dependencies, async SQLAlchemy sessions, PostgreSQL row-level security context, semantic and lexical retrieval, reciprocal rank fusion, pgvector HNSW search, request deadlines, and application lifespan cleanup.

Hattrick evaluates the security and retrieval contracts together. It checks whether tenant identity is trustworthy before RLS consumes it, whether incompatible embedding generations can mix, and which performance claims still require representative recall and latency measurements.

Engineering questions answered

  • Can caller-controlled identity headers permit cross-workspace access?
  • Should retrieval filter by embedding model and dimensions before vector search?
  • How should filtered HNSW recall, deadlines, pools, and shutdown be tested in production?

How Hattrick approaches the question

  1. 01

    Tenant-boundary review

    Follows identity from the HTTP request through transaction-local settings, explicit predicates, RLS, and the runtime database role.

  2. 02

    Retrieval-version analysis

    Checks whether stored model and dimension metadata actually constrain semantic candidates and index selection.

  3. 03

    Operational evidence plan

    Separates visible code guarantees from recall, latency, pooling, and cancellation behavior that require integration measurements.

Practical takeaway

RLS is only as trustworthy as its identity source

Transaction-local tenant settings and explicit predicates are useful defense in depth, but they cannot authenticate caller-supplied headers. Verify the principal first, then keep embedding generations and operational measurements explicit.