Build 5 — FastAPI Service (Production-Ready)

# Build 5 — FastAPI Service ## The brief Build a small but production-ready FastAPI service. Goals: clean structure, typed throughout, structured logging, configurable, dockerized, tested. ## Service idea Choose one: - **URL shortener API**: POST /shorten {url} → returns short code; GET /:code → 301 redirect. - **Notes API**: CRUD for personal notes; per-user. - **Pollster**: create polls, vote, see results. We'll use URL shortener for the walkthrough. ## Step 1: Setup (10 min) ```bash uv init url-shortener cd url-shortener uv add fastapi uvicorn pydantic-settings sqlalchemy psycopg2-binary python-json-logger uv add --dev pytest pytest-asyncio httpx pyright ``` ``` url-shortener/ ├── pyproject.toml ├── Dockerfile ├── .env.example ├── src/ │ └── url_shortener/ │ ├── __init__.py │ ├── main.py # FastAPI app │ ├── config.py # settings │ ├── models.py # SQLAlchemy + Pydantic │ ├── db.py # database setup │ ├── routes.py # endpoints │ └── logging_config.py └── tests/ ├── conftest.py └── test_api.py ``` ## Step 2: Config (10 min) ```python # src/url_shortener/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str = "sqlite:///./test.db" log_level: str = "INFO" base_url: str = "http://localhost:8000" short_code_length: int = 7 class Config: env_file = ".env" settings = Settings() ``` ``` # .env.example (commit this) DATABASE_URL=postgresql://localhost/url_shortener LOG_LEVEL=INFO BASE_URL=https://short.example.com ``` ## Step 3: Models (15 min) ```python # src/url_shortener/models.py from datetime import datetime from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from pydantic import BaseModel, HttpUrl class Base(DeclarativeBase): pass class URLEntity(Base): __tablename__ = "urls" short_code: Mapped[str] = mapped_column(primary_key=True, index=True) long_url: Mapped[str] created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow) click_count: Mapped[int] = mapped_column(default=0) class URLCreate(BaseModel): url: HttpUrl class URLResponse(BaseModel): short_code: str short_url: str long_url: str click_count: int created_at: datetime ``` ## Step 4: Database (10 min) ```python # src/url_shortener/db.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from .config import settings from .models import Base engine = create_engine(settings.database_url, echo=False) SessionLocal = sessionmaker(bind=engine) def init_db(): Base.metadata.create_all(engine) def get_db(): """FastAPI dependency: yield a DB session.""" db = SessionLocal() try: yield db finally: db.close() ``` ## Step 5: Routes (25 min) ```python # src/url_shortener/routes.py import logging import secrets import string from typing import Annotated from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import RedirectResponse from sqlalchemy.orm import Session from .config import settings from .db import get_db from .models import URLEntity, URLCreate, URLResponse logger = logging.getLogger(__name__) router = APIRouter() def generate_short_code(length: int) -> str: return ''.join(secrets.choice(string.ascii_letters + string.digits) for _ in range(length)) @router.post("/shorten", response_model=URLResponse) def create_short_url( payload: URLCreate, db: Annotated[Session, Depends(get_db)], ) -> URLResponse: # Try to generate a unique short code for _ in range(10): code = generate_short_code(settings.short_code_length) if not db.query(URLEntity).filter_by(short_code=code).first(): break else: logger.error("could not generate unique short code") raise HTTPException(500, "could not generate short code") entity = URLEntity(short_code=code, long_url=str(payload.url)) db.add(entity) db.commit() logger.info("created_short_url", extra={"short_code": code, "long_url": str(payload.url)}) return URLResponse( short_code=code, short_url=f"{settings.base_url}/{code}", long_url=str(payload.url), click_count=0, created_at=entity.created_at, ) @router.get("/{short_code}") def redirect( short_code: str, db: Annotated[Session, Depends(get_db)], ) -> RedirectResponse: entity = db.query(URLEntity).filter_by(short_code=short_code).first() if not entity: logger.warning("short_code_not_found", extra={"short_code": short_code}) raise HTTPException(404, "short code not found") entity.click_count += 1 db.commit() return RedirectResponse(entity.long_url, status_code=301) @router.get("/api/info/{short_code}", response_model=URLResponse) def get_info( short_code: str, db: Annotated[Session, Depends(get_db)], ) -> URLResponse: entity = d