Crear un acortador de URL con FastAPI, SQLModel y base62
Proyecto práctico: construimos un servicio mínimo y productivo que convierte URLs largas en códigos cortos, redirige y registra clicks. Usaremos FastAPI para la API, SQLModel (SQLAlchemy + Pydantic) para el modelo y una codificación base62 sobre el id autoincremental para generar short codes.
Requisitos previos
- Python 3.9+
- pip
- Conocimientos básicos de HTTP y Python
Instalación rápida de dependencias:
pip install fastapi uvicorn sqlmodel jinja2
Estructura de carpetas
url-shortener/
├─ main.py
├─ database.py
├─ models.py
├─ utils.py
├─ templates/
│ └─ index.html
└─ requirements.txt
Decisiones de diseño (el por qué)
- FastAPI: rápido, asincrónico y excelente para APIs JSON + templates.
- SQLModel: evita escribir esquemas Pydantic y tablas por separado; integra bien con SQLite para este proyecto.
- Base62 sobre el id auto incremental: genera códigos cortos únicos y ordenables. Evitamos colisiones y dependencias externas.
- Creación en dos pasos (insertar fila -> calcular código -> actualizar): simple y segura con transacción, no hay necesidad de buscar un código libre.
Codigo completo: archivos principales
database.py
from sqlmodel import create_engine, SQLModel
sqlite_file_name = 'database.db'
sqlite_url = f'sqlite:///{sqlite_file_name}'
# check_same_thread False para permitir sesiones en threads diferentes (uvicorn)
engine = create_engine(sqlite_url, echo=False, connect_args={'check_same_thread': False})
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
models.py
from sqlmodel import SQLModel, Field
from typing import Optional
from datetime import datetime
class URL(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
original_url: str
short_code: Optional[str] = None
created_at: datetime = Field(default_factory=datetime.utcnow)
clicks: int = 0
utils.py
from urllib.parse import urlparse
# base62: 0-9, a-z, A-Z
_ALPHABET = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
_BASE = len(_ALPHABET)
def base62_encode(num: int) -> str:
if num == 0:
return _ALPHABET[0]
s = ''
while num > 0:
num, rem = divmod(num, _BASE)
s = _ALPHABET[rem] + s
return s
def validate_url(u: str) -> bool:
try:
p = urlparse(u)
return p.scheme in ('http', 'https') and bool(p.netloc)
except Exception:
return False
main.py
from fastapi import FastAPI, Request, Form, HTTPException
from fastapi.responses import RedirectResponse, HTMLResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session, select
from database import engine, create_db_and_tables
from models import URL
from utils import base62_encode, validate_url
app = FastAPI()
templates = Jinja2Templates(directory='templates')
# crea la base de datos al arrancar (para este ejemplo simple)
create_db_and_tables()
@app.get('/', response_class=HTMLResponse)
def index(request: Request):
return templates.TemplateResponse('index.html', {'request': request})
@app.post('/shorten', response_class=HTMLResponse)
def shorten_form(request: Request, original_url: str = Form(...)):
if not validate_url(original_url):
return templates.TemplateResponse('index.html', {'request': request, 'error': 'URL inválida'})
with Session(engine) as session:
# 1) Insertar fila sin short_code
url = URL(original_url=original_url)
session.add(url)
session.commit()
session.refresh(url) # ahora url.id tiene valor
# 2) Generar short_code y actualizar
url.short_code = base62_encode(url.id)
session.add(url)
session.commit()
short = request.url_for('redirect_short', code=url.short_code)
return templates.TemplateResponse('index.html', {'request': request, 'short_url': str(short)})
@app.post('/api/shorten')
def api_shorten(payload: dict):
original = payload.get('url') if isinstance(payload, dict) else None
if not original or not validate_url(original):
raise HTTPException(status_code=400, detail='URL inválida')
with Session(engine) as session:
url = URL(original_url=original)
session.add(url)
session.commit()
session.refresh(url)
url.short_code = base62_encode(url.id)
session.add(url)
session.commit()
return {'short_code': url.short_code, 'short_url': f'/{url.short_code}'}
@app.get('/{code}')
def redirect_short(code: str):
with Session(engine) as session:
statement = select(URL).where(URL.short_code == code)
result = session.exec(statement).first()
if not result:
raise HTTPException(status_code=404, detail='No encontrado')
result.clicks += 1
session.add(result)
session.commit()
return RedirectResponse(url=result.original_url)
@app.get('/admin/{code}')
def stats(code: str):
with Session(engine) as session:
result = session.exec(select(URL).where(URL.short_code == code)).first()
if not result:
raise HTTPException(status_code=404, detail='No encontrado')
return {'original_url': result.original_url, 'clicks': result.clicks, 'created_at': result.created_at.isoformat()}
templates/index.html
<!doctype html>
<html lang='es'>
<head>
<meta charset='utf-8'>
<meta name='viewport' content='width=device-width,initial-scale=1'>
<title>Acortador de URL</title>
</head>
<body>
<h1>Acortador de URL</h1>
<form method='post' action='/shorten'>
<input name='original_url' placeholder='https://ejemplo.com/una/url/larga' style='width:400px' required/>
<button type='submit'>Acortar</button>
</form>
{% if error %}
<p style='color:red'>{{ error }}</p>
{% endif %}
{% if short_url %}
<p>URL acortada: <a href='{{ short_url }}'>{{ short_url }}</a></p>
{% endif %}
<hr/>
<p>También puedes usar la API: POST /api/shorten con JSON { 'url': 'https://...' }</p>
</body>
</html>
Ejecutar la aplicación
uvicorn main:app --reload --port 8000
Visita http://127.0.0.1:8000 para probar el formulario. Para la API: curl -X POST -H 'Content-Type: application/json' -d '{"url":"https://example.com/long/path"}' http://127.0.0.1:8000/api/shorten
Notas técnicas y consideraciones
- Race conditions: el patrón insertar-commit-refresh-actualizar evita colisiones porque usamos el id autoincremental. En DBs con sharding u otros esquemas necesitarías otro enfoque (hashids, UUID+base62, o un generador centralizado).
- Validación de URLs: aquí hacemos una comprobación básica del scheme y netloc. Puedes agregar listas de bloqueo, verificación de existencia, o escaneo de contenido malicioso.
- Escalado: para alto tráfico, cambia SQLite por Postgres y añade caching (Redis) para las redirecciones, y un CDN si expones muchas redirecciones estáticas.
- Analítica y privacidad: este ejemplo incrementa un contador simple. Para analíticas avanzadas guarda user-agent, IP (con cuidado legal) y timestamps en otra tabla.
- SEO y seguridad: evita crear redirecciones hacia esquemas peligrosos (file:, javascript:) y considera añadir headers de seguridad.
Siguiente paso recomendado: protege el endpoint de creación con autenticación y añade un sistema de rate limiting. Si vas a exponer la API públicamente, instrumenta límites por IP/API key en combinación con un cache de respuestas para reducir latencia.
¿Quieres comentar?
Inicia sesión con Telegram para participar en la conversación