FastAPI URL shortener with metrics

# Create short URL
curl -X POST https://shorty.example.com/api/v1/links \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com/very/long/path", "alias": "fav"}'
# → {"id": "abc123", "alias": "fav", "url": "https://example.com/...", "short_url": "https://shorty.example.com/fav"}

# Visit redirect
curl -i https://shorty.example.com/fav
# → 302 Location: https://example.com/very/long/path

# Stats
curl https://shorty.example.com/api/v1/links/fav/stats
# → {"clicks": 42, "last_clicked": "...", "referrers": {"google.com": 18, ...}}

src/shorty/
├── __init__.py
├── main.py                  # FastAPI app
├── config.py                # pydantic-settings (env vars)
├── deps.py                  # Depends(): db_session, redis, current_user
├── db/
│   ├── __init__.py
│   ├── engine.py            # async engine + session_maker
│   └── models.py            # User, Link, ClickEvent
├── repos/
│   ├── link_repo.py
│   └── click_repo.py
├── services/
│   ├── shortener.py         # alias generation, collision retry
│   ├── click_tracker.py     # async background tracking
│   └── rate_limit.py        # Redis sliding-window
├── api/
│   ├── auth.py              # JWT issue/verify
│   ├── links.py             # POST /links, GET /links/{alias}
│   └── stats.py             # GET /links/{alias}/stats
├── middleware/
│   ├── request_id.py
│   ├── logging.py
│   └── metrics.py
└── observability/
    ├── logging.py           # structlog config
    └── metrics.py           # prometheus collectors
alembic/
├── env.py
└── versions/
tests/
├── conftest.py              # fixtures: app, db, redis (testcontainers)
├── test_create_link.py
├── test_redirect.py
├── test_stats.py
├── test_rate_limit.py
└── test_load.py             # locust / k6 sample

class User(Base):
    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    hashed_password: Mapped[str]
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

class Link(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    alias: Mapped[str] = mapped_column(String(32), unique=True, index=True)
    url: Mapped[str] = mapped_column(Text)
    user_id: Mapped[UUID] = mapped_column(ForeignKey("user.id"))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    expires_at: Mapped[datetime | None]

class ClickEvent(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    link_id: Mapped[int] = mapped_column(ForeignKey("link.id"), index=True)
    ts: Mapped[datetime] = mapped_column(server_default=func.now(), index=True)
    referrer: Mapped[str | None]
    user_agent: Mapped[str | None]
    country: Mapped[str | None]

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine, AsyncSession

engine = create_async_engine(settings.database_url, pool_size=20, max_overflow=0)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session

@app.get("/{alias}", include_in_schema=False)
async def redirect(
    alias: str,
    request: Request,
    bg: BackgroundTasks,
    redis: Redis = Depends(get_redis),
    db: AsyncSession = Depends(get_db),
) -> RedirectResponse:
    url = await redis.get(f"alias:{alias}")
    if url is None:
        link = await link_repo.get_by_alias(db, alias)
        if not link or (link.expires_at and link.expires_at < datetime.now()):
            raise HTTPException(404, "not found")
        url = link.url
        await redis.set(f"alias:{alias}", url, ex=3600)
    bg.add_task(enqueue_click, alias, request)
    return RedirectResponse(url, status_code=302)

async def rate_limit(redis: Redis, key: str, *, limit: int, window: int) -> bool:
    now = int(time.time() * 1000)
    cutoff = now - window * 1000
    async with redis.pipeline() as pipe:
        pipe.zremrangebyscore(key, 0, cutoff)
        pipe.zcard(key)
        pipe.zadd(key, {str(now): now})
        pipe.expire(key, window)
        _, count, *_ = await pipe.execute()
    return count < limit

@pytest.fixture(scope="session")
async def app():
    with PostgresContainer("postgres:16") as pg, \
         RedisContainer("redis:7") as redis_c:
        settings.database_url = pg.get_connection_url(driver="asyncpg")
        settings.redis_url = redis_c.get_connection_url()
        await run_migrations()
        yield create_app()

# fly.toml
app = "shorty"
[build]
  dockerfile = "Dockerfile"
[env]
  PORT = "8000"
[http_service]
  internal_port = 8000
  force_https = true
[[vm]]
  cpu_kind = "shared"
  cpus = 2
  memory_mb = 1024

fly launch
fly postgres create
fly redis create
fly deploy