E5: Base de Datos SQL

Requisitos Previos

REQUIERE E4 (Persistencia Archivos) completada. Se migrará de JSON a SQL manteniendo el mismo interfaz Repository. Los endpoints NO deben cambiar.

Clases Relacionadas

Estas clases preparan los conocimientos necesarios para esta evaluación:

Clase 02: SQLAlchemy Modelos DeclarativeBase, Columnas, tipos
Clase 03: SQLAlchemy Relaciones ForeignKey, relationship, backref

Clase 04: CRUD SQLAlchemy Session, add, commit, query
Clase 05: Alembic Migraciones Versionado de base de datos

Objetivo de Aprendizaje

Migrar la persistencia a base de datos relacional con SQLAlchemy ORM y migraciones Alembic. Mantener la misma interfaz Repository para que los endpoints sean idénticos.

Modalidad de Trabajo

Individual o en parejas (máximo 2 personas). Debe ser el mismo equipo de E1-E4.

Formato de entrega: Repositorio GitHub con: modelos SQLAlchemy, migraciones Alembic en carpeta alembic/, base de datos SQLite o script para crearla. README.md con comando alembic upgrade head.

Requisitos Técnicos Detallados

1. Modelos SQLAlchemy

Descripción: Definir modelos ORM que mapean a tablas de base de datos.

Modelos obligatorios:

UsuarioModel(Base): Tabla usuarios
ProyectoModel(Base): Tabla proyectos con FK a usuarios
TareaModel(Base): Tabla tareas con FK a proyectos

Columnas requeridas:

Usuario: id (PK), username (unique), email, nombre_completo, activo, fecha_registro
Proyecto: id (PK), nombre, descripcion, lider_id (FK), fecha_creacion
Tarea: id (PK), titulo, descripcion, prioridad, estado, proyecto_id (FK), fechas

Separación de modelos:

Modelos ORM en infrastructure/models.py
Clases de dominio en domain/ permanecen igual
El repositorio hace la conversión entre ambos

2. Relaciones One-to-Many

Descripción: Configurar relaciones entre tablas correctamente.

Relaciones obligatorias:

Usuario → Proyectos: Un usuario líder puede tener muchos proyectos
Proyecto → Tareas: Un proyecto tiene muchas tareas (composición)

Configuración con relationship:

relationship("ProyectoModel", back_populates="lider")
relationship("TareaModel", back_populates="proyecto", cascade="all, delete-orphan")

Cascade delete: Al eliminar un proyecto, sus tareas deben eliminarse automáticamente (composición).

3. Repository SQLAlchemy

Descripción: Nueva implementación del Repository usando SQL en lugar de JSON.

Implementaciones obligatorias:

UsuarioRepositorySQL(IUsuarioRepository)
ProyectoRepositorySQL(IProyectoRepository)
TareaRepositorySQL(ITareaRepository)

Uso de Session:

Inyectar Session de SQLAlchemy vía Depends()
Usar session.add(), session.commit(), session.rollback()
Manejo de excepciones de integridad (IntegrityError)

Conversión ORM ↔ Dominio:

_to_domain(model): Convierte Modelo ORM a clase de dominio
_to_model(entity): Convierte clase de dominio a Modelo ORM

4. Migraciones Alembic

Descripción: Versionar la base de datos con migraciones automáticas.

Comandos obligatorios:

alembic init alembic: Inicializar Alembic
alembic revision --autogenerate -m "Initial tables": Crear migración
alembic upgrade head: Aplicar migraciones

Archivos requeridos:

alembic.ini: Configuración (URL de base de datos)
alembic/env.py: Importar modelos, setear target_metadata
alembic/versions/*.py: Migraciones generadas

PROHIBIDO: Usar Base.metadata.create_all(). Obligatorio usar migraciones Alembic.

Ejemplo de Implementación

infrastructure/models.py

from datetime import datetime
from sqlalchemy import String, Integer, Boolean, DateTime, ForeignKey, Enum as SQLEnum
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from src.domain.enums import PrioridadTarea, EstadoTarea


class Base(DeclarativeBase):
    """Base para todos los modelos."""
    pass


class UsuarioModel(Base):
    """Modelo ORM para usuarios."""
    __tablename__ = "usuarios"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    nombre_completo: Mapped[str | None] = mapped_column(String(100))
    activo: Mapped[bool] = mapped_column(default=True)
    fecha_registro: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # Relación con proyectos donde es líder
    proyectos: Mapped[list["ProyectoModel"]] = relationship(
        "ProyectoModel", back_populates="lider"
    )


class ProyectoModel(Base):
    """Modelo ORM para proyectos."""
    __tablename__ = "proyectos"

    id: Mapped[int] = mapped_column(primary_key=True)
    nombre: Mapped[str] = mapped_column(String(100), nullable=False)
    descripcion: Mapped[str | None] = mapped_column(String(500))
    fecha_creacion: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # Foreign Key al líder
    lider_id: Mapped[int] = mapped_column(ForeignKey("usuarios.id"))
    lider: Mapped["UsuarioModel"] = relationship("UsuarioModel", back_populates="proyectos")

    # Relación con tareas (cascade delete)
    tareas: Mapped[list["TareaModel"]] = relationship(
        "TareaModel", back_populates="proyecto", cascade="all, delete-orphan"
    )


class TareaModel(Base):
    """Modelo ORM para tareas."""
    __tablename__ = "tareas"

    id: Mapped[int] = mapped_column(primary_key=True)
    titulo: Mapped[str] = mapped_column(String(100), nullable=False)
    descripcion: Mapped[str | None] = mapped_column(String(500))
    prioridad: Mapped[PrioridadTarea] = mapped_column(
        SQLEnum(PrioridadTarea), default=PrioridadTarea.MEDIA
    )
    estado: Mapped[EstadoTarea] = mapped_column(
        SQLEnum(EstadoTarea), default=EstadoTarea.PENDIENTE
    )
    fecha_creacion: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    fecha_completado: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

    # Foreign Key al proyecto padre
    proyecto_id: Mapped[int] = mapped_column(ForeignKey("proyectos.id"))
    proyecto: Mapped["ProyectoModel"] = relationship("ProyectoModel", back_populates="tareas")

infrastructure/repository_sql.py

from typing import Optional, List
from sqlalchemy.orm import Session
from src.domain.usuario import Usuario
from src.infrastructure.models import UsuarioModel
from src.infrastructure.repository import IUsuarioRepository


class UsuarioRepositorySQL(IUsuarioRepository):
    """Implementación SQL del repositorio de usuarios."""

    def __init__(self, session: Session):
        self.session = session

    def _to_domain(self, model: UsuarioModel) -> Usuario:
        """Convierte modelo ORM a dominio."""
        return Usuario(
            username=model.username,
            email=model.email,
            nombre_completo=model.nombre_completo
        )

    def _to_model(self, usuario: Usuario, usuario_id: int = None) -> UsuarioModel:
        """Convierte dominio a modelo ORM."""
        model = UsuarioModel(
            username=usuario.username,
            email=usuario.email,
            nombre_completo=usuario.nombre_completo,
            activo=usuario.activo,
            fecha_registro=usuario.fecha_registro
        )
        if usuario_id:
            model.id = usuario_id
        return model

    def guardar(self, usuario: Usuario, usuario_id: int = None) -> int:
        """Guarda un usuario. Retorna el ID."""
        try:
            model = self._to_model(usuario, usuario_id)
            self.session.add(model)
            self.session.commit()
            return model.id
        except Exception as e:
            self.session.rollback()
            raise

    def obtener(self, usuario_id: int) -> Optional[Usuario]:
        """Obtiene un usuario por ID."""
        model = self.session.query(UsuarioModel).filter_by(id=usuario_id).first()
        return self._to_domain(model) if model else None

    def listar(self) -> List[Usuario]:
        """Lista todos los usuarios."""
        models = self.session.query(UsuarioModel).all()
        return [self._to_domain(m) for m in models]

    def eliminar(self, usuario_id: int) -> bool:
        """Elimina un usuario."""
        model = self.session.query(UsuarioModel).filter_by(id=usuario_id).first()
        if model:
            self.session.delete(model)
            self.session.commit()
            return True
        return False

Estructura de Entrega

Organización de Archivos Obligatoria

src/infrastructure/ ├── __init__.py ├── database.py # engine, SessionLocal, get_db() ├── models.py # Modelos SQLAlchemy ORM ├── repository.py # Interfaces (sin cambios) ├── repository_json.py # Implementación JSON (backup) └── repository_sql.py # Nueva implementación SQL api/ ├── main.py # Sin cambios en endpoints └── dependencies.py # Cambiar a RepositorySQL alembic/ # Migraciones ├── env.py ├── versions/ │ └── xxxx_initial_tables.py └── script.py.mako alembic.ini # Configuración taskflow.db # Base de datos SQLite requirements.txt # + sqlalchemy, alembic

Pasos para Completar la Evaluación

1Configurar SQLAlchemy

engine, SessionLocal, get_db() en database.py

2Crear modelos ORM

UsuarioModel, ProyectoModel, TareaModel con relaciones

3Configurar Alembic

alembic init, editar env.py, crear migración inicial

4Implementar RepositorySQL

Mismos métodos que JSON pero con Session

5Actualizar dependencias

Cambiar Depends(get_json_repo) a Depends(get_sql_repo)

6Probar migración

alembic upgrade head, verificar que API funciona igual

Rúbrica de Evaluación (100%)

Criterio	Excelente (5.0)	Aceptable (3.5)	Insuficiente	Peso
ORM y Modelos	Modelos bien definidos, relaciones correctas, tipos apropiados	Modelos básicos, relaciones parciales	Sin ORM o SQL crudo en strings	30%
Migraciones Alembic	alembic upgrade head funciona, migraciones limpias	Migración funciona pero con warnings	Sin Alembic o create_all()	25%
Repository SQL	Implementa interfaz, conversión ORM↔Dominio correcta	Repository funcional sin separación dominio	Lógica SQL mezclada en endpoints	25%
Integración	API idéntica, solo cambió Depends(), tests pasan	Cambios menores en endpoints	API rota o reescrita	20%

Penalizaciones

SQL crudo (strings): Nota máxima 2.0
Sin Alembic (usando create_all): Nota máxima 3.0
Filtrar en Python en lugar de SQL: -1.0 punto (ineficiente)
Sin cascade delete: -0.5 puntos
Sin relaciones configuradas: -1.0 punto

Conexión con E6

Con la base de datos SQL funcionando, estarás listo para la E6: Sustentación Final donde presentarás el proyecto completo.

← E4: Archivos E6: Final →

Evaluación 5

Persistencia Relacional SQL