Build a URL Shortener from Scratch — Part 3: API Endpoints with FastAPI

# app/schemas.py
from pydantic import BaseModel, HttpUrl, Field
from datetime import datetime

class URLCreate(BaseModel):
    url: HttpUrl  # Validates URL format automatically
    custom_code: str | None = Field(None, max_length=20, pattern=r'^[a-zA-Z0-9_-]+$')
    expires_in_days: int | None = Field(None, ge=1, le=365)

class URLResponse(BaseModel):
    short_code: str
    short_url: str
    original_url: str
    created_at: datetime
    expires_at: datetime | None

    class Config:
        from_attributes = True

class URLAnalytics(BaseModel):
    short_code: str
    original_url: str
    click_count: int
    created_at: datetime

    class Config:
        from_attributes = True

# app/services/shortener.py
from sqlalchemy.orm import Session
from app.models import URL
from app.utils.base62 import generate_short_code
from app.config import settings

class ShortenerService:
    def __init__(self, db: Session):
        self.db = db

    def create(self, original_url: str, custom_code: str | None = None,
               expires_in_days: int | None = None) -> URL:
        # Check custom code uniqueness
        if custom_code:
            existing = self.db.query(URL).filter(URL.short_code == custom_code).first()
            if existing:
                raise ValueError("Custom code already taken")

        # Create the URL record first to get an ID
        url = URL(original_url=original_url)
        self.db.add(url)
        self.db.flush()  # Get the ID without committing

        # Generate short code
        if custom_code:
            url.short_code = custom_code
        else:
            url.short_code = generate_short_code(url.id)

        # Set expiration
        if expires_in_days:
            from datetime import datetime, timedelta, timezone
            url.expires_at = datetime.now(timezone.utc) + timedelta(days=expires_in_days)

        self.db.commit()
        self.db.refresh(url)
        return url

    def get_by_code(self, code: str) -> URL | None:
        return self.db.query(URL).filter(URL.short_code == code).first()

# app/api/routes.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.schemas import URLCreate, URLResponse, URLAnalytics
from app.services.shortener import ShortenerService
from app.core.database import get_db
from app.config import settings

router = APIRouter()

@router.post("/shorten", response_model=URLResponse, status_code=201)
def shorten_url(body: URLCreate, db: Session = Depends(get_db)):
    service = ShortenerService(db)
    try:
        url = service.create(
            original_url=str(body.url),
            custom_code=body.custom_code,
            expires_in_days=body.expires_in_days
        )
        return {
            "short_code": url.short_code,
            "short_url": f"{settings.base_url}/{url.short_code}",
            "original_url": url.original_url,
            "created_at": url.created_at,
            "expires_at": url.expires_at
        }
    except ValueError as e:
        raise HTTPException(status_code=409, detail=str(e))

@router.get("/{code}")
def redirect_to_url(code: str, db: Session = Depends(get_db)):
    service = ShortenerService(db)
    url = service.get_by_code(code)

    if not url:
        raise HTTPException(status_code=404, detail="Short URL not found")

    # Check expiration
    from datetime import datetime, timezone
    if url.expires_at and url.expires_at < datetime.now(timezone.utc):
        raise HTTPException(status_code=410, detail="Short URL has expired")

    # Increment click count
    url.click_count += 1
    db.commit()

    from fastapi.responses import RedirectResponse
    return RedirectResponse(url=url.original_url, status_code=307)

@router.get("/stats/{code}", response_model=URLAnalytics)
def get_url_stats(code: str, db: Session = Depends(get_db)):
    service = ShortenerService(db)
    url = service.get_by_code(code)
    if not url:
        raise HTTPException(status_code=404, detail="Short URL not found")
    return url

# app/main.py
from fastapi import FastAPI
from app.api.routes import router
from app.core.database import engine
from app.models import Base

# Create tables (for development; use Alembic in production)
Base.metadata.create_all(bind=engine)

app = FastAPI(title="URL Shortener", version="1.0.0")

app.include_router(router)

# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "postgresql://postgres:password@localhost:5432/url_shortener"
    base_url: str = "http://localhost:8000"
    redis_url: str = "redis://localhost:6379"

    class Config:
        env_file = ".env"

settings = Settings()

# Start the server
uvicorn app.main:app --reload

# Create a short URL
curl -X POST http://localhost:8000/shorten \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com/very-long-url", "expires_in_days": 30}'

# Response:
# {
#   "short_code": "q0Z",
#   "short_url": "http://localhost:8000/q0Z",
#   "original_url": "https://example.com/very-long-url",
#   "created_at": "2026-03-01T10:00:00Z",
#   "expires_at": "2026-03-31T10:00:00Z"
# }

# Test redirect (follow the redirect with -L)
curl -L http://localhost:8000/q0Z
# Redirects to https://example.com/very-long-url

# Check stats
curl http://localhost:8000/stats/q0Z
# {"short_code":"q0Z","original_url":"...","click_count":1,...}