Proyecto práctico: API de gestión de tareas con FastAPI, SQLModel y JWT (Python)

22 ene 2026 688 vistas
python Proyecto práctico: API de gestión de tareas con FastAPI, SQLModel y JWT (Python)

Proyecto práctico: API de gestión de tareas con FastAPI, SQLModel y JWT

Construiremos una API REST para gestionar tareas (TODO) con autenticación por token JWT. El objetivo es un ejemplo real y completo, listo para pruebas locales y extensible a producción.

Requisitos previos

  • Python 3.10+
  • Conocimientos básicos de Python y HTTP/REST
  • pip o un entorno virtual (venv)

Dependencias

Usaremos:

  • FastAPI: framework web moderno
  • Uvicorn: servidor ASGI para desarrollo
  • SQLModel: ORM ligero (integra Pydantic + SQLAlchemy)
  • python-jose: generación/verificación de JWT
  • passlib[bcrypt]: hash de contraseñas

requirements.txt

fastapi>=0.95
uvicorn[standard]>=0.20
sqlmodel>=0.0.8
python-jose>=3.3
passlib[bcrypt]>=1.7

Estructura de carpetas

todo-api/
├── requirements.txt
└── app/
    ├── main.py          # entrada de la app y routers
    ├── config.py        # configuración (secret, expiración)
    ├── database.py      # engine y sesión
    ├── models.py        # modelos SQLModel (User, Task)
    ├── schemas.py       # esquemas Pydantic si se desea (a veces redundante con SQLModel)
    ├── auth.py          # hashing y utilidades JWT
    └── crud.py          # operaciones DB

Código completo de los archivos principales

app/config.py

from datetime import timedelta

# En producción mueve esto a variables de entorno
SECRET_KEY = "cambiame_por_una_clave_muy_larga_y_secreta"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60  # 1 hora

# Para uso en tests locales
DATABASE_URL = "sqlite:///./todo.db"

app/database.py

from sqlmodel import SQLModel, create_engine, Session
from .config import DATABASE_URL

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

def init_db():
    # Crea tablas si no existen
    SQLModel.metadata.create_all(engine)

# dependencia para endpoints
def get_session():
    with Session(engine) as session:
        yield session

app/models.py

from typing import Optional
from sqlmodel import SQLModel, Field, Relationship

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True, unique=True)
    hashed_password: str

    tasks: list["Task"] = Relationship(back_populates="owner")

class TaskBase(SQLModel):
    title: str
    description: Optional[str] = None
    completed: bool = False

class Task(TaskBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    owner_id: Optional[int] = Field(default=None, foreign_key="user.id")

    owner: Optional[User] = Relationship(back_populates="tasks")

class TaskCreate(TaskBase):
    pass

class TaskRead(TaskBase):
    id: int
    owner_id: int

class UserCreate(SQLModel):
    username: str
    password: str

class Token(SQLModel):
    access_token: str
    token_type: str = "bearer"

app/auth.py

from datetime import datetime, timedelta
from typing import Optional
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlmodel import Session, select

from .config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
from .models import User
from .database import get_session

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    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 get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="No se pudo validar las credenciales",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = session.exec(select(User).where(User.username == username)).first()
    if user is None:
        raise credentials_exception
    return user

app/crud.py

from sqlmodel import Session, select
from .models import User, Task, TaskCreate
from .auth import get_password_hash, verify_password

# Usuarios

def get_user_by_username(session: Session, username: str):
    return session.exec(select(User).where(User.username == username)).first()

def create_user(session: Session, username: str, password: str) -> User:
    hashed = get_password_hash(password)
    user = User(username=username, hashed_password=hashed)
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

# Autenticación

def authenticate_user(session: Session, username: str, password: str):
    user = get_user_by_username(session, username)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

# Tasks

def create_task(session: Session, owner_id: int, task_in: TaskCreate) -> Task:
    task = Task.from_orm(task_in)  # TaskCreate -> TaskBase data
    task.owner_id = owner_id
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

def get_tasks_for_user(session: Session, owner_id: int):
    return session.exec(select(Task).where(Task.owner_id == owner_id)).all()

def get_task_by_id(session: Session, task_id: int):
    return session.get(Task, task_id)

def update_task(session: Session, task: Task, **fields):
    for key, value in fields.items():
        setattr(task, key, value)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

def delete_task(session: Session, task: Task):
    session.delete(task)
    session.commit()

app/main.py

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlmodel import Session
from datetime import timedelta

from .database import init_db, get_session
from .models import UserCreate, Token, TaskCreate, TaskRead
from .crud import create_user, authenticate_user, create_task, get_tasks_for_user, get_task_by_id, update_task, delete_task
from .auth import create_access_token, get_current_user
from .config import ACCESS_TOKEN_EXPIRE_MINUTES

app = FastAPI(title="Todo API con FastAPI y JWT")

@app.on_event("startup")
def on_startup():
    init_db()

# Registro
@app.post("/register", response_model=Token)
def register(user_in: UserCreate, session: Session = Depends(get_session)):
    existing = session.exec(
        session.query(User).filter(User.username == user_in.username)
    ).first()
    # Pero como usamos SQLModel, mejor usar CRUD:
    from .crud import get_user_by_username
    if get_user_by_username(session, user_in.username):
        raise HTTPException(status_code=400, detail="Usuario ya existe")
    user = create_user(session, user_in.username, user_in.password)
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

# Token (login)
@app.post("/token", response_model=Token)
def login(form_data: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)):
    user = authenticate_user(session, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Usuario o contraseña incorrectos")
    access_token = create_access_token(data={"sub": user.username}, expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    return {"access_token": access_token, "token_type": "bearer"}

# CRUD de tasks (protegidos)
@app.post("/tasks", response_model=TaskRead)
def create_task_endpoint(task_in: TaskCreate, current_user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    task = create_task(session, current_user.id, task_in)
    return task

@app.get("/tasks", response_model=list[TaskRead])
def list_tasks(current_user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    return get_tasks_for_user(session, current_user.id)

@app.get("/tasks/{task_id}", response_model=TaskRead)
def get_task(task_id: int, current_user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    task = get_task_by_id(session, task_id)
    if not task or task.owner_id != current_user.id:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    return task

@app.put("/tasks/{task_id}", response_model=TaskRead)
def update_task_endpoint(task_id: int, task_in: TaskCreate, current_user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    task = get_task_by_id(session, task_id)
    if not task or task.owner_id != current_user.id:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    updated = update_task(session, task, title=task_in.title, description=task_in.description, completed=task_in.completed)
    return updated

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task_endpoint(task_id: int, current_user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    task = get_task_by_id(session, task_id)
    if not task or task.owner_id != current_user.id:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    delete_task(session, task)
    return None

Por qué esta arquitectura y decisiones

  • FastAPI: rendimiento, validación automática con Pydantic y documentación interactiva (OpenAPI).
  • SQLModel: combina modelos de datos (Pydantic) y ORM (SQLAlchemy) en una única capa, reduciendo duplicación.
  • SQLite: ideal para desarrollo local. Cambia a PostgreSQL en producción y configura correctamente el engine.
  • JWT (python-jose): tokens stateless, simples de usar para APIs. Controla caducidad y revocación si necesitas logout efectivo (ver notas).
  • passlib + bcrypt: hashing seguro de contraseñas.
  • Separación en módulos (auth, crud, models) facilita pruebas y mantenimiento.

Cómo ejecutar (local)

  1. Crear entorno virtual e instalar deps: 
    python -m venv .venv
source .venv/bin/activate  # o .venv\Scripts\activate en Windows
pip install -r requirements.txt
  2. Iniciar la app: 
    uvicorn app.main:app --reload
  3. La documentación interactiva estará en http://127.0.0.1:8000/docs

Pruebas rápidas (curl)

# Registrar
curl -X POST "http://127.0.0.1:8000/register" -H "Content-Type: application/json" -d '{"username":"alice","password":"secret"}'

# Login
curl -X POST "http://127.0.0.1:8000/token" -F "username=alice" -F "password=secret"

# Crear tarea (usa el token devuelto como Bearer)
curl -X POST "http://127.0.0.1:8000/tasks" -H "Authorization: Bearer " -H "Content-Type: application/json" -d '{"title":"Comprar leche","description":"2 litros","completed":false}'

# Listar tareas
curl -X GET "http://127.0.0.1:8000/tasks" -H "Authorization: Bearer "

Buenas prácticas y extensiones posibles

  • Migraciones: añade Alembic si vas a usar PostgreSQL en producción.
  • Rotación y revocación de tokens: implementa refresh tokens y una blacklist si necesites logout inmediato.
  • Rate limiting y protección contra fuerza bruta en endpoints de auth.
  • Validación adicional y tareas por prioridad/etiquetas si la app crece.

Consejo avanzado: nunca guardes la clave SECRET_KEY en el repositorio. En producción usa un gestor de secretos o variables de entorno, protege los endpoints de autenticación con límites de intentos y considera implementar tokens de refresco y revocación (lista negra) para poder invalidar sesiones en caso de compromiso.

Comentarios
¿Quieres comentar?

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

Iniciar Sesión

Comentarios (0)

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

Más en python
Todas las Categorías
java 18 php 18 python 10 rust 11 sql 8
Iniciar Sesión