Backend Live Coding

# ======================================================================
# Q: How do you set up logging so you can find a problem in production?
#
# Structured JSON logs with a request_id/correlation id that passes from the API
# to Celery tasks and to calls to other services. That way I filter all the logs
# of one request in one place (ELK, Loki, CloudWatch).
# I never write passwords, tokens or personal data (GDPR) to logs. I add a
# filter that masks them.
# ======================================================================

import contextvars, logging, uuid
request_id_var = contextvars.ContextVar('request_id', default='-')

class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id_var.get()
        return True

@app.middleware('http')
async def add_request_id(request, call_next):
    rid = request.headers.get('X-Request-ID') or uuid.uuid4().hex
    token = request_id_var.set(rid)
    try:
        response = await call_next(request)
    finally:
        request_id_var.reset(token)
    response.headers['X-Request-ID'] = rid
    return response

# Celery: pass the id in the task headers
process_import.apply_async(args=[import_id], headers={'request_id': request_id_var.get()})

# ======================================================================
# Q: Describe a CI/CD pipeline.
#
# Stages: lint/format → type check → tests with a real PostgreSQL (not SQLite,
# because it behaves differently) → security scan (pip-audit, image scan) →
# build the image tagged with the commit SHA → deploy to staging → smoke tests →
# production with rolling or blue/green. Migrations run as a separate step
# before the new code and are backward compatible, so the old and new code work
# together during the rolling deploy.
# ======================================================================

# .github/workflows/ci.yml
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env: { POSTGRES_PASSWORD: test }
        ports: ['5432:5432']
        options: --health-cmd pg_isready
      redis:
        image: redis:7
        ports: ['6379:6379']
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.12' }
      - run: pip install -r requirements-dev.txt
      - run: ruff check . && ruff format --check .
      - run: mypy app
      - run: alembic upgrade head && alembic check     # model == migrations
      - run: pytest --cov=app --cov-fail-under=80
      - run: pip-audit                                  # known CVEs

# ======================================================================
# Q: Write a production Dockerfile for a Python service.
#
# Multi-stage for a smaller image, a non-root user, pinned versions, a
# .dockerignore (no .env or .git inside the image), PYTHONUNBUFFERED so logs
# show up immediately. Secrets come from env/a secret manager at runtime, never
# at build time. The same image runs as the API, as a Celery worker and as beat,
# with a different command.
# ======================================================================

FROM python:3.12-slim AS build
ENV PIP_NO_CACHE_DIR=1
WORKDIR /app
COPY requirements.txt .
RUN pip wheel --wheel-dir /wheels -r requirements.txt

FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
RUN useradd --create-home --uid 10001 app
WORKDIR /app
COPY --from=build /wheels /wheels
RUN pip install --no-cache-dir /wheels/* && rm -rf /wheels
COPY --chown=app:app . .
USER app                                  # not root
EXPOSE 8000
HEALTHCHECK CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"
CMD ["gunicorn", "app.main:app", "-k", "uvicorn.workers.UvicornWorker",
     "-w", "4", "-b", "0.0.0.0:8000"]

-- ======================================================================
-- Q: Dashboards with aggregations over hundreds of millions of events. What do
-- you do?
--
-- I separate OLTP from OLAP. Steps in order of cost: (1) proper indexes and a
-- BRIN index on created_at for append-only tables, (2) partitioning by month
-- (fast dropping of old data for retention), (3) precomputed rollups (per
-- hour/day) or materialized views, (4) a read replica for reports, (5) a column
-- store such as ClickHouse or TimescaleDB when the volume requires it.
-- ======================================================================

CREATE TABLE events (
    id bigint GENERATED ALWAYS AS IDENTITY,
    tenant_id int NOT NULL,
    severity smallint NOT NULL,
    created_at timestamptz NOT NULL,
    payload jsonb,
    PRIMARY KEY (id, created_at)          -- the partition key inside the PK
) PARTITION BY RANGE (created_at);

CREATE TABLE events_2026_10 PARTITION OF events
    FOR VALUES FROM ('2026-10-01') TO ('2026-11-01');
CREATE INDEX ON events USING BRIN (created_at);

-- hourly rollup, refreshed by a scheduled job
INSERT INTO events_hourly (tenant_id, hour, severity, cnt)
SELECT tenant_id, date_trunc('hour', created_at), severity, count(*)
FROM events WHERE created_at >= :from AND created_at < :to
GROUP BY 1, 2, 3
ON CONFLICT (tenant_id, hour, severity) DO UPDATE SET cnt = EXCLUDED.cnt;

-- ======================================================================
-- Q: You have graph-like relationships (asset → depends on → asset, user →
-- member of → group). Do you need a graph database?
--
-- For hierarchies and a few levels, no: PostgreSQL with a recursive CTE is
-- enough. A graph database (Neo4j, Neptune) is worthwhile when the queries are
-- mostly multi-hop paths (blast radius, shortest path, "who is connected to this
-- one within 4 hops") on a large graph.
-- UNION (not UNION ALL) discards duplicate rows and helps avoid looping forever
-- on cycles. The depth limit is a second safeguard.
-- ======================================================================

-- All assets affected if asset 10 goes down (blast radius)
WITH RECURSIVE impacted AS (
    SELECT dependent_id AS id, 1 AS depth
    FROM asset_dependencies WHERE depends_on_id = 10
  UNION
    SELECT d.dependent_id, i.depth + 1
    FROM asset_dependencies d JOIN impacted i ON d.depends_on_id = i.id
    WHERE i.depth < 6                         -- limit for cycles/cost
)
SELECT a.id, a.hostname, min(i.depth) AS distance
FROM impacted i JOIN assets a ON a.id = i.id
GROUP BY a.id, a.hostname ORDER BY distance;