Proyecto práctico: API RESTful con FastAPI, autenticación JWT y PostgreSQL

Construiremos una API minimal pero realista con registro, login (JWT) y un endpoint protegido. Cubriremos la estructura de proyecto, código completo de los archivos principales y por qu se toman ciertas decisiones. El proyecto puede desplegarse con Docker y PostgreSQL.

Requisitos previos

Python 3.9+ o Docker y Docker Compose
Conocimientos básicos de Python y REST APIs
Opcional: editor y Postman o curl para probar endpoints

Estructura de carpetas

project-root/
  docker-compose.yml
  Dockerfile
  requirements.txt
  .env.example
  app/
    __init__.py
    main.py
    models.py
    schemas.py
    database.py
    crud.py
    auth.py
    routes.py

Archivos principales (cdigo completo)

.env.example

POSTGRES_USER=postgres
POSTGRES_PASSWORD=postgres
POSTGRES_DB=fastapi_db
POSTGRES_HOST=db
DATABASE_URL=postgresql+psycopg2://postgres:postgres@db:5432/fastapi_db
SECRET_KEY=changemeplease_set_a_strong_random_value
ACCESS_TOKEN_EXPIRE_MINUTES=30

requirements.txt

fastapi
uvicorn[standard]
sqlalchemy
psycopg2-binary
passlib[bcrypt]
python-jose[cryptography]
python-dotenv

docker-compose.yml

version: '3.8'

services:
  db:
    image: postgres:14
    environment:
      - POSTGRES_USER=${POSTGRES_USER}
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=${POSTGRES_DB}
    volumes:
      - db_data:/var/lib/postgresql/data
    ports:
      - '5432:5432'

  web:
    build: .
    env_file: .env.example
    depends_on:
      - db
    ports:
      - '8000:8000'
    volumes:
      - ./:/app

volumes:
  db_data:

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' ]

app/database.py

import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from dotenv import load_dotenv

load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), '..', '.env.example'))

DATABASE_URL = os.getenv('DATABASE_URL')

engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# dependency for routes
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app/models.py

from sqlalchemy import Column, Integer, String, Boolean
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)
    is_active = Column(Boolean, default=True)

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
    is_active: bool

    class Config:
        orm_mode = True

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    id: Optional[int] = None

app/crud.py

from sqlalchemy.orm import Session
from . import models
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')

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, email: str, password: str):
    hashed_password = pwd_context.hash(password)
    db_user = models.User(email=email, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def authenticate_user(db: Session, email: str, password: str):
    user = get_user_by_email(db, email)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

app/auth.py

import os
from datetime import datetime, timedelta
from typing import Optional
from jose import jwt, JWTError
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from . import crud, schemas, database
from dotenv import load_dotenv

load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), '..', '.env.example'))

SECRET_KEY = os.getenv('SECRET_KEY') or 'changeme'
ALGORITHM = 'HS256'
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv('ACCESS_TOKEN_EXPIRE_MINUTES') or 30)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

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 get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(database.get_db)) -> schemas.UserOut:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail='Could not validate credentials',
        headers={'WWW-Authenticate': 'Bearer'}
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get('sub')
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = db.query(crud.get_user_by_email.__qualname__.split('.')[0]).first() if False else db.query
    # we avoid confusing reflection: fetch by id explicitly
    from .models import User as UserModel
    user = db.query(UserModel).filter(UserModel.id == user_id).first()
    if user is None:
        raise credentials_exception
    return user

app/routes.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from fastapi.security import OAuth2PasswordRequestForm
from datetime import timedelta

from . import schemas, crud, database, auth

router = APIRouter()

@router.post('/register', response_model=schemas.UserOut)
def register(user_in: schemas.UserCreate, db: Session = Depends(database.get_db)):
    existing = crud.get_user_by_email(db, user_in.email)
    if existing:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Email already registered')
    user = crud.create_user(db, user_in.email, user_in.password)
    return user

@router.post('/token', response_model=schemas.Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(database.get_db)):
    user = crud.authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Incorrect username or password')
    access_token_expires = timedelta(minutes=auth.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = auth.create_access_token(data={'sub': str(user.id)}, expires_delta=access_token_expires)
    return {'access_token': access_token, 'token_type': 'bearer'}

@router.get('/users/me', response_model=schemas.UserOut)
def read_users_me(current_user=Depends(auth.get_current_user)):
    return current_user

app/main.py

from fastapi import FastAPI
from . import models, database
from .routes import router

app = FastAPI()
app.include_router(router)

@app.on_event('startup')
def on_startup():
    # crea tablas si no existen; en produccin usa migrations
    database.Base.metadata.create_all(bind=database.engine)

@app.get('/')
def root():
    return {'message': 'API ejecutando'}

Por qu se disea1a aseed

FastAPI ofrece velocidad de desarrollo y validacif3n autome1tica con pydantic.
SQLAlchemy es un ORM maduro y ampliamente usado; usamos el patrof3n de session por request para simplicidad.
passlib con bcrypt para hashing seguro de contrasef1as.
JWT con python-jose para tokens compactos y sin estado; mantuvimos el subject ('sub') con el id del usuario.
Docker Compose orquesta PostgreSQL y la aplicacif3n, facilitando reproducibilidad.

Comandos para ejecutar

Con Docker: ejecutar en la carpeta del proyecto: docker-compose up --build
Sin Docker: crear un virtualenv, instalar requirements.txt y ejecutar: uvicorn app.main:app --reload

Endpoints importantes:

POST /register con body json { 'email': 'tu@correo', 'password': 'clave' }
POST /token con form data username=tu@correo y password=clave para obtener access_token
GET /users/me con header Authorization: Bearer <token> para usuario autenticado

Errores comunes y por qu solucionarlos ased

Conexion a DB fallida: revisar DATABASE_URL y esperar al contenedor db en docker-compose.
Passwords en claro: nunca guardes contrasef1as en texto; usar passlib.
Tokens no verificables: cambiar SECRET_KEY por un valor fuerte en produccif3n y rotarlo si es necesario.

Siguientes pasos recomendados

Agregar migraciones con Alembic en vez de create_all para tener control de cambios en esquema.
Implementar refresco de tokens (refresh tokens) y revocacif3n si necesitas logout real.
Limitar intentos de login y agregar rate limiting para proteger contra fuerza bruta.

Consejo avanzado: en produccif3n almacena la SECRET_KEY en un gestor de secretos, usa HTTPS obligatorio, y considera usar short-lived access tokens combinados con refresh tokens guardados en httpOnly cookies para mitigar XSS. Una prf3xima mejora es migrar a SQLAlchemy async o usar un pool asedncrono si tu carga lo justifica.

Proyecto: API RESTful con FastAPI, autenticación JWT y PostgreSQL (Docker)