API Engineering

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field

router = APIRouter(prefix="/api/v1")

class QuoteResponse(BaseModel):
    symbol: str = Field(..., example="CSCO")
    price: float = Field(..., ge=0, example=62.45)
    currency: str = Field(default="USD")
    timestamp: str

@router.get("/quote/{symbol}", response_model=QuoteResponse)
async def get_quote(symbol: str, user=Depends(get_current_user)):
    """Fetch the latest stock quote for a given ticker symbol."""
    symbol = symbol.upper().strip()
    if not symbol.isalpha() or len(symbol) > 5:
        raise HTTPException(status_code=400, detail="Invalid ticker symbol")
    quote = await fetch_quote(symbol)
    if not quote:
        raise HTTPException(status_code=404, detail="Symbol not found")
    return quote

@app.route("/api/v1/shorten", methods=["POST"])
@require_auth
@rate_limit(max_requests=30, window=60)
def create_short_link():
    """Create a new shortened URL. Requires JWT in Authorization header."""
    data = request.get_json(silent=True)
    if not data or not data.get("url"):
        return jsonify({"error": "Missing 'url' field"}), 400

    url = data["url"].strip()
    if not url.startswith(("http://", "https://")):
        return jsonify({"error": "URL must use http or https"}), 400

    short = generate_short_code()
    db.execute("INSERT INTO links (code, url, user_id) VALUES (?, ?, ?)",
               (short, url, g.user_id))
    return jsonify({"short_url": f"https://jamieblair.co.uk/s/{short}"}), 201

import os, jwt
from datetime import datetime, timedelta, timezone

# Secret loaded from environment : never hardcoded
JWT_SECRET = os.environ["JWT_SECRET"]
JWT_ALGORITHM = "HS256"
ACCESS_TTL = timedelta(minutes=15)
REFRESH_TTL = timedelta(days=7)

def create_access_token(user_id: int) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "iat": now,
        "exp": now + ACCESS_TTL,
        "type": "access",
    }
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Invalid ticker symbol",
    "field": "symbol",
    "request_id": "req_8f3a2b1c"
  }
}

@app.errorhandler(Exception)
def handle_exception(exc):
    """Catch-all: log the real error, return a safe response."""
    request_id = getattr(g, "request_id", "unknown")

    if isinstance(exc, HTTPException):
        return jsonify({
            "error": {
                "code": exc.description,
                "message": str(exc),
                "request_id": request_id,
            }
        }), exc.code

    # Log the full traceback server-side
    app.logger.error("Unhandled: %s [%s]", exc, request_id, exc_info=True)

    # Return a generic message : never expose internal details
    return jsonify({
        "error": {
            "code": "INTERNAL_ERROR",
            "message": "Something went wrong",
            "request_id": request_id,
        }
    }), 500

openapi: "3.0.3"
info:
  title: "Market Snapshot API"
  version: "1.0.0"
  description: "Live stock quotes and daily snapshots"
paths:
  /api/v1/quote/{symbol}:
    get:
      summary: "Get live quote"
      parameters:
        - name: symbol
          in: path
          required: true
          schema: { type: string, pattern: "^[A-Z]{1,5}$" }
      responses:
        "200":
          description: "Success"
          content:
            application/json:
              schema: { $ref: "#/components/schemas/QuoteResponse" }
        "400":
          description: "Invalid symbol"
        "404":
          description: "Symbol not found"

def test_create_short_link_requires_auth(client):
    """POST /api/v1/shorten without a token returns 401."""
    resp = client.post("/api/v1/shorten", json={"url": "https://example.com"})
    assert resp.status_code == 401
    assert resp.json["error"]["code"] == "AUTH_REQUIRED"


def test_create_short_link_valid(client, auth_header):
    """POST /api/v1/shorten with valid token creates a link."""
    resp = client.post(
        "/api/v1/shorten",
        json={"url": "https://example.com/article"},
        headers=auth_header,
    )
    assert resp.status_code == 201
    assert "short_url" in resp.json


def test_reject_invalid_url(client, auth_header):
    """POST /api/v1/shorten with non-http URL returns 400."""
    resp = client.post(
        "/api/v1/shorten",
        json={"url": "ftp://not-allowed.com"},
        headers=auth_header,
    )
    assert resp.status_code == 400