IF0100 - Programación OO II

Objetivos de Aprendizaje

• Conectar FastAPI con base de datos SQLite
• Crear modelos con SQLAlchemy ORM
• Implementar operaciones CRUD
• Usar dependencias para sesiones de BD

SQLAlchemy (10 min)

SQLAlchemy es un ORM (Object Relational Mapper) para Python.

Instalación

pip install sqlalchemy

Conexión a SQLite

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Crear motor de base de datos
SQLALCHEMY_DATABASE_URL = "sqlite:///./taskflow.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, 
    connect_args={"check_same_thread": False}
)

# Crear sesión
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base para modelos
Base = declarative_base()

# Función para obtener sesión
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Modelos ORM (20 min)

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from database import Base

class TareaDB(Base):
    __tablename__ = "tareas"
    
    id = Column(Integer, primary_key=True, index=True)
    titulo = Column(String, nullable=False)
    descripcion = Column(String)
    completada = Column(Boolean, default=False)
    prioridad = Column(Integer, default=1)
    fecha_creacion = Column(DateTime, server_default=func.now())

class UsuarioDB(Base):
    __tablename__ = "usuarios"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, nullable=False)
    email = Column(String, unique=True, nullable=False)
    password_hash = Column(String, nullable=False)
    activo = Column(Boolean, default=True)

# Crear tablas
Base.metadata.create_all(bind=engine)

Pydantic Models para API

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

class TareaBase(BaseModel):
    titulo: str
    descripcion: Optional[str] = None
    prioridad: int = 1

class TareaCreate(TareaBase):
    pass

class Tarea(TareaBase):
    id: int
    completada: bool
    fecha_creacion: datetime
    
    class Config:
        from_attributes = True  # Permite convertir desde ORM

Operaciones CRUD (25 min)

from sqlalchemy.orm import Session
from models import TareaDB

# CREATE
def create_tarea(db: Session, tarea: TareaCreate):
    db_tarea = TareaDB(**tarea.model_dump())
    db.add(db_tarea)
    db.commit()
    db.refresh(db_tarea)
    return db_tarea

# READ
def get_tarea(db: Session, tarea_id: int):
    return db.query(TareaDB).filter(TareaDB.id == tarea_id).first()

def get_tareas(db: Session, skip: int = 0, limit: int = 100):
    return db.query(TareaDB).offset(skip).limit(limit).all()

# UPDATE
def update_tarea(db: Session, tarea_id: int, tarea_update: TareaCreate):
    db_tarea = db.query(TareaDB).filter(TareaDB.id == tarea_id).first()
    if db_tarea:
        for key, value in tarea_update.model_dump().items():
            setattr(db_tarea, key, value)
        db.commit()
        db.refresh(db_tarea)
    return db_tarea

# DELETE
def delete_tarea(db: Session, tarea_id: int):
    db_tarea = db.query(TareaDB).filter(TareaDB.id == tarea_id).first()
    if db_tarea:
        db.delete(db_tarea)
        db.commit()
        return True
    return False

FastAPI + Base de Datos (15 min)

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db, engine
from models import Base, TareaDB
from schemas import Tarea, TareaCreate
from crud import create_tarea, get_tarea, get_tareas, update_tarea, delete_tarea

# Crear tablas
Base.metadata.create_all(bind=engine)

app = FastAPI(title="TaskFlow con BD")

@app.post("/tareas", response_model=Tarea)
def crear_tarea(tarea: TareaCreate, db: Session = Depends(get_db)):
    return create_tarea(db=db, tarea=tarea)

@app.get("/tareas", response_model=list[Tarea])
def listar_tareas(
    skip: int = 0, 
    limit: int = 100, 
    db: Session = Depends(get_db)
):
    return get_tareas(db, skip=skip, limit=limit)

@app.get("/tareas/{tarea_id}", response_model=Tarea)
def obtener_tarea(tarea_id: int, db: Session = Depends(get_db)):
    db_tarea = get_tarea(db, tarea_id=tarea_id)
    if db_tarea is None:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    return db_tarea

@app.put("/tareas/{tarea_id}", response_model=Tarea)
def actualizar_tarea(
    tarea_id: int, 
    tarea: TareaCreate, 
    db: Session = Depends(get_db)
):
    db_tarea = update_tarea(db, tarea_id=tarea_id, tarea_update=tarea)
    if db_tarea is None:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    return db_tarea

@app.delete("/tareas/{tarea_id}")
def eliminar_tarea(tarea_id: int, db: Session = Depends(get_db)):
    success = delete_tarea(db, tarea_id=tarea_id)
    if not success:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    return {"message": "Tarea eliminada"}

Estructura de Archivos

taskflow/
├── main.py           # FastAPI app
├── database.py       # Conexión SQLAlchemy
├── models.py         # Modelos ORM
├── schemas.py        # Pydantic models
├── crud.py          # Operaciones CRUD
└── taskflow.db      # Base de datos SQLite

Ejercicio: Usuarios con BD (10 min)

Implementa CRUD completo para el modelo Usuario con base de datos.

Requisitos

• Tabla usuarios con: id, username, email, password_hash, activo
• Validar username único antes de crear
• Validar email único antes de crear
• No retornar password_hash en las respuestas

# schemas.py
class UsuarioBase(BaseModel):
    username: str
    email: str

class UsuarioCreate(UsuarioBase):
    password: str

class Usuario(UsuarioBase):
    id: int
    activo: bool
    
    class Config:
        from_attributes = True

# crud.py
def create_usuario(db: Session, usuario: UsuarioCreate):
    # Verificar si username existe
    if db.query(UsuarioDB).filter(UsuarioDB.username == usuario.username).first():
        raise HTTPException(400, "Username ya existe")
    
    # Crear usuario (hash password en producción)
    db_usuario = UsuarioDB(
        username=usuario.username,
        email=usuario.email,
        password_hash=f"hash_{usuario.password}"  # Simplificado
    )
    db.add(db_usuario)
    db.commit()
    db.refresh(db_usuario)
    return db_usuario

Resumen Unidad 3

FastAPI + Pydantic + SQLAlchemy

• FastAPI: Framework para APIs rápidas
• Pydantic: Validación de datos
• Depends: Inyección de dependencias
• TestClient: Testing de APIs
• SQLAlchemy: ORM para base de datos