API REST con FastAPI, JWT y PostgreSQL (con Docker) — tutorial práctico

python API REST con FastAPI, JWT y PostgreSQL (con Docker) — tutorial práctico

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

  1. POST /register con JSON {"email":"t@ejemplo.com","password":"secret"} para crear usuario.
  2. POST /token usando form data username=t@ejemplo.com y password=secret — obtendrás access_token.
  3. 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.

Comentarios
¿Quieres comentar?

Inicia sesión con Telegram para participar en la conversación


Comentarios (0)

Aún no hay comentarios. ¡Sé el primero en comentar!

Iniciar Sesión