IF0100 - Programacion OO II

Objetivos de Aprendizaje

Al finalizar esta clase, seras capaz de:

• Usar Session para gestionar transacciones
• Implementar Create - crear registros
• Implementar Read - consultar registros
• Implementar Update - actualizar registros
• Implementar Delete - eliminar registros
• Crear un Repositorio generico

Session (10 min)

La Session gestiona las operaciones de base de datos en una transaccion.

from sqlalchemy.orm import Session
from database import SessionLocal

# Crear sesion
db: Session = SessionLocal()

try:
    # Operaciones de base de datos aqui
    db.commit()  # Guardar cambios
except Exception as e:
    db.rollback()  # Revertir en caso de error
    raise e
finally:
    db.close()  # Siempre cerrar sesion

CREATE - Crear Registros (10 min)

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

pwd_context = CryptContext(schemes=["bcrypt"])

def crear_usuario(db: Session, username: str, email: str, password: str) -> Usuario:
    # Crear instancia
    usuario = Usuario(
        username=username,
        email=email,
        password_hash=pwd_context.hash(password)
    )
    
    # Agregar a la sesion
    db.add(usuario)
    
    # Commit para guardar
    db.commit()
    
    # Refresh para obtener ID generado
    db.refresh(usuario)
    
    return usuario

# Uso
with SessionLocal() as db:
    nuevo = crear_usuario(db, "ana", "ana@test.com", "secret123")
    print(f"Usuario creado con ID: {nuevo.id}")

READ - Consultar Registros (15 min)

from typing import Optional, List
from sqlalchemy import select

def obtener_usuario_por_id(db: Session, id: int) -> Optional[Usuario]:
    """Obtener un usuario por ID."""
    return db.query(Usuario).filter(Usuario.id == id).first()

def obtener_usuario_por_username(db: Session, username: str) -> Optional[Usuario]:
    """Obtener un usuario por username."""
    return db.query(Usuario).filter(Usuario.username == username).first()

def obtener_todos_usuarios(db: Session, skip: int = 0, limit: int = 100) -> List[Usuario]:
    """Obtener todos los usuarios con paginacion."""
    return db.query(Usuario).offset(skip).limit(limit).all()

def buscar_usuarios_activos(db: Session) -> List[Usuario]:
    """Obtener usuarios activos."""
    return db.query(Usuario).filter(Usuario.activo == True).all()

def contar_usuarios(db: Session) -> int:
    """Contar usuarios."""
    return db.query(Usuario).count()

# Con SELECT moderno (SQLAlchemy 2.0)
def obtener_usuario_select(db: Session, id: int) -> Optional[Usuario]:
    stmt = select(Usuario).where(Usuario.id == id)
    return db.execute(stmt).scalar_one_or_none()

UPDATE - Actualizar Registros (10 min)

def actualizar_usuario(
    db: Session, 
    id: int, 
    username: str = None, 
    email: str = None
) -> Optional[Usuario]:
    """Actualizar un usuario."""
    usuario = db.query(Usuario).filter(Usuario.id == id).first()
    
    if not usuario:
        return None
    
    # Actualizar campos
    if username:
        usuario.username = username
    if email:
        usuario.email = email
    
    # Commit para guardar
    db.commit()
    db.refresh(usuario)
    
    return usuario

def activar_usuario(db: Session, id: int) -> bool:
    """Activar un usuario."""
    usuario = db.query(Usuario).filter(Usuario.id == id).first()
    if usuario:
        usuario.activo = True
        db.commit()
        return True
    return False

DELETE - Eliminar Registros (10 min)

def eliminar_usuario(db: Session, id: int) -> bool:
    """Eliminar un usuario."""
    usuario = db.query(Usuario).filter(Usuario.id == id).first()
    
    if not usuario:
        return False
    
    db.delete(usuario)
    db.commit()
    
    return True

# Eliminar con filtro directo
def eliminar_usuario_por_username(db: Session, username: str) -> int:
    """Eliminar por username. Retorna filas eliminadas."""
    resultado = db.query(Usuario).filter(Usuario.username == username).delete()
    db.commit()
    return resultado  # 0 si no encontro, 1 si elimino

Repositorio Generico (20 min)

# repositories/base.py
from typing import TypeVar, Generic, Type, Optional, List
from sqlalchemy.orm import Session
from database import Base

T = TypeVar("T", bound=Base)

class RepositorioGenerico(Generic[T]):
    """Repositorio CRUD generico."""
    
    def __init__(self, db: Session, modelo: Type[T]):
        self.db = db
        self.modelo = modelo
    
    def crear(self, entidad: T) -> T:
        """Crear nueva entidad."""
        self.db.add(entidad)
        self.db.commit()
        self.db.refresh(entidad)
        return entidad
    
    def obtener_por_id(self, id: int) -> Optional[T]:
        """Obtener por ID."""
        return self.db.query(self.modelo).filter(self.modelo.id == id).first()
    
    def obtener_todos(self, skip: int = 0, limit: int = 100) -> List[T]:
        """Obtener todos con paginacion."""
        return self.db.query(self.modelo).offset(skip).limit(limit).all()
    
    def actualizar(self, id: int, datos: dict) -> Optional[T]:
        """Actualizar por ID."""
        entidad = self.obtener_por_id(id)
        if not entidad:
            return None
        
        for key, value in datos.items():
            if hasattr(entidad, key):
                setattr(entidad, key, value)
        
        self.db.commit()
        self.db.refresh(entidad)
        return entidad
    
    def eliminar(self, id: int) -> bool:
        """Eliminar por ID."""
        entidad = self.obtener_por_id(id)
        if not entidad:
            return False
        
        self.db.delete(entidad)
        self.db.commit()
        return True


# Uso del repositorio generico
class UsuarioRepositorio(RepositorioGenerico[Usuario]):
    def __init__(self, db: Session):
        super().__init__(db, Usuario)
    
    def obtener_por_username(self, username: str) -> Optional[Usuario]:
        return self.db.query(Usuario).filter(Usuario.username == username).first()
    
    def obtener_activos(self) -> List[Usuario]:
        return self.db.query(Usuario).filter(Usuario.activo == True).all()

Resumen

• Session: Unidad de trabajo con la base de datos
• db.add(): Agregar entidad
• db.query(): Crear consulta
• .filter(): Filtrar resultados
• db.commit(): Guardar cambios
• db.rollback(): Revertir cambios