API de tareas con FastAPI, JWT y SQLite (proyecto práctico en Python)

27 ene 2026 90 vistas
python API de tareas con FastAPI, JWT y SQLite (proyecto práctico en Python)

API de tareas con FastAPI, JWT y SQLite (proyecto práctico)

En este tutorial vas a construir una API REST en Python para gestionar tareas (CRUD) con autenticación por JWT. Usaremos FastAPI por su productividad y rendimiento, SQLModel/SQLite para persistencia y passlib + python-jose para autenticación segura. Te doy la estructura, el código completo de los archivos principales y explicaciones del porqué de cada decisión.

Requisitos previos

  • Python 3.10+
  • pip
  • Conocimientos básicos de HTTP y Python

Instala dependencias (mejor en un virtualenv):

pip install fastapi uvicorn sqlmodel passlib[bcrypt] python-jose[cryptography]

Estructura de carpetas

project/
├─ app/
│  ├─ __init__.py
│  ├─ main.py
│  ├─ database.py
│  ├─ models.py
│  ├─ schemas.py
│  └─ auth.py
├─ requirements.txt

Nos centramos en los archivos dentro de app/. A continuación verás el código completo de cada uno.

app/database.py

from sqlmodel import SQLModel, create_engine, Session
from typing import Generator
import os

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
engine = create_engine(DATABASE_URL, echo=False)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session() -> Generator[Session, None, None]:
    with Session(engine) as session:
        yield session

Por qué: SQLModel combina Pydantic y SQLAlchemy, lo que hace los modelos más simples y tipados. SQLite es ideal para desarrollo y demos.

app/models.py

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

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 Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    description: Optional[str] = None
    completed: bool = Field(default=False)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    owner_id: Optional[int] = Field(default=None, foreign_key="user.id")
    owner: Optional[User] = Relationship(back_populates="tasks")

Por qué: Separamos User y Task claramente. Relationship permite consultas que traigan dueño y tareas relacionadas si las necesitas.

app/schemas.py

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

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

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

class TaskCreate(BaseModel):
    title: str
    description: Optional[str] = None

class TaskRead(BaseModel):
    id: int
    title: str
    description: Optional[str]
    completed: bool
    created_at: datetime
    owner_id: Optional[int]

class TaskUpdate(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

Por qué: Separar esquemas (Pydantic) de modelos de DB da control sobre lo que expones en la API. Evitas filtrar campos sensibles como hashed_password.

app/auth.py

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

from .models import User
from .database import get_session

SECRET_KEY = os.getenv("SECRET_KEY", "CHANGE_THIS_SECRET")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

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

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 authenticate_user(session: Session, username: str, password: str):
    statement = select(User).where(User.username == username)
    user = session.exec(statement).first()
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async 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="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
    statement = select(User).where(User.id == int(user_id))
    user = session.exec(statement).first()
    if user is None:
        raise credentials_exception
    return user

Por qué: Uso OAuth2PasswordBearer para integrarme con la documentación automática de Swagger UI. El token contiene el claim "sub" con el user id para buscar el usuario en cada petición autenticada.

app/main.py

from fastapi import FastAPI, Depends, HTTPException, status
from sqlmodel import Session, select
from typing import List
from fastapi.security import OAuth2PasswordRequestForm

from .database import create_db_and_tables, get_session
from .models import User, Task
from .schemas import UserCreate, Token, TaskCreate, TaskRead, TaskUpdate
from .auth import (
    get_password_hash,
    authenticate_user,
    create_access_token,
    get_current_user,
    ACCESS_TOKEN_EXPIRE_MINUTES,
)
from datetime import timedelta

app = FastAPI()

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

@app.post("/register", status_code=201)
def register(user_in: UserCreate, session: Session = Depends(get_session)):
    statement = select(User).where(User.username == user_in.username)
    existing = session.exec(statement).first()
    if existing:
        raise HTTPException(status_code=400, detail="Username already registered")
    user = User(username=user_in.username, hashed_password=get_password_hash(user_in.password))
    session.add(user)
    session.commit()
    session.refresh(user)
    return {"id": user.id, "username": user.username}

@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=401, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": str(user.id)}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.post("/tasks/", response_model=TaskRead)
def create_task(task_in: TaskCreate, current_user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    task = Task(title=task_in.title, description=task_in.description, owner_id=current_user.id)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

@app.get("/tasks/", response_model=List[TaskRead])
def list_tasks(current_user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    statement = select(Task).where(Task.owner_id == current_user.id)
    tasks = session.exec(statement).all()
    return tasks

@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)):
    statement = select(Task).where(Task.id == task_id, Task.owner_id == current_user.id)
    task = session.exec(statement).first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@app.patch("/tasks/{task_id}", response_model=TaskRead)
def update_task(task_id: int, task_in: TaskUpdate, current_user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    statement = select(Task).where(Task.id == task_id, Task.owner_id == current_user.id)
    task = session.exec(statement).first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    task_data = task_in.dict(exclude_unset=True)
    for key, value in task_data.items():
        setattr(task, key, value)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, current_user: User = Depends(get_current_user), session: Session = Depends(get_session)):
    statement = select(Task).where(Task.id == task_id, Task.owner_id == current_user.id)
    task = session.exec(statement).first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    session.delete(task)
    session.commit()
    return None

Por qué: Endpoints simples, con dependencias claras. El login usa OAuth2PasswordRequestForm que funciona con la UI de Swagger. Cada endpoint protegido usa get_current_user para verificar token y obtener el usuario actual.

requirements.txt

fastapi
uvicorn
sqlmodel
passlib[bcrypt]
python-jose[cryptography]

Ponerlo en marcha

  1. Exporta variables de entorno (importante): 
    export SECRET_KEY="tu_super_secreto_aqui"
export DATABASE_URL="sqlite:///./app.db"
  2. Inicia la app: 
    uvicorn app.main:app --reload --port 8000
  3. Abre http://127.0.0.1:8000/docs para probar la API con Swagger UI.

Pruebas rápidas (ejemplo con curl)

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

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

# Crear tarea (sustituye TOKEN)
curl -X POST "http://127.0.0.1:8000/tasks/" -H "Authorization: Bearer TOKEN" -H "Content-Type: application/json" -d '{"title":"Comprar leche"}'

Decisiones técnicas y por qué

  • FastAPI: documentación automática, validación basada en tipos y alto rendimiento.
  • SQLModel: modelos tipados que evitan duplicidad entre Pydantic y ORM; reduce boilerplate.
  • SQLite: sencillo para demos; puedes cambiar DATABASE_URL a Postgres en producción.
  • JWT (python-jose): tokens stateless, fáciles de escalar en arquitecturas sin sesión.
  • passlib bcrypt: hashing seguro y probado para contraseñas.

Mejoras y siguientes pasos

  • Migraciones: añade Alembic si usas Postgres y esquemas más complejos.
  • Refresh tokens: implementa refresco seguro si necesitas sesiones más largas.
  • Rate limiting, logging y observabilidad para producción.
  • Tests unitarios e integración (pytest + httpx + test database).

Consejo avanzado: no almacenes SECRET_KEY en el código. Usa un gestor de secretos (HashiCorp Vault, AWS Secrets Manager) o variables de entorno inyectadas por tu orquestador. Además, valida siempre los scopes/claims de tus JWT si planeas soportar roles o permisos. Si vas a exponer esta API públicamente, obliga HTTPS y considera rotación de claves y revocación de tokens (lista negra) para mayor seguridad.

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