# Production RAG with FastAPI, PostgreSQL, and pgvector - Hat Trick Report

## Prompt

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.

## Verdict

Authentication is only trusted-header scaffolding, so direct exposure would permit tenant impersonation; require verified signed claims. Embedding version metadata is stored but not filtered during search, allowing incompatible model generations to mix. The request timeout, transaction-local RLS context, parameterized SQL, RRF query, iterative HNSW scan, and engine disposal are visible, but the dossier cannot verify their SQLAlchemy/pgvector correctness or production tuning.

## Detected Context

- Language: Python
- Framework: FastAPI
- Runtime: Python
- Generated: 2026-08-27T10:48:47.957Z

## Detected Techniques

### Environment-based configuration
- Category: Configuration
- Evidence: `Settings.from_env()` requires `DATABASE_URL` from the process environment.

### Explicit resource management
- Category: Lifecycle
- Evidence: The FastAPI lifespan creates the engine, checks connectivity, yields, and calls `await engine.dispose()` in `finally`.

### Asynchronous control flow
- Category: Concurrency
- Evidence: Route handlers, embedding, database operations, timeout handling, and lifespan management use `async`/`await`.

### Structured error handling
- Category: Reliability
- Evidence: Identity parsing and search timeout failures are translated into HTTP 401 and 504 responses.

### Function decomposition
- Category: Design
- Evidence: Configuration, authentication, storage, engine creation, lifecycle, health, and search are separated into named units.

### Server-side route handlers
- Category: API
- Evidence: FastAPI defines GET `/healthz` and POST `/v1/search` handlers with dependency injection and response validation.

## Techniques and Modernity

### Environment-based configuration
- Status: acceptable

The database URL is externally configured, but environment variables alone do not establish secret rotation or least privilege.

**Recommendation:** Supply the URL through a managed secret mechanism and use a restricted database role subject to RLS. Sources: S3.

### Explicit resource management
- Status: acceptable

The lifespan context guarantees engine disposal in a `finally` block, including cancellation paths. The supplied ECMAScript resource-management source is not applicable to Python syntax.

**Recommendation:** Retain FastAPI lifespan ownership and ensure every added client or task is also closed or cancelled there. Sources: S2, S8.

### Asynchronous control flow
- Status: acceptable

The API awaits embedding and database I/O and applies an overall asyncio deadline. The dossier did not complete authoritative SQLAlchemy or pgvector review.

**Recommendation:** Validate driver cancellation and pool behavior with integration tests before production deployment. Sources: S8.

### Structured error handling
- Status: acceptable

Malformed UUID headers and elapsed deadlines receive explicit HTTP errors, while unexpected failures propagate to framework handling.

**Recommendation:** Add centralized, sanitized error logging and stable error bodies without exposing credentials or SQL. Sources: S2, S5.

### Function decomposition
- Status: acceptable

Lifecycle, dependencies, persistence, and route concerns are separated, although SQL and schema remain embedded in one module.

**Recommendation:** Move migrations and query code into independently tested modules as the service grows. Sources: [FastAPI: Bigger Applications](https://fastapi.tiangolo.com/tutorial/bigger-applications/), [Alembic tutorial](https://alembic.sqlalchemy.org/en/latest/tutorial.html).

### Server-side route handlers
- Status: acceptable

FastAPI dependencies and response models provide typed request handling and output validation.

**Recommendation:** Keep handlers thin and place authorization and retrieval invariants in reusable dependencies or services. Sources: S2, S11, S12.

## Security

### Unsigned identity headers can permit tenant impersonation.
- Severity: high

The application accepts caller-controlled UUID headers without verifying a signed token. If an untrusted client can reach it directly, that client can choose another workspace identity; deployment topology might mitigate this, so this is conditional rather than proof of exploitation.

**Recommendation:** Verify token signature, issuer, audience, expiry, and workspace membership in the application or a mutually authenticated trusted gateway; do not trust public identity headers. Sources: S2, S3.

### Database credentials need managed handling and least privilege.
- Severity: medium

External configuration avoids hard-coding, but the excerpt does not show storage, rotation, or database-role restrictions.

**Recommendation:** Use a secret manager, rotate credentials, and grant the runtime role only required operations while ensuring it cannot bypass RLS. Sources: S3.

## Performance

No material performance concern was identified in the submitted scope.

## Maintainability

The implementation has clear typed boundaries, but production claims about async SQLAlchemy, pgvector recall, pooling, caching, observability, and evaluation remain unverified by the supplied research.

### Strengths
- Lifecycle cleanup and request deadlines are explicit.
- SQL values are bound parameters rather than interpolated strings.
- Schema records embedding model and dimensions.

### Improvements
- Filter retrieval by embedding model, dimensions, and intended corpus version.
- Add metrics and evaluations for latency, timeout rate, pool pressure, tenant leakage, and retrieval quality.
- Move schema creation into versioned migrations and test RLS using the actual runtime role.

## Alternatives

### Trusting `X-User-Id` and `X-Workspace-Id` directly. -> Derive the principal from a verified signed access token or a strongly authenticated gateway.
Caller-provided identity is not authentication; secret and credential controls should support least privilege. Sources: S2, S3.

### Searching all non-null embeddings regardless of metadata. -> Constrain semantic candidates to the active embedding model and dimensions, and re-embed through a controlled migration.
The table already stores version metadata, but retrieval does not use it. This recommendation is an inference from the submitted schema and query, supported by the embedding API's explicit dimensions contract and pgvector's model-scoped indexing pattern. Sources: [OpenAI embeddings API](https://developers.openai.com/api/reference/resources/embeddings/methods/create), [pgvector filtering and variable-dimension indexes](https://github.com/pgvector/pgvector#filtering).

## Upgrade Plan

### Establish tenant authentication and verify RLS behavior.
- Priority: now
1. Replace public identity-header trust with verified claims.
2. Run cross-workspace integration tests through the actual restricted runtime role.
3. Keep transaction-local workspace and user settings plus explicit tenant predicates.
Sources: S2, S3.

### Enforce embedding-version consistency.
- Priority: next
1. Add model and dimension parameters to semantic filtering.
2. Index or partition according to measured query plans.
3. Re-embed old rows before changing the active model or vector width.
Sources: [OpenAI embeddings API](https://developers.openai.com/api/reference/resources/embeddings/methods/create), [pgvector filtering and iterative scans](https://github.com/pgvector/pgvector#filtering), [Alembic tutorial](https://alembic.sqlalchemy.org/en/latest/tutorial.html).

### Build production evidence for retrieval and operations.
- Priority: later
1. Measure lexical, semantic, and fused recall against a labeled evaluation set.
2. Test filtered HNSW recall and latency with representative tenant sizes.
3. Instrument deadlines, pool waits, database time, embedding time, cache outcomes, and shutdown cancellation.
Sources: S8, [pgvector performance and recall guidance](https://github.com/pgvector/pgvector#performance), [OpenTelemetry Python instrumentation](https://opentelemetry.io/docs/languages/python/instrumentation/).

## Educational Examples

### Verified principal dependency.
**Concept:** Do not derive authorization solely from caller-controlled identity headers.

This runnable FastAPI example uses a placeholder verifier boundary; replace its body with a real signed-token verifier before deployment. Sources: S2, S3.

```python
from uuid import UUID
from fastapi import Depends, FastAPI, Header, HTTPException
from pydantic import BaseModel

app = FastAPI()
class Principal(BaseModel):
    user_id: UUID
    workspace_id: UUID

def verify_token(token: str) -> Principal:
    # Demo only: production code verifies signature and claims.
    if token != "demo":
        raise HTTPException(401, "Invalid token")
    return Principal(user_id=UUID(int=1), workspace_id=UUID(int=2))

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

@app.get("/whoami")
async def whoami(p: Principal = Depends(principal)) -> Principal:
    return p
```

### Cancellation-safe FastAPI lifespan.
**Concept:** Own shared asynchronous resources in one lifecycle context.

The `finally` block closes the resource during normal shutdown or cancellation. Sources: S2, S8.

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI

class Client:
    async def close(self) -> None:
        print("closed")

@asynccontextmanager
async def lifespan(app: FastAPI):
    client = Client()
    app.state.client = client
    try:
        yield
    finally:
        await client.close()

app = FastAPI(lifespan=lifespan)

@app.get("/healthz")
async def healthz():
    return {"ok": True}
```

## Research Sources

### S2: FastAPI documentation
- URL: https://fastapi.tiangolo.com/
- Publisher: FastAPI
- Type: official-doc
- Official: yes
- Relevance: Documents current validation, dependency, and route-handler patterns.
- Summary: Official FastAPI framework documentation.

### S3: OWASP Secrets Management Cheat Sheet
- URL: https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html
- Publisher: OWASP Foundation
- Type: engineering-article
- Official: yes
- Relevance: Supports recommendations about key storage, rotation, and least privilege.
- Summary: Security guidance for managing application credentials.

### S5: RFC 9110: HTTP Semantics
- URL: https://www.rfc-editor.org/rfc/rfc9110
- Publisher: RFC Editor
- Type: standard
- Official: yes
- Published: 2022-06-01T00:00:00.000Z
- Relevance: Provides normative evidence for status, method, and caching behavior.
- Summary: The current core specification for HTTP semantics.

### S8: Python 3.14 task cancellation guidance
- URL: https://docs.python.org/3.14/library/asyncio-task.html#task-cancellation
- Publisher: Python Software Foundation
- Type: official-doc
- Official: yes
- Relevance: Defines the cancellation behavior relevant to setup and shutdown paths.
- Summary: Documents cancellation delivery, CancelledError propagation, and cleanup guidance for asyncio tasks.

### S11: fastapi/fastapi 0.141.1
- URL: https://github.com/fastapi/fastapi/releases/tag/0.141.1
- Publisher: fastapi
- Type: release-note
- Official: yes
- Published: 2026-07-29T17:17:26Z
- Relevance: Provides primary release timing and migration context.
- Summary: Latest stable GitHub release: 0.141.1.

### S12: fastapi/fastapi repository
- URL: https://github.com/fastapi/fastapi
- Publisher: fastapi
- Type: github
- Official: yes
- Published: 2026-08-26T17:54:56Z
- Relevance: Primary implementation repository, active as of 2026-08-26T17:54:56Z.
- Summary: FastAPI framework, high performance, easy to learn, fast to code, ready for production

### S13: FastAPI - Bigger Applications
- URL: https://fastapi.tiangolo.com/tutorial/bigger-applications/
- Publisher: FastAPI
- Type: official-doc
- Official: yes
- Relevance: Documents separating larger FastAPI applications into routers, dependencies, and modules.
- Summary: Official guidance for structuring a FastAPI application across multiple files.

### S14: Alembic Tutorial
- URL: https://alembic.sqlalchemy.org/en/latest/tutorial.html
- Publisher: SQLAlchemy
- Type: official-doc
- Official: yes
- Relevance: Documents version-controlled relational database migration scripts.
- Summary: Official Alembic migration workflow and project structure.

### S15: pgvector documentation
- URL: https://github.com/pgvector/pgvector#filtering
- Publisher: pgvector
- Type: official-doc
- Official: yes
- Relevance: Documents filtered approximate search, iterative scans, partial indexes, partitioning, and model-scoped vector indexes.
- Summary: Primary pgvector implementation and operational guidance.

### S16: OpenAI embeddings API
- URL: https://developers.openai.com/api/reference/resources/embeddings/methods/create
- Publisher: OpenAI
- Type: official-doc
- Official: yes
- Relevance: Defines the embedding model and output-dimensions contract used by versioned embeddings.
- Summary: Official embeddings API reference.

### S17: OpenTelemetry Python instrumentation
- URL: https://opentelemetry.io/docs/languages/python/instrumentation/
- Publisher: OpenTelemetry
- Type: official-doc
- Official: yes
- Relevance: Documents application telemetry setup for traces and metrics.
- Summary: Official Python observability instrumentation guidance.

## Submitted Code

```python
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);
"""

```