# ── Direct command ─────────────────────────────────────────────────────────
# -w = number of worker processes
# -k = worker class (UvicornWorker = async ASGI worker)
# --bind = address:port
gunicorn main:app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 60 \
  --graceful-timeout 30 \
  --keep-alive 5

# Worker count rule of thumb:
# CPU-bound: workers = CPU_cores + 1
# I/O-bound (typical FastAPI): workers = CPU_cores * 2 + 1
# With 4 cores → 4*2+1 = 9 workers for an I/O-heavy API

# gunicorn.conf.py — configuration file (cleaner than command-line flags)
import multiprocessing

# Worker settings
worker_class = "uvicorn.workers.UvicornWorker"
workers = multiprocessing.cpu_count() * 2 + 1
threads = 1       # Uvicorn workers are async — threads add no benefit

# Networking
bind = "0.0.0.0:8000"
backlog = 2048    # max pending connections before refusing

# Timeouts (seconds)
timeout = 60           # kill worker if it doesn't respond in 60s
graceful_timeout = 30  # wait 30s for in-flight requests on shutdown
keepalive = 5          # seconds to keep idle connections open

# Process management
max_requests = 1000        # restart worker after 1000 requests (prevents leaks)
max_requests_jitter = 100  # +/- 100 random jitter (avoids all workers restarting simultaneously)
preload_app = True         # load app code once in master, copy-on-write to workers

# Logging
accesslog = "-"     # log to stdout (captured by Docker/K8s)
errorlog  = "-"
loglevel  = "info"

def on_starting(server):
    print("Gunicorn starting")

def worker_init(worker):
    print(f"Worker {worker.pid} initialized")

def worker_exit(worker, server):
    print(f"Worker {worker.pid} exiting")

# ── Stage 1: build dependencies ───────────────────────────────────────────
FROM python:3.12-slim AS builder

WORKDIR /build

RUN pip install --upgrade pip

COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# ── Stage 2: production image ──────────────────────────────────────────────
FROM python:3.12-slim AS production

# Create non-root user — never run as root in production
RUN addgroup --system app && adduser --system --group app

WORKDIR /app

COPY --from=builder /install /usr/local
COPY --chown=app:app . .

USER app

EXPOSE 8000

# Use exec form — PID 1 receives signals correctly
CMD ["gunicorn", "main:app", \
     "-k", "uvicorn.workers.UvicornWorker", \
     "-w", "4", \
     "--bind", "0.0.0.0:8000", \
     "--timeout", "60"]

# docker-compose.yml — local development stack
version: "3.9"

services:
  api:
    build:
      context: .
      target: production
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/mydb
      - REDIS_URL=redis://redis:6379
      - SECRET_KEY=dev-secret-key
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    volumes:
      - .:/app
    command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d mydb"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  postgres_data:

from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import PostgresDsn, RedisDsn, SecretStr, AnyHttpUrl
from functools import lru_cache
from typing import Literal

class Settings(BaseSettings):
    # ── App ────────────────────────────────────────────────────────────────
    app_name: str = "My FastAPI App"
    environment: Literal["development", "staging", "production"] = "development"
    debug: bool = False
    log_level: str = "INFO"

    # ── Security ───────────────────────────────────────────────────────────
    # SecretStr prevents the value from appearing in repr/logs
    secret_key: SecretStr
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    # ── Database ───────────────────────────────────────────────────────────
    # PostgresDsn validates the URL format
    database_url: PostgresDsn
    db_pool_size: int = 10
    db_max_overflow: int = 20

    # ── Redis (optional) ──────────────────────────────────────────────────
    redis_url: RedisDsn | None = None

    # ── CORS ───────────────────────────────────────────────────────────────
    allowed_origins: list[AnyHttpUrl] = ["http://localhost:3000"]

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    @property
    def is_production(self) -> bool:
        return self.environment == "production"

    @property
    def db_url_str(self) -> str:
        return str(self.database_url)

# lru_cache: settings are read ONCE and cached for the app lifetime.
@lru_cache
def get_settings() -> Settings:
    return Settings()

from fastapi import Depends
from typing import Annotated

SettingsDep = Annotated[Settings, Depends(get_settings)]

# Override in tests:
# def get_test_settings():
#     return Settings(database_url="postgresql+asyncpg://user:pass@localhost/test", ...)
# app.dependency_overrides[get_settings] = get_test_settings

import structlog, logging, sys
from contextvars import ContextVar
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

# ── Configure structlog ───────────────────────────────────────────────────
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,   # inject bound context vars
        structlog.stdlib.add_log_level,            # add "level" field
        structlog.stdlib.add_logger_name,          # add "logger" field
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.JSONRenderer(),       # final output as JSON
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
)

log = structlog.get_logger()

# ── Middleware: bind request context to all logs ──────────────────────────
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        import uuid
        req_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))

        structlog.contextvars.bind_contextvars(
            request_id=req_id,
            method=request.method,
            path=str(request.url.path),
        )

        response = await call_next(request)
        log.info("request_complete", status_code=response.status_code)

        structlog.contextvars.clear_contextvars()
        response.headers["X-Request-ID"] = req_id
        return response

app = FastAPI()
app.add_middleware(LoggingMiddleware)

# ── Usage in route handlers ───────────────────────────────────────────────
@app.post("/users")
async def create_user(name: str, email: str):
    log.info("creating_user", name=name, email=email)
    user_id = 42
    structlog.contextvars.bind_contextvars(user_id=user_id)
    log.info("user_created", user_id=user_id)
    return {"id": user_id}

# Output (one JSON line per log — queryable in Datadog/CloudWatch):
# {"event":"creating_user","name":"Alice","email":"a@b.com",
#  "request_id":"abc-123","method":"POST","path":"/users",
#  "level":"info","timestamp":"2024-01-15T10:30:00Z"}

# pip install opentelemetry-sdk opentelemetry-instrumentation-fastapi
#             opentelemetry-exporter-otlp opentelemetry-instrumentation-sqlalchemy
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.resources import Resource
from fastapi import FastAPI

# ── 1. Configure the tracer ───────────────────────────────────────────────
resource = Resource.create({"service.name": "my-fastapi-service", "deployment.environment": "production"})

exporter = OTLPSpanExporter(endpoint="http://otel-collector:4317", insecure=True)

provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

app = FastAPI()

# ── 2. Auto-instrument FastAPI ────────────────────────────────────────────
# Automatically creates spans for every route with HTTP attributes.
FastAPIInstrumentor.instrument_app(app)

# ── 3. Manual spans for custom operations ─────────────────────────────────
@app.get("/users/{uid}")
async def get_user(uid: int):
    with tracer.start_as_current_span("fetch_user_from_db") as span:
        span.set_attribute("user.id", uid)
        span.set_attribute("db.operation", "SELECT")

        user = {"id": uid, "name": "Alice"}

        if not user:
            span.set_status(trace.StatusCode.ERROR, "User not found")
            span.record_exception(ValueError(f"User {uid} not found"))

    return user

# ── 4. Propagate trace context in outgoing HTTP calls ─────────────────────
# HTTPXClientInstrumentor().instrument()  # auto-propagates W3C traceparent header

from fastapi import FastAPI, Response
from pydantic import BaseModel
from typing import Literal
import asyncio

app = FastAPI()

# ── Liveness probe ────────────────────────────────────────────────────────
# Just proves the process is alive. K8s restarts pod if this fails.
# Keep it FAST — no DB calls.
@app.get("/health/live", include_in_schema=False)
async def liveness():
    return {"status": "alive"}

# ── Readiness probe ───────────────────────────────────────────────────────
class DependencyStatus(BaseModel):
    status: Literal["ok", "error"]
    latency_ms: float | None = None
    error: str | None = None

class ReadinessResponse(BaseModel):
    status: Literal["ready", "degraded", "not_ready"]
    dependencies: dict[str, DependencyStatus]

@app.get("/health/ready", response_model=ReadinessResponse, include_in_schema=False)
async def readiness(response: Response):
    checks: dict[str, DependencyStatus] = {}

    # Check database
    try:
        import time
        start = time.perf_counter()
        await asyncio.sleep(0.001)   # replace with: pool.fetchval("SELECT 1")
        checks["database"] = DependencyStatus(
            status="ok",
            latency_ms=round((time.perf_counter() - start) * 1000, 2),
        )
    except Exception as e:
        checks["database"] = DependencyStatus(status="error", error=str(e))

    # Check Redis
    try:
        start = time.perf_counter()
        # await redis.ping()
        checks["redis"] = DependencyStatus(
            status="ok",
            latency_ms=round((time.perf_counter() - start) * 1000, 2),
        )
    except Exception as e:
        checks["redis"] = DependencyStatus(status="error", error=str(e))

    all_ok = all(v.status == "ok" for v in checks.values())
    critical_ok = checks.get("database", DependencyStatus(status="error")).status == "ok"

    if all_ok:
        overall, response.status_code = "ready", 200
    elif critical_ok:
        overall, response.status_code = "degraded", 200   # still serve traffic
    else:
        overall, response.status_code = "not_ready", 503  # remove from load balancer

    return ReadinessResponse(status=overall, dependencies=checks)

# ── Startup probe (K8s 1.16+) ─────────────────────────────────────────────
@app.get("/health/startup", include_in_schema=False)
async def startup_probe():
    return {"initialized": True}

# ── WRONG: stateful designs that break horizontal scaling ─────────────────

# ❌ In-memory session store — only visible on THIS pod
in_memory_sessions: dict[str, dict] = {}

@app.post("/login-bad")
async def login_bad(username: str):
    import uuid
    session_id = str(uuid.uuid4())
    in_memory_sessions[session_id] = {"username": username}  # lost on restart!
    return {"session_id": session_id}

# ❌ Local file storage — only on THIS pod's filesystem
@app.post("/upload-bad")
async def upload_bad(file: bytes):
    with open(f"/tmp/{uuid.uuid4()}.bin", "wb") as f:
        f.write(file)   # other pods can't access /tmp on this pod
    return {"saved": True}

# ── CORRECT: stateless designs ────────────────────────────────────────────

# ✅ JWT — self-contained token, no server-side state needed
from jose import jwt as jose_jwt

@app.post("/login")
async def login(username: str, secret: str = "secret-key"):
    token = jose_jwt.encode(
        {"sub": username, "exp": 9999999999},
        secret, algorithm="HS256"
    )
    return {"access_token": token}

# ✅ Redis-backed sessions (shared across pods)
# async def get_session(session_id: str, redis) -> dict | None:
#     data = await redis.get(f"session:{session_id}")
#     return json.loads(data) if data else None

# ✅ S3/object storage for files (shared across pods)
# async def upload_file(filename: str, data: bytes, s3_client) -> str:
#     await s3_client.put_object(Bucket="my-bucket", Key=filename, Body=data)
#     return f"s3://my-bucket/{filename}"

# ── Stateless checklist ───────────────────────────────────────────────────
# ✅ Auth: JWT (stateless) or Redis sessions
# ✅ Rate limits: Redis counters (not in-memory)
# ✅ Caches: Redis (not dict)
# ✅ Files: S3 / GCS / Azure Blob (not local disk)
# ✅ Background jobs: task queue (Celery/ARQ) — not asyncio tasks on one pod
# ✅ WebSocket connections: Redis Pub/Sub for cross-pod messaging (see topic 52)
# ✅ App config: env vars (not files baked into image with local state)

# ── Option 1: FastAPI BackgroundTasks (simple, in-process) ────────────────
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email_sync(email: str, message: str):
    """Runs after response is sent — in the same process."""
    print(f"Sending email to {email}: {message}")

@app.post("/register-simple")
async def register_simple(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email_sync, email, "Welcome!")
    return {"status": "registered"}

# ── Option 2: ARQ (async, Redis-backed, production-grade) ─────────────────
# pip install arq redis
import arq

async def send_welcome_email(ctx: dict, email: str, name: str):
    """ARQ task — ctx contains the Redis connection and settings."""
    print(f"Sending welcome email to {email}")
    return {"sent_to": email}

async def generate_report(ctx: dict, report_id: int):
    print(f"Generating report {report_id}")
    return {"report_id": report_id, "status": "complete"}

class WorkerSettings:
    functions = [send_welcome_email, generate_report]
    redis_settings = arq.connections.RedisSettings(host="localhost", port=6379)
    max_jobs = 10
    job_timeout = 300
    keep_result = 3600

@app.post("/register")
async def register(email: str, name: str):
    redis = await arq.create_pool(arq.connections.RedisSettings())
    await redis.enqueue_job("send_welcome_email", email, name)
    return {"status": "registered", "email_queued": True}

# Run worker: arq app.jobs.WorkerSettings

# ── Option 3: Celery (mature, multi-language workers) ────────────────────
# from celery import Celery
# celery_app = Celery("tasks", broker="redis://localhost:6379", backend="redis://localhost:6379")
#
# @celery_app.task(bind=True, max_retries=3)
# def send_email_task(self, email: str, message: str):
#     try:
#         pass  # ... send email ...
#     except Exception as exc:
#         raise self.retry(exc=exc, countdown=60)

import asyncio, json
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import redis.asyncio as aioredis

app = FastAPI()

# ── Local connection manager (per-pod) ───────────────────────────────────
class LocalConnectionManager:
    def __init__(self):
        # room_id -> set of WebSocket connections ON THIS POD
        self.rooms: dict[str, set[WebSocket]] = {}

    async def connect(self, ws: WebSocket, room: str):
        await ws.accept()
        self.rooms.setdefault(room, set()).add(ws)

    def disconnect(self, ws: WebSocket, room: str):
        if room in self.rooms:
            self.rooms[room].discard(ws)

    async def broadcast_local(self, room: str, message: str):
        """Send to all WebSocket connections on THIS pod."""
        for ws in list(self.rooms.get(room, set())):
            try:
                await ws.send_text(message)
            except Exception:
                self.disconnect(ws, room)

manager = LocalConnectionManager()
redis_client = aioredis.from_url("redis://localhost:6379")

async def redis_listener():
    """Subscribe to Redis and forward messages to local connections."""
    pubsub = redis_client.pubsub()
    await pubsub.psubscribe("room:*")   # subscribe to all room channels

    async for message in pubsub.listen():
        if message["type"] == "pmessage":
            channel = message["channel"].decode()
            room_id = channel.split(":", 1)[1]
            data = message["data"].decode()
            await manager.broadcast_local(room_id, data)

@app.on_event("startup")
async def start_redis_listener():
    asyncio.create_task(redis_listener())

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(ws: WebSocket, room_id: str):
    await manager.connect(ws, room_id)
    try:
        while True:
            data = await ws.receive_text()
            # Publish to Redis — ALL pods subscribed to this room receive it
            payload = json.dumps({"room": room_id, "message": data})
            await redis_client.publish(f"room:{room_id}", payload)
    except WebSocketDisconnect:
        manager.disconnect(ws, room_id)

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from typing import AsyncGenerator

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"

engine = create_async_engine(
    DATABASE_URL,
    # ── Pool size math ────────────────────────────────────────────────────
    # Total max per worker = pool_size + max_overflow
    # With 4 Gunicorn workers: total DB connections = 4 × (5 + 10) = 60
    pool_size=5,
    max_overflow=10,

    pool_timeout=30,       # seconds to wait for a free connection before error
    pool_recycle=1800,     # recycle connections after 30min (prevents stale state)
    pool_pre_ping=True,    # test connection health before lending from pool

    echo=False,            # set True to log all SQL (dev only — noisy in prod)
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
    autocommit=False,
)

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# ── Pool sizing formula ────────────────────────────────────────────────────
# Max DB connections = pool_size × gunicorn_workers
# PostgreSQL default max_connections = 100
# With 4 workers:
#   pool_size=20 → 4 × 20 = 80 connections (safe)
#   pool_size=30 → 4 × 30 = 120 → ERROR: too many connections

def pool_status() -> dict:
    pool = engine.pool
    return {
        "size": pool.size(),
        "checked_in": pool.checkedin(),
        "checked_out": pool.checkedout(),
        "overflow": pool.overflow(),
    }

import asyncio
from fastapi import FastAPI

app = FastAPI()

# ── 1. asyncio.gather — run multiple tasks concurrently ───────────────────
@app.get("/dashboard")
async def dashboard(user_id: int):
    # Without gather: sequential, ~300ms total
    # With gather: concurrent, ~100ms total (slowest wins)
    user, orders, notifications = await asyncio.gather(
        fetch_user(user_id),
        fetch_orders(user_id),
        fetch_notifications(user_id),
    )
    return {"user": user, "orders": orders, "notifications": notifications}

# ── 2. gather with error handling ─────────────────────────────────────────
@app.get("/dashboard-safe")
async def dashboard_safe(user_id: int):
    # return_exceptions=True: failed tasks return the exception, not raise it
    results = await asyncio.gather(
        fetch_user(user_id),
        fetch_orders(user_id),
        fetch_notifications(user_id),
        return_exceptions=True,
    )
    return {
        "user":          results[0] if not isinstance(results[0], Exception) else None,
        "orders":        results[1] if not isinstance(results[1], Exception) else [],
        "notifications": results[2] if not isinstance(results[2], Exception) else [],
    }

# ── 3. asyncio.wait_for — timeout on slow operations ─────────────────────
@app.get("/slow-service")
async def call_slow_service():
    try:
        result = await asyncio.wait_for(slow_external_api(), timeout=2.0)
        return result
    except asyncio.TimeoutError:
        from fastapi import HTTPException
        raise HTTPException(504, "Upstream service timed out")

# ── 4. Semaphore — limit concurrency ──────────────────────────────────────
_db_semaphore = asyncio.Semaphore(10)

async def fetch_with_limit(item_id: int):
    async with _db_semaphore:   # blocks here if 10 are already running
        return await fetch_item(item_id)

@app.get("/bulk")
async def bulk_fetch(ids: list[int]):
    return await asyncio.gather(*[fetch_with_limit(i) for i in ids])

# ── 5. TaskGroup (Python 3.11+) — structured concurrency ────────────────
@app.get("/taskgroup/{user_id}")
async def with_task_group(user_id: int):
    async with asyncio.TaskGroup() as tg:
        user_task   = tg.create_task(fetch_user(user_id))
        orders_task = tg.create_task(fetch_orders(user_id))
        # If ANY task raises, all others are CANCELLED immediately
    return {"user": user_task.result(), "orders": orders_task.result()}

async def fetch_user(uid): await asyncio.sleep(0.1); return {"id": uid}
async def fetch_orders(uid): await asyncio.sleep(0.05); return []
async def fetch_notifications(uid): await asyncio.sleep(0.08); return []
async def slow_external_api(): await asyncio.sleep(5); return {}
async def fetch_item(item_id): await asyncio.sleep(0.01); return {"id": item_id}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: fastapi-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: fastapi-app
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1        # allow 1 extra pod during rollout
      maxUnavailable: 0  # never have fewer than desired count
  template:
    metadata:
      labels:
        app: fastapi-app
    spec:
      containers:
      - name: fastapi
        image: myrepo/fastapi-app:v1.2.3   # always use specific tags, never "latest"
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health/live
            port: 8000
          initialDelaySeconds: 15
          periodSeconds: 10
          failureThreshold: 3
        readinessProbe:
          httpGet:
            path: /health/ready
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
          failureThreshold: 3
        env:
        - name: SECRET_KEY
          valueFrom:
            secretKeyRef:
              name: fastapi-secrets
              key: secret-key
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: fastapi-secrets
              key: database-url
---
apiVersion: v1
kind: Service
metadata:
  name: fastapi-service
spec:
  selector:
    app: fastapi-app
  ports:
  - port: 80
    targetPort: 8000
  type: ClusterIP
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: fastapi-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: fastapi-app
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

name: CI/CD

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
          POSTGRES_DB: testdb
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 5s
          --health-timeout 5s
          --health-retries 5

      redis:
        image: redis:7-alpine
        ports:
          - 6379:6379

    steps:
    - uses: actions/checkout@v4

    - uses: actions/setup-python@v5
      with:
        python-version: "3.12"
        cache: pip

    - name: Install dependencies
      run: pip install -r requirements.txt -r requirements-dev.txt

    - name: Run Alembic migrations
      env:
        DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/testdb
      run: alembic upgrade head

    - name: Run tests with coverage
      env:
        DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/testdb
        REDIS_URL: redis://localhost:6379
        SECRET_KEY: test-secret-key-ci
      run: pytest --cov=app --cov-report=xml -v

    - name: Upload coverage
      uses: codecov/codecov-action@v4

  build-and-push:
    needs: test
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v4

    - name: Login to Container Registry
      uses: docker/login-action@v3
      with:
        registry: ghcr.io
        username: ${{ github.actor }}
        password: ${{ secrets.GITHUB_TOKEN }}

    - name: Build and push
      uses: docker/build-push-action@v5
      with:
        context: .
        push: true
        tags: |
          ghcr.io/${{ github.repository }}:${{ github.sha }}
          ghcr.io/${{ github.repository }}:latest
        cache-from: type=gha
        cache-to: type=gha,mode=max

from fastapi import FastAPI, APIRouter, Header, HTTPException
from typing import Annotated

# ── Approach 1: URL path versioning ───────────────────────────────────────
app = FastAPI()

v1_router = APIRouter(prefix="/v1")
v2_router = APIRouter(prefix="/v2")

@v1_router.get("/users/{uid}")
async def get_user_v1(uid: int):
    return {"id": uid, "name": "Alice"}

@v2_router.get("/users/{uid}")
async def get_user_v2(uid: int):
    return {"id": uid, "name": "Alice", "created_at": "2024-01-15", "role": "user"}

app.include_router(v1_router)
app.include_router(v2_router)

# ── Approach 2: Header versioning ─────────────────────────────────────────
@app.get("/users/{uid}")
async def get_user_versioned(
    uid: int,
    api_version: Annotated[str, Header(alias="API-Version")] = "1",
):
    if api_version == "2":
        return {"id": uid, "name": "Alice", "created_at": "2024-01-15"}
    elif api_version == "1":
        return {"id": uid, "name": "Alice"}
    raise HTTPException(400, f"Unsupported API version: {api_version}")

# ── Approach 3: Mount separate apps (maximum isolation) ──────────────────
v1_app = FastAPI(title="API v1", docs_url="/docs")
v2_app = FastAPI(title="API v2", docs_url="/docs")

@v1_app.get("/users/{uid}")
async def v1_users(uid: int):
    return {"id": uid, "format": "v1"}

@v2_app.get("/users/{uid}")
async def v2_users(uid: int):
    return {"id": uid, "format": "v2", "extra": True}

main_app = FastAPI(docs_url=None)
main_app.mount("/v1", v1_app)
main_app.mount("/v2", v2_app)

# ── Version deprecation strategy ─────────────────────────────────────────
from starlette.middleware.base import BaseHTTPMiddleware

class DeprecationWarningMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        if request.url.path.startswith("/v1/"):
            response.headers["Deprecation"] = "true"
            response.headers["Sunset"] = "Sat, 31 Dec 2025 23:59:59 GMT"
            response.headers["Link"] = '</v2/>; rel="successor-version"'
        return response

from fastapi import FastAPI, Request, Response
from fastapi.responses import ORJSONResponse
import pyinstrument

# ── Use ORJSONResponse globally (5-10x faster JSON serialization) ─────────
app = FastAPI(default_response_class=ORJSONResponse)

# ── 1. Profiling middleware (enable in dev/staging only) ───────────────────
class ProfilingMiddleware:
    def __init__(self, app, profile_path: str = "/_profile"):
        self.app = app
        self.profile_path = profile_path

    async def __call__(self, scope, receive, send):
        if scope.get("path") == self.profile_path or \
           scope.get("headers", {}).get(b"x-profile") == b"true":
            profiler = pyinstrument.Profiler()
            profiler.start()
            await self.app(scope, receive, send)
            profiler.stop()
            html = profiler.output_html()
            response = Response(content=html, media_type="text/html")
            await response(scope, receive, send)
        else:
            await self.app(scope, receive, send)

# app.add_middleware(ProfilingMiddleware)  # enable for profiling sessions

# ── 2. N+1 query problem and fix ──────────────────────────────────────────
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from sqlalchemy.orm import selectinload

# ❌ N+1: 1 query for users + N queries for each user's items
async def get_users_bad(session: AsyncSession):
    users = (await session.execute(select(User))).scalars().all()
    for user in users:
        items = user.items   # triggers lazy load — 100 users = 101 queries!

# ✅ Eager loading: 1 query with JOIN (or 2 queries total)
async def get_users_good(session: AsyncSession):
    users = (await session.execute(
        select(User).options(
            selectinload(User.items)   # loads all users' items in ONE extra query
        )
    )).scalars().all()
    return users

# ── 3. Enable SQL query logging (find slow queries) ───────────────────────
import logging
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

# ── 4. Route-level profiling ──────────────────────────────────────────────
@app.get("/profile-me")
async def profile_me():
    profiler = pyinstrument.Profiler(async_mode="enabled")
    with profiler:
        result = await some_expensive_operation()
    print(profiler.output_text(unicode=True, color=True))
    return result

async def some_expensive_operation():
    import asyncio; await asyncio.sleep(0.01); return {"done": True}