End-to-End: Build, Test, Containerize, Ship, Observe¶

from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    """All configuration comes from environment variables.
    Pydantic validates types and provides defaults.
    No configuration is hard-coded in source.
    """
    model_config = SettingsConfigDict(
        env_prefix="URLSHORT_",   # URLSHORT_REDIS_URL, URLSHORT_DEBUG, etc.
        env_file=".env",
        env_file_encoding="utf-8",
    )

    # Application
    app_name:    str  = "urlshort"
    environment: str  = "production"    # production | staging | development
    debug:       bool = False
    log_level:   str  = "INFO"
    code_length: int  = 7               # length of generated short codes

    # Redis
    redis_url:      str = "redis://localhost:6379/0"
    redis_max_conn: int = 20
    url_ttl_days:   int = 30            # 0 = never expire

    # Observability
    otlp_endpoint: str = ""            # empty = no OTLP export
    metrics_port:  int = 9090

    # Rate limiting
    rate_limit_per_minute: int = 60

    @property
    def url_ttl_seconds(self) -> int:
        return self.url_ttl_days * 86_400

@lru_cache
def get_settings() -> Settings:
    """Cached singleton so env is read once.
    Tests override by calling get_settings.cache_clear() before patching env.
    """
    return Settings()

import secrets, time
from contextlib import asynccontextmanager
from typing import AsyncGenerator

import structlog
from fastapi import Depends, FastAPI, HTTPException, Request, Response
from fastapi.responses import RedirectResponse
from pydantic import BaseModel, HttpUrl

from .config import Settings, get_settings
from .metrics import instrument_app, REQUEST_COUNT, REQUEST_LATENCY
from .storage import StorageBackend, get_storage

log = structlog.get_logger()

# ── Request / response models ─────────────────────────────────────────────
class ShortenRequest(BaseModel):
    url: HttpUrl
    custom_code: str | None = None

class ShortenResponse(BaseModel):
    code:      str
    short_url: str
    long_url:  str

# ── Application factory ───────────────────────────────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    log.info("startup", environment=get_settings().environment)
    yield
    log.info("shutdown")

def create_app(settings: Settings | None = None) -> FastAPI:
    cfg = settings or get_settings()
    app = FastAPI(
        title="URL Shortener",
        version="1.0.0",
        docs_url="/api/v1/docs",
        openapi_url="/api/v1/openapi.json",
        lifespan=lifespan,
    )
    instrument_app(app)
    app.include_router(router, prefix="/api/v1")
    return app

# ── Router ────────────────────────────────────────────────────────────────
from fastapi import APIRouter
router = APIRouter()

@router.post("/shorten", response_model=ShortenResponse, status_code=201)
async def shorten(
    body: ShortenRequest,
    request: Request,
    storage: StorageBackend = Depends(get_storage),
    cfg: Settings = Depends(get_settings),
) -> ShortenResponse:
    long_url = str(body.url)
    code = body.custom_code or secrets.token_urlsafe(cfg.code_length)[:cfg.code_length]

    existing = await storage.get(code)
    if existing and existing != long_url:
        raise HTTPException(status_code=409, detail="Code already taken")

    await storage.set(code, long_url, ttl=cfg.url_ttl_seconds or None)
    base = str(request.base_url).rstrip("/")
    log.info("shortened", code=code, long_url=long_url)
    return ShortenResponse(code=code, short_url=f"{base}/r/{code}", long_url=long_url)

@router.get("/r/{code}")
async def resolve(
    code: str,
    storage: StorageBackend = Depends(get_storage),
) -> RedirectResponse:
    long_url = await storage.get(code)
    if not long_url:
        raise HTTPException(status_code=404, detail="Code not found")
    await storage.increment_clicks(code)
    return RedirectResponse(url=long_url, status_code=302)

@router.get("/healthz")
async def healthz(storage: StorageBackend = Depends(get_storage)) -> dict:
    """Liveness + readiness probe. Returns 503 if Redis is unreachable."""
    healthy = await storage.ping()
    if not healthy:
        raise HTTPException(status_code=503, detail="Storage unavailable")
    return {"status": "ok", "storage": "redis"}

@router.get("/analytics/top")
async def top_urls(
    n: int = 10,
    storage: StorageBackend = Depends(get_storage),
) -> list[dict]:
    """Return the top-n most-clicked short codes."""
    return await storage.top_clicked(n)

app = create_app()

def main() -> None:
    import uvicorn
    uvicorn.run("urlshort.app:app", host="0.0.0.0", port=8000, reload=False)

from typing import Any
import redis.asyncio as aioredis
from .config import get_settings

class StorageBackend:
    """Thin async wrapper around Redis.

    Key schema:
      url:{code}       STRING  the long URL
      clicks:{code}    STRING  integer click counter (INCR)
      clicks:zset      ZSET    (code -> click count) for top-N queries
    """

    def __init__(self, client: aioredis.Redis) -> None:
        self._r = client

    async def ping(self) -> bool:
        try:
            return bool(await self._r.ping())
        except Exception:
            return False

    async def get(self, code: str) -> str | None:
        val: Any = await self._r.get(f"url:{code}")
        return val.decode() if val else None

    async def set(self, code: str, long_url: str, ttl: int | None = None) -> None:
        key = f"url:{code}"
        if ttl:
            await self._r.setex(key, ttl, long_url)
        else:
            await self._r.set(key, long_url)

    async def increment_clicks(self, code: str) -> None:
        pipe = self._r.pipeline()
        pipe.incr(f"clicks:{code}")
        pipe.zincrby("clicks:zset", 1, code)
        await pipe.execute()

    async def get_clicks(self, code: str) -> int:
        val: Any = await self._r.get(f"clicks:{code}")
        return int(val) if val else 0

    async def top_clicked(self, n: int = 10) -> list[dict]:
        """Returns top-n codes by click count using the sorted set."""
        results: list[tuple[bytes, float]] = await self._r.zrevrange(
            "clicks:zset", 0, n - 1, withscores=True
        )
        return [
            {"code": code.decode(), "clicks": int(score)}
            for code, score in results
        ]

# ── Dependency for FastAPI ────────────────────────────────────────────────
_pool: aioredis.Redis | None = None

async def get_storage() -> StorageBackend:
    global _pool
    if _pool is None:
        cfg = get_settings()
        _pool = aioredis.from_url(
            cfg.redis_url,
            max_connections=cfg.redis_max_conn,
            decode_responses=False,
        )
    return StorageBackend(_pool)

import time
from prometheus_client import Counter, Histogram, make_asgi_app
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

REQUEST_COUNT = Counter(
    "urlshort_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status"],
)

REQUEST_LATENCY = Histogram(
    "urlshort_request_duration_seconds",
    "HTTP request latency",
    ["method", "endpoint"],
    buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5),
)

REDIRECTS_TOTAL = Counter(
    "urlshort_redirects_total",
    "Total successful URL redirects",
    ["code"],
)

class PrometheusMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: Any) -> Response:
        start = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start

        # Normalise path to avoid cardinality explosion on /r/{code}
        path = request.url.path
        if path.startswith("/r/"):
            path = "/r/{code}"

        REQUEST_COUNT.labels(
            method=request.method, endpoint=path, status=response.status_code
        ).inc()
        REQUEST_LATENCY.labels(
            method=request.method, endpoint=path
        ).observe(duration)

        return response

def instrument_app(app: FastAPI) -> None:
    app.add_middleware(PrometheusMiddleware)
    # Expose /metrics for Prometheus scraping
    metrics_app = make_asgi_app()
    app.mount("/metrics", metrics_app)

import pytest
import fakeredis.aioredis as fakeredis
from httpx import ASGITransport, AsyncClient
from urlshort.app import create_app
from urlshort.config import Settings
from urlshort.storage import StorageBackend

@pytest.fixture
def test_settings() -> Settings:
    return Settings(
        environment="testing",
        redis_url="redis://localhost:6379/15",  # overridden by fake
        url_ttl_days=0,
        debug=True,
    )

@pytest.fixture
async def fake_redis() -> fakeredis.FakeRedis:
    r = fakeredis.FakeRedis()
    yield r
    await r.flushall()
    await r.aclose()

@pytest.fixture
async def client(test_settings: Settings, fake_redis: fakeredis.FakeRedis) -> AsyncClient:
    app = create_app(settings=test_settings)
    storage = StorageBackend(fake_redis)
    app.dependency_overrides[get_storage] = lambda: storage
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        yield ac

import pytest
from httpx import AsyncClient

pytestmark = pytest.mark.asyncio

async def test_shorten_then_resolve(client: AsyncClient) -> None:
    r = await client.post("/api/v1/shorten", json={"url": "https://example.com/long-path"})
    assert r.status_code == 201
    body = r.json()
    assert "code" in body
    assert body["long_url"] == "https://example.com/long-path"

    # Resolve should redirect
    r2 = await client.get(f"/api/v1/r/{body['code']}", follow_redirects=False)
    assert r2.status_code == 302
    assert r2.headers["location"] == "https://example.com/long-path"

async def test_unknown_code_is_404(client: AsyncClient) -> None:
    r = await client.get("/api/v1/r/notexist")
    assert r.status_code == 404

async def test_custom_code(client: AsyncClient) -> None:
    r = await client.post("/api/v1/shorten", json={
        "url": "https://example.com", "custom_code": "mycode"
    })
    assert r.status_code == 201
    assert r.json()["code"] == "mycode"

async def test_duplicate_custom_code_conflict(client: AsyncClient) -> None:
    await client.post("/api/v1/shorten", json={
        "url": "https://first.com", "custom_code": "clash"
    })
    r = await client.post("/api/v1/shorten", json={
        "url": "https://second.com", "custom_code": "clash"
    })
    assert r.status_code == 409

async def test_healthz(client: AsyncClient) -> None:
    r = await client.get("/api/v1/healthz")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"

async def test_click_counter_and_analytics(client: AsyncClient) -> None:
    r = await client.post("/api/v1/shorten", json={"url": "https://popular.com"})
    code = r.json()["code"]

    # Click three times
    for _ in range(3):
        await client.get(f"/api/v1/r/{code}", follow_redirects=False)

    top = await client.get("/api/v1/analytics/top?n=1")
    assert top.status_code == 200
    data = top.json()
    assert data[0]["code"] == code
    assert data[0]["clicks"] == 3