Saltar a contenido

EXCEPCIONES Y REPOSITORIO EN SQLITE

  • Módulo: PD4 - CEPY
  • Unidad: UT3 — Persistencia con bases de datos
  • Prerrequisitos: apuntes UT3-1, clases Python, herencia, excepciones básicas

1. El gestor de contexto with conn:

Por qué usarlo

En los apuntes anteriores llamábamos a commit() manualmente al terminar cada operación para que los cambios pendientes se vuelquen en disco. Pero nunca gestionamos qué ocurre si algo falla. Ahora que vamos a trabajar con excepciones necesitamos también deshacer la transacción en caso de error.

El gestor de contexto with conn: nos evita tener que hacerlo a mano:

  • Si el bloque termina sin errores → se ejecuta commit() automático. Cambios se escriben en disco. Se procesa transacción
  • Si el bloque termina con una excepción → se ejecuta rollback() de forma automática, deshaciendo los cambios en memoria. Se deshace transacción

Patrón recomendado

El siguiente patrón a la hora de ejecutar desde Python consultas en la BBDD evita gestionar manualmente las transacciones y reduce el riesgo de dejar cambios "a medias".

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
conn = sqlite3.connect("expendedora.db")    ## Conectamos/creamos fichero con base de datos
try:                                        ## Intentamos ejecutar transacción
    with conn:                              ## abre transacción
        cursor = conn.cursor()              ## objeto intermediario entre python y bbdd
        cursor.execute(
            "INSERT INTO productos VALUES (?, ?, ?, ?)",
            ("B2", "Zumo", 1.50, 20)
        )
    ## commit() automático al salir del bloque with sin error
except sqlite3.IntegrityError as e:        ## capturamos el error que puede producirse
    print(f"Error de integridad: {e}")
    ## rollback() ya fue ejecutado automáticamente por with conn:
finally:
    conn.close()                           ## siempre cerramos

No cierra la conexión

with conn: gestiona la transacción, no la conexión. Hay que cerrar la conexión con conn.close() en el bloque finally.

sqlite3.IntegrityError es uno de los tipos de excepción de SQLite — los veremos en detalle en la sección siguiente.

si el fichero no existe se crea

Recuerda que si al intentar abrir el fichero de la base de datos este no existe, se crea automáticamente. Si queremos comprobar que el fichero existe antes de abrirlo es recuerda que el módulos Path incluye el método exists() para verificar si un fichero existe: 

from pathlib import Path

if not Path("expendedora.db").exists():
    print("Error: la base de datos no existe.")   # O lanzamos una excepción
else:
    conn = sqlite3.connect("expendedora.db")

Comparado con el patrón manual de UT3-1, desaparecen las llamadas explícitas a commit() y rollback() — el código queda más limpio y menos propenso a errores.

Patrón avanzado

Si no queremos tener que cerrar manualmente la conexión tendríamos que añadir un with para la propia conexión:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
with sqlite3.connect("expendedora.db") as conn:   # abrimos contexto para conexión
    try:
        with conn:                                # abrimos contexto para transaccióncommit/rollback automático
            cursor = conn.cursor()
            cursor.execute(
                "INSERT INTO productos VALUES (?, ?, ?, ?)",
                ("B2", "Zumo", 1.50, 20)
            )
    except sqlite3.IntegrityError as e:
        print(f"Error de integridad: {e}")

De esta forma tenemos dos context managers anidados, cada uno con su responsabilidad:

  • with sqlite3.connect(...) as conn: → gestiona la conexión (la cierra al salir)
  • with conn: → gestiona la transacción (commit o rollback)

Ahora no necesitamos el finally: conn.close().

En el lab de la unidad no usaremos este patrón sino el que vimos en el apartado anterior

2. Manejo de excepciones en SQLite

Jerarquía de excepciones

El módulo sqlite3 define su propia jerarquía de excepciones, todas derivadas de sqlite3.Error:

sqlite3.Error
└── sqlite3.DatabaseError
    ├── sqlite3.IntegrityError    ← violación de restricciones (PK, FK, NOT NULL, UNIQUE)
    ├── sqlite3.OperationalError  ← errores del motor (tabla inexistente, SQL incorrecto…)
    ├── sqlite3.ProgrammingError  ← errores de uso de la API (parámetros incorrectos…)
    ├── sqlite3.DataError         ← datos fuera de rango o tipo incorrecto
    └── sqlite3.NotSupportedError ← operación no soportada por el motor

En la práctica, las dos que encontrarás con más frecuencia son IntegrityError y OperationalError.

IntegrityError — violación de restricciones

Se lanza cuando se viola una restricción de la tabla: PRIMARY KEY, NOT NULL, UNIQUE o FOREIGN KEY.

| Causa | Excepción | Ejemplo | |---|---| | Clave primaria duplicada o campos únicos | UNIQUE constraint failed | Dos productos con el mismo codigo | | Campo NOT NULL con None | NOT NULL constraint failed | None en una columna obligatoria | | Clave foránea sin referencia | FOREIGN KEY constraint failed | codigo en descuentos que no existe en productos |

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
conn = sqlite3.connect("expendedora.db")
try:
    with conn:
        cursor = conn.cursor()
        cursor.execute("INSERT INTO productos VALUES ('A1', 'Agua', 1.00, 10)")
        cursor.execute("INSERT INTO productos VALUES ('A1', 'Duplicado', 0.50, 5)")  ## codigo repetido
except sqlite3.IntegrityError as e:
    print(f"Error de integridad: {e}")
finally:
    conn.close()

Error de integridad: UNIQUE constraint failed: productos.codigo

OperationalError — errores del motor

Se lanza ante problemas del motor SQLite: tabla inexistente, SQL incorrecto, o intentar crear una tabla que ya existe.

Causa Ejemplo
Tabla inexistente SELECT * FROM tabla_que_no_existe
SQL con sintaxis incorrecta SELEC * FROM productos
Tabla ya existente sin IF NOT EXISTS CREATE TABLE productos (...) cuando ya existe 
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_no_existe")   ## la tabla no existe
except sqlite3.OperationalError as e:
    print(f"Error operacional: {e}")
finally:
    conn.close()

Error operacional: no such table: ventas

IF NOT EXIST al crear tabla

Para CREATE TABLE deberíamos ejecutar la consulta con IF NOT EXISTS, en lugar de capturar la excepción: 

1
2
3
4
5
6
7
8
cursor.execute("""
    CREATE TABLE IF NOT EXISTS productos (
        codigo   TEXT    PRIMARY KEY,
        nombre   TEXT    NOT NULL,
        precio   REAL    NOT NULL,
        cantidad INTEGER NOT NULL
    )
""")

ProgrammingError — uso incorrecto de la API

Se lanza cuando el número de parámetros ? no coincide con los valores proporcionados, o se usa la conexión después de cerrarla.

1
2
3
4
5
6
7
8
9
conn = sqlite3.connect("expendedora.db")
try:
    with conn:
        cursor = conn.cursor()
        cursor.execute("INSERT INTO productos VALUES (?, ?, ?)", ("A1", "Agua", 1.00, 10))  ## 3 ? pero 4 valores
except sqlite3.ProgrammingError as e:
    print(f"Error de programación: {e}")
finally:
    conn.close()

Error de programación: binding 4 has no name

Resumen de excepciones

Excepción Cuándo se lanza Qué hacer
sqlite3.IntegrityError PK duplicada, NOT NULL, FK sin referencia Informar al usuario — with ya hizo el rollback
sqlite3.OperationalError Tabla inexistente, SQL incorrecto, tabla ya existente Revisar la consulta o el esquema
sqlite3.ProgrammingError Número de ? incorrecto, conexión cerrada Corregir el código
sqlite3.Error Cualquier error de sqlite3 Solo como último recurso

Captura siempre la excepción más concreta primero

Ordena los except de más específico a más general. Si pones sqlite3.Error antes que sqlite3.IntegrityError, el bloque genérico lo interceptará todo y perderás información sobre la causa del fallo.

3. Excepciones propias de dominio

Por qué crear excepciones propias

Cuando el repositorio captura un sqlite3.IntegrityError y lo deja propagarse, el menú o la capa de aplicación recibe una excepción de infraestructura que no debería conocer. El menú no debería saber nada de SQLite: solo debería saber de productos, precios y stocks.

La solución es que el repositorio capture las excepciones de SQLite y las transforme en excepciones propias del dominio:

sqlite3.IntegrityError   →   ProductoYaExisteError
sqlite3.OperationalError →   ErrorPersistencia

El menú captura ProductoYaExisteError y muestra un mensaje legible. Nunca ve sqlite3.IntegrityError.

Definir excepciones propias

Las excepciones propias son clases que heredan de Exception. La convención es crear una excepción base del módulo y derivar de ella las específicas:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
class ErrorRepositorio(Exception):
    """Excepción base para 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 del motor de base de datos."""
    pass

Herencia en las excepciones propias

Que todas hereden de ErrorRepositorio permite al código que llama capturar cualquier error de persistencia con un solo except ErrorRepositorio, o capturar errores concretos con except ProductoYaExisteError. La misma lógica que con la jerarquía de sqlite3.

Usar las excepciones propias

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
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)
            )
    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 acceder a la base de datos: {e}")
    finally:
        conn.close()

Ahora el código que llama a guardar() solo necesita conocer ProductoYaExisteError:

1
2
3
4
try:
    repo.guardar(Item("A1", "Agua", 1.00, 10))
except ProductoYaExisteError as e:
    print(f"No se pudo guardar: {e}")

4. El repositorio como frontera

Las capas de la aplicación

Una aplicación bien estructurada separa el código en capas con responsabilidades distintas. Cada capa solo conoce la que tiene inmediatamente debajo:

Capa Responsabilidad
Menú / Interfaz de usuario Captura excepciones de dominio
Dominio (Item, Maquina…) Lógica de negocio, sin SQLite
Repositorio ← FRONTERA: traduce SQLite ↔ dominio
SQLite / Base de datos Lanza excepciones sqlite3.*

El repositorio es la única capa que importa sqlite3. Todo lo que está por encima solo ve objetos de dominio y excepciones de dominio.

Ejemplo completo: repositorio con excepciones propias

 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
import sqlite3
from domain.item import Item, ItemConDescuento
from infrastructure.errores import (
    ProductoYaExisteError,
    ProductoNoEncontradoError,
    ErrorPersistencia,
)

class RepositorioProductosSQLite:

    def __init__(self, ruta_bd: str):
        self._ruta_bd = ruta_bd

    def guardar(self, item: Item) -> None:
        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: {e}")
        finally:
            conn.close()

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

Cómo queda el menú

El menú nunca importa sqlite3. Solo captura excepciones de dominio:

1
2
3
4
5
6
7
8
def opcion_guardar(repo, item):
    try:
        repo.guardar(item)
        print(f"Producto '{item.codigo}' guardado correctamente.")
    except ProductoYaExisteError as e:
        print(f"Error: {e}")
    except ErrorPersistencia as e:
        print(f"Error inesperado de base de datos: {e}")

Ventaja de esta separación

Si en el futuro cambias SQLite por PostgreSQL o por un fichero JSON, solo tienes que reescribir el repositorio. El menú, el dominio y los tests de dominio no cambian porque nunca supieron que existía SQLite.