Build a URL Shortener from Scratch — Part 5: Analytics, Docker, and Deployment

# app/services/analytics.py
import requests
from app.models import Click
from app.core.cache import redis_client

def record_click(url_id: int, request_headers: dict, client_ip: str):
    """Record a click with referrer and country data."""
    country = get_country_from_ip(client_ip)
    referrer = request_headers.get("referer", "")

    # Write to Redis for async processing
    redis_client.lpush("click_queue", f"{url_id}|{referrer}|{country}|{client_ip}")

def get_country_from_ip(ip: str) -> str:
    """Lookup country from IP using a free geo IP service."""
    try:
        resp = requests.get(f"http://ip-api.com/json/{ip}?fields=countryCode", timeout=2)
        return resp.json().get("countryCode", "XX")
    except Exception:
        return "XX"

def process_click_queue(db_session):
    """Batch process clicks from Redis to PostgreSQL."""
    while True:
        raw = redis_client.rpop("click_queue")
        if not raw:
            break
        url_id, referrer, country, _ = raw.split("|")
        click = Click(url_id=int(url_id), referrer=referrer, country=country)
        db_session.add(click)
    db_session.commit()

@router.get("/stats/{code}")
def get_url_stats(code: str, db: Session = Depends(get_db)):
    url = db.query(URL).filter(URL.short_code == code).first()
    if not url:
        raise HTTPException(404)

    # Get daily clicks for last 30 days
    from sqlalchemy import func
    thirty_days_ago = datetime.now(timezone.utc) - timedelta(days=30)

    daily = (
        db.query(
            func.date(Click.clicked_at).label("date"),
            func.count().label("clicks")
        )
        .filter(Click.url_id == url.id, Click.clicked_at >= thirty_days_ago)
        .group_by(func.date(Click.clicked_at))
        .order_by("date")
        .all()
    )

    top_referrers = (
        db.query(Click.referrer, func.count().label("count"))
        .filter(Click.url_id == url.id)
        .group_by(Click.referrer)
        .order_by(func.count().desc())
        .limit(5)
        .all()
    )

    return {
        "short_code": code,
        "original_url": url.original_url,
        "total_clicks": url.click_count,
        "daily_clicks": [{"date": str(d), "clicks": c} for d, c in daily],
        "top_referrers": [{"referrer": r, "count": c} for r, c in top_referrers]
    }

FROM python:3.12-slim

WORKDIR /app

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

COPY . .

EXPOSE 8000

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

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql://postgres:password@db:5432/url_shortener
      REDIS_URL: redis://redis:6379
      BASE_URL: https://short.ly
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      retries: 3

  db:
    image: postgres:16
    volumes:
      - pgdata:/var/lib/postgresql/data
    environment:
      POSTGRES_PASSWORD: password
      POSTGRES_DB: url_shortener
    healthcheck:
      test: ["CMD-SHELL", "pg_isready"]
      interval: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

volumes:
  pgdata:
  redis_data:

# Build and start
docker-compose up -d --build

# Run migrations
docker-compose exec api alembic upgrade head

# Verify
curl -X POST https://short.ly/shorten \
  -d '{"url": "https://example.com"}'
# {"short_code":"q0Z","short_url":"https://short.ly/q0Z",...}