Saltar a contenido

Lab guiado: Excepciones y repositorio SQLite

  • Unidad: UT3 — Persistencia con bases de datos
  • Prerrequisitos: Lab UT3-A1
  • Agrupamiento: Parejas (conductor/navegante)
  • Recursos:
  • Ficheros de trabajo: probar_errores.py, probar_repositorio.py, infrastructure/errores.py, infrastructure/repositorio_sqlite.py, infrastructure/repositorio_memoria.py, infrastructure/datos_iniciales.py, presentation/menu.py, domain/repositorio_productos.py, domain/maquina.py, tests/test_maquina.py

Cómo trabajar en parejas

Este lab se realiza en parejas conductor/navegante. Rotad de rol al final de cada paso.

Rol Qué hace
Conductor Escribe el código y ejecuta los pasos. Solo el conductor toca el teclado.
Navegante Lee el enunciado, propone soluciones y detecta errores.

Entrega

Carpeta comprimida en formato zip con todo el proyecto (la carpeta que contiene expendedora/, crear_bd.py y expendedora.db).

Antes de empezar

Este lab parte del proyecto expendedora/ del lab anterior con la base de datos ya creada. Todos los comandos se ejecutan desde la carpeta padre de expendedora/ — la misma desde la que arrancas la aplicación con python -m expendedora.presentation.menu.

Verifica que tienes el fichero expendedora.db y que los tests del proyecto siguen pasando:

python -m unittest

Paso 1 — Capturar IntegrityError · probar_errores.py

Conceptos

IntegrityError se lanza cuando se viola una restricción de la tabla: PRIMARY KEY duplicada, campo NOT NULL con None, o FOREIGN KEY sin referencia. Con with conn: el rollback() es automático — solo hay que capturar el error e informar.

Objetivo

Comprobar qué ocurre cuando intentamos insertar un producto con un código ya existente.

Tarea

Crea un fichero probar_errores.py en la raíz del proyecto con el siguiente contenido:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
import sqlite3

conn = sqlite3.connect("expendedora.db")
try:
    with conn:
        cursor = conn.cursor()
        cursor.execute("INSERT INTO productos VALUES ('A1', 'Agua duplicada', 0.50, 3)")
except sqlite3.IntegrityError as e:
    print(f"IntegrityError: {e}")
finally:
    conn.close()

Ejecuta el script:

python probar_errores.py

Comprobación

IntegrityError: UNIQUE constraint failed: productos.codigo

Pregunta

¿Era necesario llamar a rollback() manualmente? ¿Por qué?

Respuesta

No. with conn: llama a rollback() automáticamente cuando el bloque termina con una excepción. Solo necesitamos capturar el error e informar.

Paso 2 — Capturar IntegrityError por None en campo NOT NULL · probar_errores.py

Conceptos

En Python, None se traduce a NULL en SQL. Si una columna tiene la restricción NOT NULL, insertar None lanza IntegrityError.

Objetivo

Comprobar el comportamiento al insertar None en un campo obligatorio.

Tarea

Añade al final de probar_errores.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
conn = sqlite3.connect("expendedora.db")
try:
    with conn:
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO productos VALUES (?, ?, ?, ?)",
            ("Z1", None, 1.00, 5)   # nombre es None → NULL → IntegrityError
        )
except sqlite3.IntegrityError as e:
    print(f"IntegrityError por None: {e}")
finally:
    conn.close()

Ejecuta y observa la salida.

Comprobación

IntegrityError por None: NOT NULL constraint failed: productos.nombre

Pregunta

¿Qué pasaría si en lugar de None pasaras ""? ¿También lanzaría error?

Respuesta

No. "" es una cadena vacía, no NULL. SQLite lo acepta aunque el campo tenga NOT NULL, porque sí tiene un valor (texto vacío). Por eso es importante validar los datos en Python antes de insertarlos.

Paso 3 — Capturar OperationalError · probar_errores.py

Conceptos

OperationalError se lanza ante problemas del motor: tabla inexistente, SQL incorrecto, fichero bloqueado, etc.

Objetivo

Comprobar qué ocurre al consultar una tabla que no existe.

Tarea

Añade al final de probar_errores.py:

1
2
3
4
5
6
7
8
9
conn = sqlite3.connect("expendedora.db")
try:
    with conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM ventas")   # tabla inexistente
except sqlite3.OperationalError as e:
    print(f"OperationalError: {e}")
finally:
    conn.close()

Comprobación

OperationalError: no such table: ventas

Pregunta

¿Qué diferencia hay entre IntegrityError y OperationalError? ¿Cuándo aparece cada uno?

Respuesta

IntegrityError aparece cuando los datos violan una regla de la tabla (duplicado, nulo, FK). OperationalError aparece cuando hay un problema con el motor o el esquema (tabla inexistente, SQL mal escrito). Son causas distintas y por eso tienen excepciones distintas.

Paso 4 — Definir excepciones propias de dominio · expendedora/infrastructure/errores.py

Conceptos

El menú no debería conocer sqlite3. Si capturamos sqlite3.IntegrityError en el menú, el menú depende de la infraestructura. La solución es que el repositorio transforme las excepciones de SQLite en excepciones propias de dominio que el menú sí conoce.

Objetivo

Crear el módulo de excepciones propias del repositorio.

Tarea

Crea el fichero expendedora/infrastructure/errores.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
class ErrorRepositorio(Exception):
    """Excepción base para todos los errores de persistencia."""
    pass


class ProductoYaExisteError(ErrorRepositorio):
    """Se lanza cuando se intenta guardar un producto con un código ya existente."""
    pass


class ProductoNoEncontradoError(ErrorRepositorio):
    """Se lanza cuando se busca un producto que no existe en la base de datos."""
    pass


class ErrorPersistencia(ErrorRepositorio):
    """Se lanza ante cualquier otro error inesperado del motor de base de datos."""
    pass

No ejecutes nada todavía: este módulo lo usaremos en el paso siguiente.

Pregunta

¿Por qué todas las excepciones propias heredan de ErrorRepositorio y no directamente de Exception?

Respuesta

Para que el código que llama pueda capturar cualquier error de persistencia con un solo except ErrorRepositorio, o capturar errores concretos con except ProductoYaExisteError. Es la misma lógica de jerarquía que sqlite3.Error → sqlite3.IntegrityError.

Paso 5 — Crear el repositorio con excepciones propias · expendedora/infrastructure/repositorio_sqlite.py

Conceptos

El repositorio es la frontera entre la base de datos y el resto de la aplicación. Captura las excepciones de sqlite3 y las transforma en excepciones de dominio. El menú solo ve ProductoYaExisteError, nunca sqlite3.IntegrityError.

Objetivo

Implementar los métodos guardar() y obtener() del repositorio con manejo de excepciones.

Tarea

Crea el fichero expendedora/infrastructure/repositorio_sqlite.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
import sqlite3
from expendedora.domain.item import Item, ItemConDescuento
from expendedora.infrastructure.errores import (
    ProductoYaExisteError,
    ProductoNoEncontradoError,
    ErrorPersistencia,
)


class RepositorioProductosSQLite:

    def __init__(self, ruta_bd="expendedora.db"):
        self._ruta_bd = ruta_bd

    def guardar(self, item):
        conn = sqlite3.connect(self._ruta_bd)
        try:
            with conn:
                cursor = conn.cursor()
                cursor.execute("PRAGMA foreign_keys = ON")
                cursor.execute(
                    "INSERT INTO productos VALUES (?, ?, ?, ?)",
                    (item.codigo, item.nombre, item.precio, item.cantidad)
                )
                if isinstance(item, ItemConDescuento):
                    cursor.execute(
                        "INSERT INTO descuentos VALUES (?, ?)",
                        (item.codigo, item.porcentaje_descuento)
                    )
        except sqlite3.IntegrityError:
            raise ProductoYaExisteError(
                f"Ya existe un producto con código '{item.codigo}'"
            )
        except sqlite3.OperationalError as e:
            raise ErrorPersistencia(f"Error al guardar el producto: {e}")
        finally:
            conn.close()

    def obtener(self, codigo):
        conn = sqlite3.connect(self._ruta_bd)
        try:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT * FROM productos WHERE codigo = ?", (codigo,)
            )
            fila = cursor.fetchone()
            if fila is None:
                raise ProductoNoEncontradoError(
                    f"No existe ningún producto con código '{codigo}'"
                )
            cod, nombre, precio, cantidad = fila
            cursor.execute(
                "SELECT porcentaje FROM descuentos WHERE codigo = ?", (cod,)
            )
            descuento = cursor.fetchone()
            if descuento:
                return ItemConDescuento(cod, nombre, precio, cantidad, descuento[0])
            return Item(cod, nombre, precio, cantidad)
        except sqlite3.OperationalError as e:
            raise ErrorPersistencia(f"Error al leer el producto: {e}")
        finally:
            conn.close()

Pregunta

¿Por qué obtener() no usa with conn:?

Respuesta

Porque obtener() solo lee datos (SELECT), no modifica la base de datos. with conn: gestiona transacciones de escritura (commit/rollback). Para lecturas puras no es necesario, aunque usarlo tampoco es incorrecto.

Patrón común de los métodos

Todos los metodos del repositorio siguen la misma estructura:

1. Abrir conexion        →  conn = sqlite3.connect(...)
2. Operar con la BD      →  try: ... (con with conn: si escribe)
3. Transformar errores   →  except sqlite3.XxxError: raise MiError(...)
4. Cerrar conexion       →  finally: conn.close()

Este patrón garantiza que:

  • La conexion siempre se cierra (incluso si hay error).
  • Las excepciones de sqlite3 nunca escapan del repositorio.
  • El resto de la aplicacion solo ve excepciones de dominio.

Paso 6 — Probar el repositorio · probar_repositorio.py (raíz del proyecto)

Objetivo

Verificar que el repositorio lanza las excepciones de dominio correctas.

Tarea

Crea probar_repositorio.py en la raíz del proyecto (junto a crear_bd.py):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from expendedora.domain.item import Item
from expendedora.infrastructure.repositorio_sqlite import RepositorioProductosSQLite
from expendedora.infrastructure.errores import ProductoYaExisteError, ProductoNoEncontradoError

repo = RepositorioProductosSQLite("expendedora.db")

# Prueba 1: obtener un producto existente
print("--- Prueba 1: obtener producto existente ---")
item = repo.obtener("A1")
print(f"{item.codigo} - {item.nombre} - {item.precio} EUR - stock: {item.cantidad}")

# Prueba 2: obtener un producto que no existe
print("\n--- Prueba 2: producto no encontrado ---")
try:
    repo.obtener("X9")
except ProductoNoEncontradoError as e:
    print(f"Error esperado: {e}")

# Prueba 3: guardar un producto nuevo
print("\n--- Prueba 3: guardar producto nuevo ---")
try:
    repo.guardar(Item("C1", "Chocolate", 2.00, 12))
    print("Producto C1 guardado correctamente.")
except ProductoYaExisteError as e:
    print(f"Ya existía: {e}")

# Prueba 4: intentar guardar un duplicado
print("\n--- Prueba 4: duplicado ---")
try:
    repo.guardar(Item("A1", "Agua duplicada", 0.50, 3))
except ProductoYaExisteError as e:
    print(f"Error esperado: {e}")

Ejecuta:

python probar_repositorio.py

Comprobación

--- Prueba 1: obtener producto existente ---
A1 - Agua - 1.0 EUR - stock: 10

--- Prueba 2: producto no encontrado ---
Error esperado: No existe ningún producto con código 'X9'

--- Prueba 3: guardar producto nuevo ---
Producto C1 guardado correctamente.

--- Prueba 4: duplicado ---
Error esperado: Ya existe un producto con código 'A1'

Paso 7 — Integrar el repositorio en el menú · expendedora/presentation/menu.py

Conceptos

El menú captura excepciones de dominio, nunca excepciones de sqlite3. Esto garantiza que si en el futuro cambias la base de datos, el menú no cambia.

En este proyecto el manejo de errores está centralizado en el main() de menu.py: las funciones individuales del menú propagan las excepciones hacia arriba y main() las captura todas en un único bloque try/except.

Objetivo

Conectar el menú al repositorio SQLite y añadir manejo de errores en el bucle principal para que ProductoYaExisteError y ErrorPersistencia se muestren correctamente sin cerrar la aplicación.

Tarea

1. Añade crear_servicio_sqlite() a infrastructure/datos_iniciales.py

Al final del fichero, añade esta función:

1
2
3
4
5
6
7
8
def crear_servicio_sqlite(ruta_bd="expendedora.db"):
    """Crea y devuelve un ServicioExpendedora con persistencia en SQLite."""
    from expendedora.domain.maquina import MaquinaExpendedora
    from expendedora.application.servicios import ServicioExpendedora
    from expendedora.infrastructure.repositorio_sqlite import RepositorioProductosSQLite
    repo = RepositorioProductosSQLite(ruta_bd)
    maquina = MaquinaExpendedora(repo)
    return ServicioExpendedora(maquina)

¿Por qué una función nueva?

En el lab anterior, datos_iniciales.py ya tenia crear_servicio(), que monta toda la cadena de objetos para que la aplicacion funcione con el repositorio en memoria.

La nueva crear_servicio_sqlite() hace exactamente lo mismo pero usando RepositorioProductosSQLite en lugar del repositorio en memoria.

Ademas, no precarga datos porque con SQLite los productos ya estan en el fichero expendedora.db (se crearon con crear_bd.py).

2. Abre expendedora/presentation/menu.py y realiza tres cambios:

2a. Sustituye el import de crear_servicio por crear_servicio_sqlite

Antes: 

from expendedora.infrastructure.datos_iniciales import crear_servicio

Después: 

1
2
from expendedora.infrastructure.datos_iniciales import crear_servicio_sqlite
from expendedora.infrastructure.errores import ProductoYaExisteError, ErrorPersistencia

2b. Actualiza la llamada en main()

Antes: 

servicio = crear_servicio()

Después: 

1
servicio = crear_servicio_sqlite()

2c. Amplía el bloque except de main()

Localiza al final de main() el except ValueError. Añade dos nuevos except antes de él, para que el bloque quede así:

1
2
3
4
5
6
        except ProductoYaExisteError as e:
            print("X " + str(e))
        except ErrorPersistencia as e:
            print("X " + str(e))
        except ValueError as e:
            print("X " + str(e))

Orden de los except

ProductoYaExisteError y ErrorPersistencia deben ir antes de ValueError, porque Python recorre los except en orden y se queda con el primero que coincide.

Pregunta

Las funciones opcion_agregar_producto y opcion_agregar_producto_con_descuento no tienen try/except propio. ¿Cómo llegan las excepciones hasta main()?

Respuesta

Cuando una función no captura una excepción, Python la propaga hacia arriba en la pila de llamadas hasta encontrar un except que la maneje. Como opcion_agregar_producto es llamada desde dentro del try de main(), cualquier excepción que lance llegará al bloque except de main().

Comprobación

Estado parcial

En este punto, el repositorio SQLite solo tiene guardar() y obtener(). Las opciones del menú que necesitan listar() (como mostrar productos) o actualizar() (como comprar o reponer) no funcionan todavía. Se completarán en el paso 8. Por ahora, verifica únicamente la opción 7 insertando un producto con un código duplicado.

Ejecuta la aplicación. Recuerda que en la carpeta padre de expendedora debes ejecutar python -m expendedora.presentation.menu

Intenta añadir un producto con un código ya existente (por ejemplo A1). El menú debe mostrar el mensaje de error y volver al menú sin cerrarse:

Elige una opcion: 7
Codigo: A1
Nombre: Agua
Precio: 1.0
Cantidad: 5
X Ya existe un producto con código 'A1'

Paso 8 — Adaptar el dominio al nuevo contrato · domain/repositorio_productos.py · domain/maquina.py · infrastructure/repositorio_sqlite.py

Conceptos

Con el repositorio en memoria, modificar item.cantidad en Python sí actualizaba el almacén, porque el diccionario guarda la misma referencia al objeto. Con SQLite, obtener() devuelve un objeto nuevo cada vez; modificarlo en Python no tiene ningún efecto en la base de datos.

Además, obtener() ahora lanza ProductoNoEncontradoError si el código no existe, en lugar de devolver None. Cualquier código de maquina.py que compruebe if item is None: es código muerto con SQLite.

Objetivo

Hacer que comprar(), reponer() y agregar_item() funcionen correctamente con SQLite añadiendo el método actualizar() al repositorio y adaptando la lógica del dominio.

Tarea

1. Actualiza el contrato del dominio

Abre expendedora/domain/repositorio_productos.py y añade el método actualizar() junto al resto de métodos:

1
2
3
    def actualizar(self, item):
        """Persiste los cambios de un item ya existente."""
        raise NotImplementedError

¿Por qué hay que añadirlo al contrato?

Si maquina.py va a llamar a self._repo.actualizar(item), el contrato debe incluir ese método. Así se garantiza que cualquier repositorio (memoria, SQLite o uno futuro) lo implementa.

2. Completa el repositorio SQLite

Abre expendedora/infrastructure/repositorio_sqlite.py y añade los tres métodos siguientes después de obtener():

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
    def actualizar(self, item):
        """Persiste la cantidad actualizada del item en la BD."""
        conn = sqlite3.connect(self._ruta_bd)
        try:
            with conn:
                cursor = conn.cursor()
                cursor.execute(
                    "UPDATE productos SET cantidad = ? WHERE codigo = ?",
                    (item.cantidad, item.codigo),
                )
        except sqlite3.OperationalError as e:
            raise ErrorPersistencia(f"Error al actualizar el producto: {e}")
        finally:
            conn.close()

    def eliminar(self, codigo):
        """Elimina el producto y su descuento (si tiene) de la BD."""
        conn = sqlite3.connect(self._ruta_bd)
        try:
            with conn:
                cursor = conn.cursor()
                cursor.execute("PRAGMA foreign_keys = ON")
                cursor.execute("DELETE FROM descuentos WHERE codigo = ?", (codigo,))
                cursor.execute("DELETE FROM productos WHERE codigo = ?", (codigo,))
        except sqlite3.OperationalError as e:
            raise ErrorPersistencia(f"Error al eliminar el producto: {e}")
        finally:
            conn.close()

    def listar(self):
        """Devuelve todos los productos como objetos Item o ItemConDescuento."""
        conn = sqlite3.connect(self._ruta_bd)
        try:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM productos")
            filas = cursor.fetchall()
            items = []
            for cod, nombre, precio, cantidad in filas:
                cursor.execute(
                    "SELECT porcentaje FROM descuentos WHERE codigo = ?", (cod,)
                )
                descuento = cursor.fetchone()
                if descuento:
                    items.append(ItemConDescuento(cod, nombre, precio, cantidad, descuento[0]))
                else:
                    items.append(Item(cod, nombre, precio, cantidad))
            return items
        except sqlite3.OperationalError as e:
            raise ErrorPersistencia(f"Error al listar los productos: {e}")
        finally:
            conn.close()

¿Por qué eliminar() borra primero de descuentos?

La tabla descuentos tiene una FOREIGN KEY que referencia a productos. Si intentaras borrar de productos primero, SQLite lanzaría un IntegrityError porque quedaría un registro huérfano en descuentos. El orden correcto es borrar siempre el lado dependiente antes que el referenciado.

Pregunta

¿Por qué actualizar() solo cambia cantidad y no todos los campos del item?

Respuesta

En este dominio, precio y nombre no cambian después de crear el producto. Solo cantidad varía (por compras de usuarios y reposiciones de mantenimiento de la máquina). Actualizar únicamente lo que cambia es más eficiente y menos propenso a errores.

Paso 9 — Adaptar dominio, memoria y tests al nuevo contrato · domain/maquina.py · infrastructure/repositorio_memoria.py · tests/test_maquina.py

Conceptos

obtener() ahora lanza ProductoNoEncontradoError si el código no existe, en lugar de devolver None. Cualquier código de maquina.py que compruebe if item is None: es código muerto con el nuevo contrato. Además, maquina.py debe llamar a actualizar() tras modificar la cantidad para que el cambio se persista en SQLite.

Objetivo

Adaptar maquina.py, el repositorio en memoria y los tests para que sean compatibles con el nuevo contrato del repositorio.

Tarea

1. Corrige maquina.py

Abre expendedora/domain/maquina.py y aplica los cuatro cambios siguientes:

a) agregar_item — elimina la comprobación de existencia previa:

Antes: 

    def agregar_item(self, item):
        if not isinstance(item, Item):
            raise ValueError("Solo se permiten objetos Item.")
        if self._repo.obtener(item.codigo) is not None:   # ← BUG: obtener() lanza excepción si no existe
            raise ValueError("El codigo ya existe.")
        self._repo.guardar(item)

Después: 

1
2
3
4
5
    def agregar_item(self, item):
        if not isinstance(item, Item):
            raise ValueError("Solo se permiten objetos Item.")
        # guardar() lanzará ProductoYaExisteError si el código ya existe.
        self._repo.guardar(item)

b) comprar — persiste el stock reducido:

Localiza el bloque que decrementa item.cantidad y añade la llamada a actualizar():

1
2
3
4
5
        item.cantidad = item.cantidad - 1
        if item.cantidad == 0:
            self._repo.eliminar(item.codigo)
        else:
            self._repo.actualizar(item)   # ← nuevo: persiste el stock en SQLite

c) seleccionar — elimina el código muerto:

Antes: 

item = self._repo.obtener(codigo)
if item is None:                          # ← nunca True: obtener() lanza excepción antes
    raise ValueError("El codigo no existe.")
self._seleccion = item

Después: 

1
2
item = self._repo.obtener(codigo)         # lanza ProductoNoEncontradoError si no existe
self._seleccion = item

d) reponer — elimina el código no usado y persiste el stock:

Antes: 

        item = self._repo.obtener(codigo)
        if item is None:                          # ← nunca True: obtener() lanza excepción antes
            raise ValueError("El codigo no existe.")
        item.cantidad = item.cantidad + unidades

Después: 

1
2
3
        item = self._repo.obtener(codigo)         # lanza ProductoNoEncontradoError si no existe
        item.cantidad = item.cantidad + unidades
        self._repo.actualizar(item)               # ← nuevo: persiste el stock en SQLite

2. Actualiza el repositorio en memoria

Al cambiar maquina.py para que ya no compruebe if item is None, el repositorio en memoria deja de ser compatible: su obtener() devuelve None cuando el código no existe, pero ahora el dominio espera que se lance ProductoNoEncontradoError.

Abre expendedora/infrastructure/repositorio_memoria.py y aplica estos cuatro cambios:

a) Añade los imports de las excepciones de dominio:

1
from expendedora.infrastructure.errores import ProductoNoEncontradoError, ProductoYaExisteError

b) guardar() — lanza ProductoYaExisteError en lugar de ValueError:

Antes: 

        if item.codigo in self._por_codigo:
            raise ValueError("El codigo ya existe.")

Después: 

1
2
        if item.codigo in self._por_codigo:
            raise ProductoYaExisteError(f"Ya existe un producto con código '{item.codigo}'")

c) obtener() — lanza ProductoNoEncontradoError en lugar de devolver None:

Antes: 

    def obtener(self, codigo):
        """Devuelve un item o None si no existe."""
        return self._por_codigo.get(codigo)

Después: 

1
2
3
4
5
6
    def obtener(self, codigo):
        """Devuelve un item o lanza ProductoNoEncontradoError si no existe."""
        item = self._por_codigo.get(codigo)
        if item is None:
            raise ProductoNoEncontradoError(f"No existe ningún producto con código '{codigo}'")
        return item

d) Añade actualizar() — en memoria no se usa porque los objetos se modifican en el diccionario

1
2
3
    def actualizar(self, item):
        """En memoria el objeto ya está mutado por referencia; no hace nada."""
        pass

Pregunta

¿Por qué actualizar() no hace nada en el repositorio en memoria?

Respuesta

El diccionario almacena una referencia al objeto, no una copia. Al hacer item.cantidad = item.cantidad - 1 en maquina.py, el propio objeto del diccionario ya queda modificado. En SQLite, en cambio, obtener() reconstruye un objeto nuevo desde la BD cada vez, por lo que los cambios en Python no se reflejan en la base de datos sin un UPDATE explícito.

3. Corrige los tests afectados

Al eliminar los if item is None de maquina.py y actualizar el repositorio en memoria para que lance ProductoNoEncontradoError, dos tests que esperaban ValueError para código inexistente dejan de pasar.

Abre expendedora/tests/test_maquina.py y aplica estos dos cambios:

a) Añade el import:

1
from expendedora.infrastructure.errores import ProductoNoEncontradoError

b) Actualiza los dos tests:

1
2
3
    def test_seleccionar_codigo_inexistente_lanza_error(self):
        with self.assertRaises(ProductoNoEncontradoError):
            self.maquina.seleccionar("ZZ")

1
2
3
    def test_reponer_codigo_inexistente_lanza_error(self):
        with self.assertRaises(ProductoNoEncontradoError):
            self.maquina.reponer("X9", 3)

Pregunta

Antes los tests esperaban ValueError. ¿Por qué ahora esperan ProductoNoEncontradoError?

Respuesta

El dominio ya no lanza ValueError para código inexistente — esa responsabilidad la asume el repositorio, que lanza ProductoNoEncontradoError. Los tests deben reflejar el comportamiento real del sistema. Usar la excepción específica también documenta mejor qué tipo de error se espera.

4. Verifica con la aplicación y los tests

Recrea la BD y arranca el menú. En la carpeta padre de expendedora ejecuta:

python crear_bd.py
python -m expendedora.presentation.menu

Comprueba que puedes: - Añadir un producto nuevo (opción 7 u 8) - Comprar y que el stock baje al volver a listar - Reponer y que el stock suba

Luego ejecuta los tests:

python -m unittest

Comprobación

................................................
----------------------------------------------------------------------
Ran 48 tests in 0.Xs

OK

Reflexión

Respuestas
  • El repositorio en memoria funcionaba por referencia: mutar el objeto también mutaba el almacén.
  • SQLite devuelve copias independientes: cualquier cambio de estado necesita una operación UPDATE explícita.
  • actualizar() extiende el contrato del repositorio con una nueva responsabilidad: persistir cambios de estado.
  • Eliminar comprobaciones redundantes (if item is None) simplifica el dominio y lo hace consistente con el nuevo contrato.

Katas

Kata 1 — Eliminar producto desde el menú

Kata

El método eliminar() ya existe en el repositorio, pero solo se llama internamente desde comprar() cuando el stock llega a 0. Añade una nueva opción al menú que pida un código y elimine el producto correspondiente.

Pistas:

  1. Crea una función opcion_eliminar_producto(servicio) en menu.py. Debe pedir el código al usuario y llamar a un método del servicio que se encargue de la eliminación.
  2. En servicios.py, añade un método eliminar_producto(codigo) que delegue en la máquina.
  3. En maquina.py, añade un método eliminar_item(codigo) que llame a self._repo.obtener(codigo) (para verificar que existe) y después a self._repo.eliminar(codigo).
  4. No necesitas capturar nada en opcion_eliminar_producto: si el código no existe, obtener() lanzará ProductoNoEncontradoError, que se propagará hasta el try/except centralizado de main().
  5. Recuerda añadir la nueva opción al texto del menú y al if/elif de main().

Comprobación: ejecuta la aplicación, elimina un producto existente y verifica que desaparece al listar. Intenta eliminar un código inexistente y comprueba que el menú muestra el error sin cerrarse.

Kata 2 — Buscar productos por nombre

Kata

Añade una operación buscar_por_nombre(nombre) que devuelva una lista de productos cuyo nombre contenga el texto buscado. Debe funcionar tanto con el repositorio SQLite como con el de memoria.

Pistas:

  1. Empieza por el contrato: añade buscar_por_nombre(self, nombre) a repositorio_productos.py con raise NotImplementedError.
  2. En el repositorio SQLite, usa SELECT ... WHERE nombre LIKE ? con el parámetro f"%{nombre}%" para buscar coincidencias parciales. Recuerda consultar también la tabla descuentos para devolver el tipo correcto (Item o ItemConDescuento).
  3. En el repositorio en memoria, filtra el diccionario con una comprensión de lista: [item for item in self._por_codigo.values() if nombre.lower() in item.nombre.lower()].
  4. Añade buscar_producto(nombre) a servicios.py y una función opcion_buscar_producto(servicio) al menú.
  5. Decide qué hacer cuando no hay resultados: puedes devolver una lista vacía y mostrar un mensaje, sin necesidad de lanzar una excepción.

Comprobación: busca "agua" y verifica que aparece el producto. Busca un texto que no coincida con nada y comprueba que el menú muestra un mensaje adecuado.

Resumen de conceptos

Excepciones de sqlite3 (pasos 1-3)

  • IntegrityError salta cuando los datos violan una restricción de la tabla (duplicado, None en NOT NULL, FK rota).
  • OperationalError salta cuando falla el motor o el esquema (tabla inexistente, SQL incorrecto, BD bloqueada).
  • with conn: gestiona la transacción automáticamente: hace commit si todo va bien y rollback si salta una excepción. No es necesario llamar a rollback() manualmente.
  • None se traduce a NULL en SQL y viola NOT NULL. Una cadena vacía "" no es NULL y SQLite la acepta.

Excepciones propias de dominio (paso 4)

  • El menú no debe importar sqlite3. Si lo hiciera, dependería de la infraestructura.
  • El repositorio transforma las excepciones de SQLite en excepciones propias (ProductoYaExisteError, ProductoNoEncontradoError, ErrorPersistencia).
  • Todas heredan de ErrorRepositorio para poder capturarlas de forma genérica o específica.

Repositorio SQLite (pasos 5-6, 8)

  • Todos los métodos siguen el mismo patrón: abrir conexión → operar → transformar errores → cerrar conexión en finally.
  • PRAGMA foreign_keys = ON activa la comprobación de claves foráneas (SQLite la trae desactivada por defecto).
  • Los ? en el SQL son parámetros sustituidos que evitan inyección SQL.
  • fetchone() devuelve una tupla o None; fetchall() devuelve una lista de tuplas.
  • El contrato (repositorio_productos.py) define qué operaciones puede pedir el dominio. Si se añade un método nuevo como actualizar(), debe declararse primero en el contrato.

Integración en el menú (paso 7)

  • crear_servicio_sqlite() monta la misma cadena de objetos que crear_servicio(), pero usando RepositorioProductosSQLite en lugar del repositorio en memoria. El resto de la aplicación no cambia.
  • Las excepciones se propagan desde las funciones del menú hasta el try/except centralizado en main().
  • Los except deben ir de más específico a más general.

Adaptar al nuevo contrato (paso 9)

  • Con el repositorio en memoria, modificar un objeto en Python sí modifica el almacén (misma referencia). Con SQLite, obtener() devuelve un objeto nuevo cada vez; hay que llamar a actualizar() para hacer el UPDATE en la BD.
  • obtener() ahora lanza excepción en lugar de devolver None. Las comprobaciones if item is None: pasan a ser código muerto y se eliminan.
  • Los tests deben reflejar el comportamiento real: si el repositorio lanza ProductoNoEncontradoError, los tests deben esperar esa excepción, no ValueError.

Checklist de entrega

  • He comprobado que IntegrityError se lanza al insertar un código duplicado.
  • He comprobado que IntegrityError se lanza al insertar None en un campo NOT NULL.
  • He comprobado que OperationalError se lanza al consultar una tabla inexistente.
  • He creado infrastructure/errores.py con las cuatro excepciones de dominio.
  • He añadido actualizar() al contrato en repositorio_productos.py.
  • He implementado RepositorioProductosSQLite con guardar(), obtener(), actualizar(), eliminar() y listar().
  • He verificado que el repositorio lanza excepciones de dominio, no excepciones de sqlite3.
  • He actualizado RepositorioProductosMemoria para que obtener() lance ProductoNoEncontradoError, guardar() lance ProductoYaExisteError y tenga el método actualizar().
  • He añadido crear_servicio_sqlite() a datos_iniciales.py y actualizado menu.py para usarla.
  • El menú captura ProductoYaExisteError sin importar sqlite3.
  • He corregido maquina.py: comprar() y reponer() persisten el stock, agregar_item() delega la comprobación de duplicados y seleccionar() elimina el código muerto.
  • He corregido los dos tests que esperaban ValueError para que esperen ProductoNoEncontradoError.
  • La aplicación completa funciona: comprar reduce stock, reponer lo aumenta, agregar añade productos.