I Built a Production-Grade AI Platform From Scratch (Here’s the Exact Folder Structure)

genai-platform/
├── .gitignore                     # what git never tracks
├── .env.example                   # env var template — committed
├── README.md                      # project front door
├── .github/
│   └── workflows/
│       ├── ci.yml                 # runs on every push
│       └── deploy.yml             # runs on main merge only
├── infrastructure/
│   └── docker-compose.yml         # PostgreSQL + Redis + Backend
└── backend/
    ├── requirements.txt           # pinned Python dependencies
    ├── pyproject.toml             # ruff + mypy + pytest config
    ├── Dockerfile                 # multi-stage production build
    ├── alembic.ini                # migration configuration
    ├── alembic/
    │   ├── env.py                 # async migration bridge
    │   └── versions/              # migration files (Phase 2+)
    └── app/
        ├── __init__.py            # package marker + version
        ├── config.py              # Pydantic Settings — all env vars
        ├── main.py                # FastAPI app + lifespan + middleware
        ├── dependencies.py        # get_db, get_redis, get_current_user
        ├── db/
        │   ├── database.py        # async engine + session factory
        │   └── init_db.py         # pgvector extension + admin seed
        ├── monitoring/
        │   ├── logger.py          # structured JSON logging (structlog)
        │   └── metrics.py         # 12 Prometheus metrics defined
        └── api/v1/
            └── health.py          # /health (liveness) + /ready (readiness)

# The pattern every production codebase uses
.env        # real secrets — never committed
.env.*      # covers .env.local, .env.production, etc.
!.env.example  # ← the ! means EXCEPTION — template IS committed

cp .env.example .env
# then fill in real values

uploads/
*.pdf
*.docx
*.pptx
*.csv

# scattered across 15 files — dangerous
db_url = os.getenv("DATABASE_URL")         # returns None silently if missing
secret = os.getenv("SECRIT_KEY")           # typo — also None, no warning
expire = int(os.getenv("TOKEN_EXPIRE"))    # crashes here if None

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, field_validator
from functools import lru_cache

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        extra="ignore",
    )

    # Required — app refuses to start without these
    DATABASE_URL: str = Field(...)
    SECRET_KEY: str = Field(...)
    GROQ_API_KEY: str = Field(...)

    # Optional — typed, with defaults
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 15   # "15" → int automatically
    DEBUG: bool = False                      # "true" → bool automatically

    # Custom validators — fail fast with clear messages
    @field_validator("SECRET_KEY")
    @classmethod
    def validate_secret_key(cls, v: str) -> str:
        if len(v) < 32:
            raise ValueError(
                "SECRET_KEY must be at least 32 characters. "
                "Generate with: openssl rand -hex 32"
            )
        return v

    @field_validator("DATABASE_URL")
    @classmethod
    def validate_database_url(cls, v: str) -> str:
        if not v.startswith("postgresql+asyncpg://"):
            raise ValueError(
                "DATABASE_URL must use asyncpg driver. "
                "Change postgresql:// to postgresql+asyncpg://"
            )
        return v

@lru_cache()
def get_settings() -> Settings:
    return Settings()

settings = get_settings()  # read once, cached forever

ValidationError: DATABASE_URL field required

ValidationError: SECRET_KEY must be at least 32 characters.
Generate with: openssl rand -hex 32

# app/db/database.py
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)
from sqlalchemy.orm import DeclarativeBase

engine = create_async_engine(
    url=settings.DATABASE_URL,  # postgresql+asyncpg:// — MUST be asyncpg
    pool_size=20,                # 20 connections in pool
    max_overflow=10,             # 10 extra when pool is full
    pool_pre_ping=True,          # test before using (prevents stale connections)
    echo=settings.DEBUG,         # log SQL in development
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    expire_on_commit=False,  # ← CRITICAL — more on this below
    class_=AsyncSession,
    autoflush=False,
)

class Base(DeclarativeBase):
    pass

user = await db.get(User, user_id)
await db.commit()
print(user.email)   # ← CRASH in async: MissingGreenlet error

user = await db.get(User, user_id)
await db.commit()
print(user.email)   # ✅ works — values kept in memory

ValidationError: DATABASE_URL must use asyncpg driver.

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ── STARTUP ──────────────────────────────────────────────
    setup_logging()        # 1. logging first — everything else logs
    setup_prometheus()     # 2. metrics
    await init_db()        # 3. DB — needs logging ready
    connect_redis()        # 4. Redis

    yield  # ← app handles requests here

    # ── SHUTDOWN ─────────────────────────────────────────────
    await redis.aclose()
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

Logging → Prometheus → Database → Redis → Ready

# Layer 1: CORS — browsers need this to call your API
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=["*"],
)

# Layer 2: Request ID — every request gets a unique ID
@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    structlog.contextvars.bind_contextvars(request_id=request_id)
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    structlog.contextvars.clear_contextvars()
    return response

# Layer 3: Request Logging — every request logged automatically
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration_ms = (time.perf_counter() - start) * 1000
    log.info("request_completed",
             method=request.method,
             path=request.url.path,
             status_code=response.status_code,
             duration_ms=round(duration_ms, 2))
    return response

# app/dependencies.py
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session        # route runs with this session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        # session closes automatically — even on exception

async def get_redis(request: Request):
    return request.app.state.redis

@router.get("/documents")
async def list_documents(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    # db is ready, current_user is verified
    # no setup code needed here
    ...

app.dependency_overrides[get_db] = override_get_db  # swap real DB for test DB

# GET /api/v1/health — liveness: is the process alive?
# NEVER checks external services
@router.get("/health")
async def health_check():
    return {"status": "ok", "version": __version__}

# GET /api/v1/ready — readiness: can this instance handle traffic?
# Checks ALL dependencies
@router.get("/ready")
async def readiness_check(request: Request):
    checks = {}
    all_ready = True

    try:
        async with engine.connect() as conn:
            await conn.execute(text("SELECT 1"))
        checks["database"] = {"status": "ok"}
    except Exception as e:
        all_ready = False
        checks["database"] = {"status": "error", "error": str(e)}

    try:
        await request.app.state.redis.ping()
        checks["redis"] = {"status": "ok"}
    except Exception as e:
        all_ready = False
        checks["redis"] = {"status": "error"}

    return JSONResponse(
        status_code=200 if all_ready else 503,
        content={"status": "ready" if all_ready else "not_ready",
                 "checks": checks},
    )

# app/monitoring/logger.py
import structlog

def setup_logging():
    structlog.configure(
        processors=[
            structlog.stdlib.add_log_level,
            structlog.stdlib.add_logger_name,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.contextvars.merge_contextvars,  # includes request_id
            structlog.processors.JSONRenderer() if not settings.DEBUG
            else structlog.dev.ConsoleRenderer(colors=True),
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        logger_factory=structlog.stdlib.LoggerFactory(),
    )

2025-03-14 10:30:00 [info] document_uploaded  filename=report.pdf size_mb=2.4

{"event":"document_uploaded","filename":"report.pdf","size_mb":2.4,
 "request_id":"abc-123","level":"info","timestamp":"2025-03-14T10:30:00Z"}

log = structlog.get_logger(__name__)
log.info("chunks_created", count=47, strategy="parent_child", duration_ms=234)
log.error("llm_failed", provider="groq", error=str(e), exc_info=True)

# Stage 1 — Builder (large, temporary)
FROM python:3.12-slim AS builder
WORKDIR /build

RUN apt-get update && apt-get install -y gcc python3-dev libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# requirements.txt BEFORE app code — enables layer caching
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# Stage 2 — Production (small, deployed)
FROM python:3.12-slim AS production

# Runtime dependencies only — no build tools
RUN apt-get update && apt-get install -y libpq5 tesseract-ocr curl \
    && rm -rf /var/lib/apt/lists/*

# Copy ONLY compiled packages — not gcc, not make, not compilers
COPY --from=builder /install /usr/local

# Non-root user — least privilege principle
RUN groupadd -r appgroup && useradd -r -g appgroup appuser
RUN mkdir -p /app/uploads && chown -R appuser:appgroup /app

USER appuser
COPY --chown=appuser:appgroup ./app /app/app

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Things that change RARELY → top of Dockerfile
Things that change OFTEN  → bottom of Dockerfile

# alembic/env.py
async def run_async_migrations() -> None:
    connectable = create_async_engine(settings.DATABASE_URL)

    async with connectable.connect() as connection:
        # run_sync() extracts a sync connection from async
        # Alembic runs inside that sync connection
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()

def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())

jobs:
  lint:      # ruff — style + security issues
  typecheck: # mypy — type errors
  test:      # pytest with real postgres + redis
    services:
      postgres:
        image: pgvector/pgvector:pg16
      redis:
        image: redis:7-alpine

  docker:    # verify image builds
    needs: [lint, typecheck, test]  # only if all pass

# 1. Clone and set up
git clone https://github.com/digvijaysingh21/genai-docqa.git
cd genai-docqa
cp .env.example .env

# 2. Generate secrets
openssl rand -hex 32  # paste as SECRET_KEY
openssl rand -hex 32  # paste as ENCRYPTION_KEY
# Get free Groq key at: console.groq.com → paste as GROQ_API_KEY

# 3. Start everything
cd infrastructure
docker compose up

# 4. Test
curl http://localhost:8000/api/v1/health
# {"status":"ok","version":"1.0.0","environment":"development"}

curl http://localhost:8000/api/v1/ready
# {"status":"ready","checks":{"database":{"status":"ok"},"redis":{"status":"ok"}}}

# 5. Open Swagger UI
open http://localhost:8000/docs