API de notas con FastAPI, SQLAlchemy y autenticación JWT (tutorial práctico)

26 feb 2026 25 vistas
python API de notas con FastAPI, SQLAlchemy y autenticación JWT (tutorial práctico)

API de notas con FastAPI, SQLAlchemy y autenticación JWT

Proyecto práctico: construye una API REST para gestionar notas con registro de usuarios, login con JWT y control de acceso por propietario. Te doy la estructura, código completo de los archivos principales y explico por qué se hace cada cosa.

Requisitos previos

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

Estructura de carpetas

project-root/
├─ .env               # variables (SECRET_KEY, DATABASE_URL opcional)
├─ requirements.txt
├─ Dockerfile
└─ app/
   ├─ main.py
   ├─ database.py
   ├─ models.py
   ├─ schemas.py
   ├─ crud.py
   ├─ auth.py
   └─ deps.py

requirements.txt

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

.env (ejemplo)

# Cambia SECRET_KEY por una larga y aleatoria
SECRET_KEY=your-very-secret-key
ACCESS_TOKEN_EXPIRE_MINUTES=60
DATABASE_URL=sqlite:///./notes.db

Archivo: app/database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
from dotenv import load_dotenv

load_dotenv()

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./notes.db")

# Para SQLite usamos check_same_thread=False
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

Archivo: app/models.py

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy.orm import relationship
from .database import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)

    notes = relationship("Note", back_populates="owner")

class Note(Base):
    __tablename__ = "notes"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    content = Column(Text)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="notes")

Archivo: app/schemas.py

from pydantic import BaseModel
from typing import Optional

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

class UserOut(BaseModel):
    id: int
    username: str

    class Config:
        orm_mode = True

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

# Note
class NoteBase(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None

class NoteCreate(NoteBase):
    title: str
    content: str

class NoteOut(NoteBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

Archivo: app/crud.py

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

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

# Users
def get_user_by_username(db: Session, username: str):
    return db.query(models.User).filter(models.User.username == username).first()

def create_user(db: Session, user: schemas.UserCreate):
    hashed = pwd_context.hash(user.password)
    db_user = models.User(username=user.username, hashed_password=hashed)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

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

# Notes
def create_note(db: Session, note: schemas.NoteCreate, owner_id: int):
    db_note = models.Note(title=note.title, content=note.content, owner_id=owner_id)
    db.add(db_note)
    db.commit()
    db.refresh(db_note)
    return db_note

def get_notes_for_user(db: Session, user_id: int):
    return db.query(models.Note).filter(models.Note.owner_id == user_id).all()

def get_note_by_id(db: Session, note_id: int):
    return db.query(models.Note).filter(models.Note.id == note_id).first()

def update_note(db: Session, db_note: models.Note, note_in: schemas.NoteCreate):
    db_note.title = note_in.title
    db_note.content = note_in.content
    db.commit()
    db.refresh(db_note)
    return db_note

def delete_note(db: Session, db_note: models.Note):
    db.delete(db_note)
    db.commit()

Archivo: app/auth.py

from datetime import datetime, timedelta
from jose import JWTError, jwt
from typing import Optional
import os
from dotenv import load_dotenv

load_dotenv()

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

def create_access_token(data: dict, expires_delta: Optional[timedelta] = 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

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            return None
        return int(user_id)
    except JWTError:
        return None

Archivo: app/deps.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from .database import SessionLocal
from . import crud
from .auth import verify_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

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

# Current user dependency
def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    user_id = verify_token(token)
    if user_id is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials")
    user = db.query(crud.models.User).filter(crud.models.User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
    return user

Archivo: app/main.py

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

from . import models, schemas, crud, auth
from .database import engine, Base
from .deps import get_db, get_current_user

# Crea tablas automáticamente (para desarrollo)
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Notes API")

@app.post("/register", response_model=schemas.UserOut)
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
    existing = crud.get_user_by_username(db, user.username)
    if existing:
        raise HTTPException(status_code=400, detail="Username already registered")
    created = crud.create_user(db, user)
    return created

@app.post("/token", response_model=schemas.Token)
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = crud.get_user_by_username(db, form_data.username)
    if not user or not crud.verify_password(form_data.password, user.hashed_password):
        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"}

@app.post("/notes", response_model=schemas.NoteOut)
def create_note(note: schemas.NoteCreate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
    return crud.create_note(db, note, owner_id=current_user.id)

@app.get("/notes", response_model=list[schemas.NoteOut])
def list_notes(current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
    return crud.get_notes_for_user(db, current_user.id)

@app.get("/notes/{note_id}", response_model=schemas.NoteOut)
def get_note(note_id: int, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
    n = crud.get_note_by_id(db, note_id)
    if not n or n.owner_id != current_user.id:
        raise HTTPException(status_code=404, detail="Note not found")
    return n

@app.put("/notes/{note_id}", response_model=schemas.NoteOut)
def update_note(note_id: int, note_in: schemas.NoteCreate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
    n = crud.get_note_by_id(db, note_id)
    if not n or n.owner_id != current_user.id:
        raise HTTPException(status_code=404, detail="Note not found")
    return crud.update_note(db, n, note_in)

@app.delete("/notes/{note_id}", status_code=204)
def delete_note(note_id: int, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
    n = crud.get_note_by_id(db, note_id)
    if not n or n.owner_id != current_user.id:
        raise HTTPException(status_code=404, detail="Note not found")
    crud.delete_note(db, n)
    return None

Archivo: Dockerfile

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV PYTHONUNBUFFERED=1
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

Por qué esta organización y decisiones

  • Separación en módulos (models, schemas, crud, auth, deps) mejora la mantenibilidad y pruebas.
  • SQLAlchemy con SessionLocal es una forma sencilla y robusta para apps pequeñas/medianas. Usamos SQLite por simplicidad; en producción cambia DATABASE_URL a PostgreSQL y ajusta el engine.
  • JWT (python-jose) emite tokens sin estado; guardamos sólo el id del usuario en el sub. Es sencillo y escalable, pero no permite revocación fácil (ver notas de seguridad más abajo).
  • Passlib con bcrypt para hashear contraseñas: evita almacenar contraseñas en texto claro.
  • OAuth2PasswordBearer facilita integración con la seguridad de FastAPI y la generación del token via /token.

Cómo ejecutar (local)

  1. Crear y activar un virtualenv: python -m venv venv && source venv/bin/activate
  2. Instalar dependencias: pip install -r requirements.txt
  3. Configurar .env con SECRET_KEY (importante)
  4. Iniciar: uvicorn app.main:app --reload
  5. Abrir documentación automática: http://127.0.0.1:8000/docs

Pruebas rápidas (ejemplos)

  1. POST /register -> {"username":"alice","password":"secret"}
  2. POST /token (form-data) username=alice, password=secret -> recibe access_token
  3. Usa Authorization: Bearer <token> para llamar a /notes

Limitaciones y siguientes pasos sugeridos

  • No hay gestión de roles ni verificación por email. Añade confirmación de cuenta para producción.
  • JWT no tiene mecanismo de revocación; si necesitas revocar tokens impleméntalo con una blacklist o reduce el tiempo de expiración y usa refresh tokens.
  • Agrega migraciones con Alembic si vas a cambiar modelos en producción.
  • Escribe tests (pytest + TestClient) para endpoints y lógica CRUD.

Consejo avanzado: en producción protege SECRET_KEY en un gestor de secretos (Vault, AWS Secrets Manager) y habilita HTTPS con certificados. También limita el scope del token (claims personalizados) y añade logging/monitoring para intentos de autenticación sospechosos.

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 20 php 22 python 14 rust 14 sql 16
Iniciar Sesión