Ship a containerized web app with a full CI/CD + IaC pipeline.

# Strategy pattern — interchangeable key-generation algorithms
# behind one interface (Sessions 4-6: behavioural design patterns).
from __future__ import annotations
import secrets
import string
from typing import Protocol

ALPHABET = string.ascii_letters + string.digits


class KeyStrategy(Protocol):
    """The stable contract every algorithm must honour."""
    def generate(self) -> str: ...


class RandomKey:
    """Cryptographically-random short key. Good default: no coordination needed."""
    def __init__(self, length: int = 7) -> None:
        self.length = length

    def generate(self) -> str:
        return "".join(secrets.choice(ALPHABET) for _ in range(self.length))


class CounterKey:
    """Deterministic base-62 of a monotonic counter. Shortest keys, but needs a source of truth."""
    def __init__(self, start: int = 1000) -> None:
        self._n = start

    def generate(self) -> str:
        n, out = self._n, []
        self._n += 1
        if n == 0:
            return ALPHABET[0]
        while n > 0:
            n, r = divmod(n, len(ALPHABET))
            out.append(ALPHABET[r])
        return "".join(reversed(out))


def get_strategy(name: str) -> KeyStrategy:
    """Factory: pick the strategy by name (driven by config, not hard-coded)."""
    strategies = {"random": RandomKey, "counter": CounterKey}
    if name not in strategies:
        raise ValueError(f"unknown key strategy: {name!r}")
    return strategies[name]()

# The HTTP layer. Depends on the KeyStrategy interface, not a concrete class.
from flask import Blueprint, request, jsonify, redirect, current_app

api = Blueprint("api", __name__)


@api.post("/shorten")
def shorten():
    data = request.get_json(silent=True) or {}
    url = data.get("url")
    if not url or not url.startswith(("http://", "https://")):
        return jsonify(error="a valid http(s) url is required"), 400

    store = current_app.config["STORE"]
    keygen = current_app.config["KEYGEN"]

    # retry on the rare key collision (random strategy)
    for _ in range(5):
        key = keygen.generate()
        if store.put_if_absent(key, url):
            short = f"{request.host_url}{key}"
            return jsonify(key=key, short_url=short, url=url), 201
    return jsonify(error="could not allocate a key"), 503


@api.get("/<key>")
def resolve(key: str):
    url = current_app.config["STORE"].get(key)
    if url is None:
        return jsonify(error="not found"), 404
    return redirect(url, code=302)

# Application factory — wires config, the store, the chosen strategy,
# and the observability endpoints. Config is injected, never hard-coded.
import os
from flask import Flask
from .api import api
from .store import MemoryStore
from .keygen import get_strategy
from .observability import register_observability


def create_app() -> Flask:
    app = Flask(__name__)
    app.config["STORE"] = MemoryStore()
    app.config["KEYGEN"] = get_strategy(os.environ.get("KEY_STRATEGY", "random"))
    app.register_blueprint(api)
    register_observability(app)
    return app

# ---------- Stage 1: builder ----------
FROM python:3.12-slim AS builder

ENV PYTHONDONTWRITEBYTECODE=1 \
    PIP_NO_CACHE_DIR=1

WORKDIR /build
COPY requirements.txt .
# build the dependency tree into an isolated venv we can copy wholesale
RUN python -m venv /opt/venv \
 && /opt/venv/bin/pip install --upgrade pip \
 && /opt/venv/bin/pip install -r requirements.txt

# ---------- Stage 2: runtime ----------
FROM python:3.12-slim AS runtime

# run as an unprivileged user (Session 11: never run containers as root)
RUN useradd --create-home --uid 10001 appuser
ENV PATH="/opt/venv/bin:$PATH" \
    PYTHONUNBUFFERED=1 \
    KEY_STRATEGY=random \
    PORT=8080

WORKDIR /app
COPY --from=builder /opt/venv /opt/venv
COPY app/ ./app/

USER appuser
EXPOSE 8080

# container-native health check — the orchestrator polls this
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD python -c "import urllib.request,os; \
      urllib.request.urlopen(f'http://127.0.0.1:{os.environ[\"PORT\"]}/healthz').read()" || exit 1

# Gunicorn: a production WSGI server, not Flask's dev server
CMD ["gunicorn", "--bind", "0.0.0.0:8080", "--workers", "2", \
     "--access-logfile", "-", "app:create_app()"]

name: ci

on:
  push:
    branches: [main]
  pull_request:

permissions:
  contents: read
  packages: write          # push the image to GHCR

env:
  IMAGE: ghcr.io/${{ github.repository }}

jobs:
  # ---------- test + lint run on every push and PR ----------
  test:
    runs-on: ubuntu-latest
    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
          pip install pytest coverage ruff
      - name: Lint
        run: ruff check app tests
      - name: Test with coverage gate
        run: |
          coverage run -m pytest -q
          coverage report --fail-under=80   # fail the build below 80%

  # ---------- build the image; on main, also push it ----------
  image:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Log in to GHCR
        if: github.ref == 'refs/heads/main'
        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@v6
        with:
          context: .
          push: ${{ github.ref == 'refs/heads/main' }}
          tags: |
            ${{ env.IMAGE }}:${{ github.sha }}
            ${{ env.IMAGE }}:latest

  # ---------- deploy only from main, after a green image ----------
  deploy:
    needs: image
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    environment: production      # gate: require an approval here
    steps:
      - uses: actions/checkout@v4
      - uses: hashicorp/setup-terraform@v3
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ secrets.AWS_DEPLOY_ROLE }}   # OIDC, no static keys
          aws-region: eu-west-1
      - name: Terraform apply
        working-directory: infra
        run: |
          terraform init
          terraform apply -auto-approve \
            -var="image=${{ env.IMAGE }}:${{ github.sha }}"

terraform {
  required_version = ">= 1.6"
  required_providers {
    aws = { source = "hashicorp/aws", version = "~> 5.0" }
  }
  # remote state so the team shares one source of truth
  backend "s3" {
    bucket = "urlshort-tfstate"
    key    = "prod/terraform.tfstate"
    region = "eu-west-1"
  }
}

provider "aws" {
  region = "eu-west-1"
}

variable "image" {
  description = "Container image tag to deploy (passed by CI)"
  type        = string
}

resource "aws_ecs_cluster" "main" {
  name = "urlshort"
}

resource "aws_ecs_task_definition" "app" {
  family                   = "urlshort"
  requires_compatibilities = ["FARGATE"]
  network_mode             = "awsvpc"
  cpu                      = 256
  memory                   = 512
  execution_role_arn       = aws_iam_role.exec.arn

  container_definitions = jsonencode([{
    name      = "urlshort"
    image     = var.image
    essential = true
    portMappings = [{ containerPort = 8080 }]
    environment  = [{ name = "KEY_STRATEGY", value = "random" }]
    healthCheck = {
      command     = ["CMD-SHELL", "python -c \"import urllib.request;urllib.request.urlopen('http://127.0.0.1:8080/healthz')\""]
      interval    = 30
      timeout     = 5
      retries     = 3
      startPeriod = 10
    }
    logConfiguration = {
      logDriver = "awslogs"
      options = {
        "awslogs-group"         = aws_cloudwatch_log_group.app.name
        "awslogs-region"        = "eu-west-1"
        "awslogs-stream-prefix" = "urlshort"
      }
    }
  }])
}

resource "aws_cloudwatch_log_group" "app" {
  name              = "/ecs/urlshort"
  retention_in_days = 30
}

resource "aws_ecs_service" "app" {
  name            = "urlshort"
  cluster         = aws_ecs_cluster.main.id
  task_definition = aws_ecs_task_definition.app.arn
  desired_count   = 2          # two tasks: rolling deploys with no downtime
  launch_type     = "FARGATE"

  network_configuration {
    subnets         = var.private_subnets
    security_groups = [aws_security_group.app.id]
  }

  load_balancer {
    target_group_arn = aws_lb_target_group.app.arn
    container_name   = "urlshort"
    container_port   = 8080
  }
}

resource "aws_lb_target_group" "app" {
  name        = "urlshort"
  port        = 8080
  protocol    = "HTTP"
  vpc_id      = var.vpc_id
  target_type = "ip"

  health_check {
    path                = "/healthz"   # the ALB polls the app's health endpoint
    healthy_threshold   = 2
    unhealthy_threshold = 3
    interval            = 15
  }
}

# Health, structured logging, and Prometheus metrics in one place.
import logging, sys, time, json
from flask import request, Response

_REQUESTS: dict[tuple[str, int], int] = {}
_LATENCY_SUM = 0.0
_LATENCY_COUNT = 0


def register_observability(app):
    _configure_json_logging()

    @app.get("/healthz")
    def healthz():
        # liveness: cheap, dependency-free, polled by Docker/ECS/ALB
        return {"status": "ok"}, 200

    @app.before_request
    def _start_timer():
        request._t0 = time.perf_counter()

    @app.after_request
    def _record(resp):
        global _LATENCY_SUM, _LATENCY_COUNT
        dt = time.perf_counter() - getattr(request, "_t0", time.perf_counter())
        key = (request.endpoint or "unknown", resp.status_code)
        _REQUESTS[key] = _REQUESTS.get(key, 0) + 1
        _LATENCY_SUM += dt
        _LATENCY_COUNT += 1
        app.logger.info(json.dumps({
            "msg": "request", "path": request.path, "method": request.method,
            "status": resp.status_code, "duration_ms": round(dt * 1000, 2),
        }))
        return resp

    @app.get("/metrics")
    def metrics():
        # Prometheus text exposition format — scraped on an interval
        lines = ["# TYPE http_requests_total counter"]
        for (endpoint, status), n in _REQUESTS.items():
            lines.append(f'http_requests_total{{endpoint="{endpoint}",status="{status}"}} {n}')
        avg = (_LATENCY_SUM / _LATENCY_COUNT) if _LATENCY_COUNT else 0.0
        lines += ["# TYPE http_request_duration_seconds_avg gauge",
                  f"http_request_duration_seconds_avg {avg:.6f}"]
        return Response("\n".join(lines) + "\n", mimetype="text/plain")


def _configure_json_logging():
    handler = logging.StreamHandler(sys.stdout)   # logs go to stdout; the platform collects them
    logging.getLogger().addHandler(handler)
    logging.getLogger().setLevel(logging.INFO)

# branch off main for one issue
git switch -c feat/shorten-endpoint

# conventional commits make the history (and changelog) readable
git commit -m "feat(api): add POST /shorten with strategy-based keys"
git commit -m "test(api): cover collision retry path"

# publish and open a PR — CI runs test + lint on the PR automatically
git push -u origin feat/shorten-endpoint
gh pr create --fill            # review + green CI required before merge

# after approval + squash-merge, main builds the image and (with approval) deploys