API REST con FastAPI, JWT y PostgreSQL (con Docker)
Proyecto práctico: crea una API mínima pero funcional que ofrece registro de usuarios, login con JWT y un endpoint protegido. Incluye Docker Compose para levantar PostgreSQL y el servicio FastAPI.
Requisitos previos
- Conocimientos básicos de Python.
- Docker y Docker Compose instalados (opcional: puedes correr local con pip).
- Python 3.9+ (si no usas Docker).
Estructura de carpetas
fastapi-jwt-postgres/
├─ app/
│ ├─ __init__.py
│ ├─ main.py
│ ├─ database.py
│ ├─ models.py
│ ├─ schemas.py
│ ├─ crud.py
│ └─ auth.py
├─ Dockerfile
├─ docker-compose.yml
└─ requirements.txt
Dependencias (requirements.txt)
fastapi
uvicorn[standard]
sqlalchemy
psycopg2-binary
passlib[bcrypt]
python-jose[cryptography]
python-dotenv
Archivos principales (código completo)
app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
import os
DATABASE_URL = os.getenv('DATABASE_URL', 'postgresql://postgres:password@db:5432/postgres')
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
app/models.py
from sqlalchemy import Column, Integer, String
from .database import Base
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
app/schemas.py
from pydantic import BaseModel, EmailStr
from typing import Optional
class UserCreate(BaseModel):
email: EmailStr
password: str
class UserOut(BaseModel):
id: int
email: EmailStr
class Config:
orm_mode = True
class Token(BaseModel):
access_token: str
token_type: str
app/crud.py
from sqlalchemy.orm import Session
from . import models, schemas
def get_user_by_email(db: Session, email: str):
return db.query(models.User).filter(models.User.email == email).first()
def create_user(db: Session, user: schemas.UserCreate, hashed_password: str):
db_user = models.User(email=user.email, hashed_password=hashed_password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
app/auth.py
from datetime import datetime, timedelta
from passlib.context import CryptContext
from jose import jwt, JWTError
from typing import Optional
import os
pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
SECRET_KEY = os.getenv('SECRET_KEY', 'changeme')
ALGORITHM = os.getenv('ALGORITHM', 'HS256')
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv('ACCESS_TOKEN_EXPIRE_MINUTES', '30'))
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({'exp': expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def decode_access_token(token: str):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
app/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
from sqlalchemy.orm import Session
from . import models, schemas, crud, auth
from .database import SessionLocal, engine, Base
from typing import Generator
import os
# Crear tablas (sólo demo; en producción usa migraciones con Alembic)
Base.metadata.create_all(bind=engine)
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
# Dependency
def get_db() -> Generator:
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post('/register', response_model=schemas.UserOut)
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.get_user_by_email(db, user.email)
if db_user:
raise HTTPException(status_code=400, detail='Email already registered')
hashed_password = auth.get_password_hash(user.password)
new_user = crud.create_user(db, user, hashed_password)
return new_user
@app.post('/token', response_model=schemas.Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
user = crud.get_user_by_email(db, form_data.username)
if not user or not auth.verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail='Incorrect username or password',
headers={'WWW-Authenticate': 'Bearer'},
)
access_token = auth.create_access_token(data={'sub': user.email})
return {'access_token': access_token, 'token_type': 'bearer'}
@app.get('/users/me', response_model=schemas.UserOut)
def read_users_me(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
payload = auth.decode_access_token(token)
if payload is None:
raise HTTPException(status_code=401, detail='Invalid authentication credentials')
email = payload.get('sub')
if email is None:
raise HTTPException(status_code=401, detail='Invalid token payload')
user = crud.get_user_by_email(db, email)
if user is None:
raise HTTPException(status_code=404, detail='User not found')
return user
Dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY ./app ./app
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
docker-compose.yml
version: '3.8'
services:
db:
image: postgres:13
environment:
POSTGRES_PASSWORD: password
volumes:
- pgdata:/var/lib/postgresql/data
ports:
- '5432:5432'
api:
build: .
depends_on:
- db
environment:
DATABASE_URL: postgresql://postgres:password@db:5432/postgres
SECRET_KEY: supersecretkey
ACCESS_TOKEN_EXPIRE_MINUTES: '30'
ports:
- '8000:8000'
volumes:
pgdata:
Por qué estas decisiones
- FastAPI: validación automática con Pydantic, rendimiento y documentación automática (OpenAPI).
- Pydantic (schemas): separación clara entre modelos de persistencia (SQLAlchemy) y modelos de entrada/salida.
- SQLAlchemy: ORM robusto y ampliamente usado; suficiente para una API CRUD.
- JWT: autenticación sin estado (stateless), útil para microservicios y clientes REST/SPA.
- passlib + bcrypt: hashing seguro para contraseñas.
- Docker Compose: reproduce fácilmente un entorno con PostgreSQL para desarrollo.
Cómo ejecutar
Con Docker Compose:
docker-compose up --build
La API quedará accesible en http://localhost:8000 y la documentación interactiva en http://localhost:8000/docs
Si prefieres ejecutar sin Docker:
python -m venv .venv
source .venv/bin/activate # o .venv\Scripts\activate en Windows
pip install -r requirements.txt
# exporta DATABASE_URL apuntando a una postgres existente
uvicorn app.main:app --reload
Pruebas rápidas
- POST /register con JSON {"email":"t@ejemplo.com","password":"secret"} para crear usuario.
- POST /token usando form data username=t@ejemplo.com y password=secret — obtendrás access_token.
- GET /users/me con Authorization: Bearer <token> para acceder al endpoint protegido.
Notas de seguridad y buenas prácticas
- No guardes SECRET_KEY en el código: usa un gestor de secretos o variables de entorno seguras.
- Activa HTTPS en producción para proteger tokens en tránsito.
- Usa expiración corta para access tokens y considera implementar refresh tokens y revocación.
- En producción, gestiona migraciones con Alembic en lugar de Base.metadata.create_all.
Siguientes pasos recomendados
Integra refresh tokens con almacenamiento seguro (Redis), añade roles/permissions, añade rate limiting y registra intentos de login fallidos. Si vas a exponer la API públicamente, configura un WAF y revisa CORS/CSRF según el tipo de cliente.
Consejo avanzado: para mayor seguridad, implementa token rotation con refresh tokens almacenados server-side (a prueba de replays) y usa un short-lived access token; además, firma y valida tokens con claves asimétricas (RS256) si gestionas roscas entre servicios.
¿Quieres comentar?
Inicia sesión con Telegram para participar en la conversación