1 · Architecture

The shape of the system

                    ┌──────────────────────────────────┐
   Any HTTP client  │  Next.js demo · docs · your app  │
                    └────────────────┬─────────────────┘
                                     │  REST + SSE
                    ┌────────────────▼─────────────────┐
                    │         FastAPI  (src/api)       │  routes: shapes only
                    ├──────────────────────────────────┤
                    │        Services (src/services)   │  authorise, orchestrate
                    ├──────────────────────────────────┤
                    │       Pipeline (src/pipeline)    │  the 9 stages
                    ├───────────────┬──────────────────┤
                    │  AI (src/ai)  │ Connectors       │  two swappable seams
                    └───────┬───────┴────────┬─────────┘
                            │                │
                  OpenAI / Anthropic /   User's Postgres,
                  Gemini / Ollama /      MySQL, BigQuery,
                  LM Studio / custom     Snowflake, CSV…

   Retail Mind's own state:  PostgreSQL (+pgvector)  ·  Redis  ·  Celery worker

The layering rule

routes → services → pipeline → (ai | connectors)

Each arrow points one way only. Concretely:

• A route never imports a connector.
• The pipeline never imports a vendor SDK (openai, anthropic, google.genai).
• Services never build SQL.

That discipline is what makes six AI providers and six databases swappable without touching the pipeline. It is also the first thing an interviewer probes when you claim “clean architecture” — be ready to name what would break if you violated it.

What each layer is responsible for

Routes — src/api/v1/

Request and response shapes, nothing else. Pydantic models define the contract; the route calls one service method and translates exceptions into status codes.

# src/api/v1/query.py
@router.post("", response_model=QueryResponse)
@limiter.limit(settings.rate_limit_query)
async def execute_query(request: Request, body: QueryRequest,
                        db: DBSession, current_key: CurrentAPIKey):
    try:
        return await QueryService(db).execute(...)
    except QueryAccessError as e:
        raise HTTPException(404, detail=str(e)) from e
    except ProviderQuotaExhausted as e:
        raise HTTPException(503, detail=QUOTA_MESSAGE,
                            headers={"Retry-After": ...}) from e

Validation lives in the Pydantic model, so bad input never reaches a service:

class QueryOptions(BaseModel):
    # Bounded so a caller cannot ask the connector to materialise an
    # unbounded result set into memory.
    max_rows: int = Field(default=1000, ge=1, le=settings.absolute_max_rows)

Services — src/services/

Authorisation and orchestration. QueryService._resolve() is the important one: it loads the data source, its schema snapshot and the AI provider, and enforces ownership, before anything expensive happens.

Ownership is enforced here rather than in the route because both the sync and streaming paths funnel through it. A source the caller does not own is reported as not found, never as forbidden — see Security.

Pipeline — src/pipeline/

The nine stages, and an orchestrator that wires them. State flows through one mutable PipelineContext. Covered in full in The Query Pipeline.

The two seams — src/ai/ and src/connectors/

Both are ABCs with multiple implementations behind a registry (a factory keyed by a type string). This is the Strategy pattern; naming it is worth a point in an interview.

Data stores, and why each exists

The request lifecycle

For POST /v1/query:

1. Middleware — structured logging, then rate limiting keyed on the API key.
2. Dependency injection — FastAPI resolves DBSession (an async session) and CurrentAPIKey (verifies the bearer token against bcrypt hashes).
3. Route — validates the body into QueryRequest.
4. Service — _resolve() authorises and loads source, snapshot, provider; builds a connector.
5. Pipeline — nine stages (below).
6. Service — writes a QueryLog row, then returns a JSON-safe dict.
7. Route — FastAPI serialises through QueryResponse.

Sync vs streaming

Two endpoints over the same pipeline:

The error difference is the one to understand. Once streaming has begun, the HTTP status line is already on the wire — you cannot go back and make it a 500. So failures must arrive as an SSE event:

except ProviderQuotaExhausted:
    yield _sse("error", {"message": QUOTA_MESSAGE,
                         "code": "provider_quota_exhausted"})
    yield _sse("done", {})

Background work

Schema discovery on a large warehouse takes minutes — far too long for a request. It runs on Celery; the API returns a job ID to poll at GET /v1/jobs/{id}.

Two queues, so a slow insight job cannot starve schema sync:

# src/workers/celery_app.py
DEFAULT_QUEUE = "celery"
QUEUES = ("schema", "insight", DEFAULT_QUEUE)
 
task_routes = {
    "src.workers.schema_worker.*":  {"queue": "schema"},
    "src.workers.insight_worker.*": {"queue": "insight"},
}

There is also an inline mode (INLINE_WORKERS=true) that runs jobs in a thread inside the API process, because the free hosting tier has no background workers. The job API is identical either way. See Async, Jobs & Event Loops — the reason it uses a thread rather than await is a genuinely good interview answer.

The monorepo

packages/
├── api/    Python · FastAPI · the product
├── demo/   Next.js 14 app router · interactive playground
└── docs/   Next.js · Nextra · this site

Managed with pnpm workspaces + Turborepo for the JS side; Poetry for Python.

The two web apps are deployed as separate Vercel projects on separate subdomains. An earlier design used Next.js Multi Zones to serve docs under /docs of the demo — worth knowing about, and worth knowing why it was dropped.

Design decisions you should be able to defend

Why API-first rather than a web app? The product is the API. Any language that can make an HTTP request can use it. The demo is a client, not the thing itself — which also forces the API to be genuinely complete, because it has no privileged back channel.

Why a pipeline of stages rather than one function? Each stage is independently testable, individually loggable, and swappable. It also makes the streaming variant possible: you cannot yield progress from the middle of a monolith without threading a callback through it.

Why store schema snapshots instead of introspecting per query? Introspection takes seconds to minutes. Snapshots are versioned rather than overwritten, so a schema change is auditable and a bad sync can be traced.

Why is Retail Mind’s own database PostgreSQL specifically? JSONB for schema snapshots and query metadata, plus pgvector for the embedding-based schema search that is the planned upgrade to keyword scoring.