En la especificación DB-API 2.0 de Python, las operaciones de manipulación de datos se realizan a través del objeto Cursor. Sin embargo, para sentencias simples, la clase Connection provee atajos (conn.execute()) que instancian un cursor implícito.
Regla de Oro de Seguridad: Independientemente de la operación, nunca uses concatenación de cadenas (
f-stringso.format()) para inyectar variables en SQL. SQLite en Python utiliza el marcador de posición?(qmark style) para enlazar parámetros de forma segura a nivel de C, neutralizando cualquier vector de inyección SQL.
6.1. Insertar Datos (Create)
Para inserciones simples, execute() es suficiente. Sin embargo, en escenarios reales de ingeniería, la inserción masiva (bulk insert) es el cuello de botella más común.
Usar un bucle for con execute() genera una sobrecarga masiva porque SQLite inicia y hace commit de una transacción por cada iteración. La solución arquitectónica es utilizar executemany() dentro de un bloque transaccional explícito.
import sqlite3
import logging
def bulk_insert_usuarios(conn: sqlite3.Connection, usuarios: list[tuple]) -> int:
"""
Inserta múltiples registros en una sola transacción minimizando las llamadas I/O.
Retorna el número de filas insertadas.
"""
query = "INSERT INTO usuarios (uuid, email) VALUES (?, ?);"
try:
# El context manager 'with conn:' emite BEGIN y COMMIT automáticamente
with conn:
cursor = conn.executemany(query, usuarios)
return cursor.rowcount
except sqlite3.IntegrityError as e:
# Captura específica para violaciones de esquema (ej. UNIQUE, NOT NULL)
logging.error(f"Fallo de integridad de datos: {e}")
raise6.2. Leer Datos (Read)
El principal riesgo al leer datos es la saturación de la memoria RAM (Out-Of-Memory) al procesar conjuntos de resultados masivos. El antipatrón más común es usar fetchall(), que carga toda la tabla en una lista de Python.
Estrategia de Memoria Eficiente:
El objeto Cursor de Python es un generador (implementa el protocolo iterador). La forma más óptima en memoria para leer registros es iterar directamente sobre el cursor, lo que hace streaming de las filas desde el motor C a Python bajo demanda.
def obtener_usuarios_activos(conn: sqlite3.Connection) -> None:
"""
Itera sobre un set de resultados de forma perezosa (lazy evaluation),
manteniendo la huella de memoria plana sin importar el tamaño de la tabla.
"""
query = "SELECT id, uuid, email FROM usuarios WHERE activo = ?;"
# conn.execute retorna un Cursor implícito
cursor = conn.execute(query, (1,))
# Iteración directa (Streaming)
for fila in cursor:
# Si conn.row_factory = sqlite3.Row fue configurado (Punto 1),
# 'fila' permite acceso por clave como un dict.
procesar_usuario(fila['uuid'], fila['email'])
# Para paginación explícita controlada, usar cursor.fetchmany(size=100)6.3. Actualizar Datos (Update)
La actualización requiere confirmación de estado. Ejecutar un UPDATE sintácticamente correcto no garantiza que los datos hayan cambiado (por ejemplo, si la cláusula WHERE no coincide con ningún registro). La propiedad cursor.rowcount es crítica para que la capa de aplicación valide el estado de la mutación.
def desactivar_sesiones_expiradas(conn: sqlite3.Connection, threshold_date: str) -> int:
"""
Actualiza registros condicionalmente y valida el impacto de la operación.
"""
query = "UPDATE sesiones SET estado = 'expirada' WHERE fecha_creacion < ? AND estado = 'activa';"
with conn:
cursor = conn.execute(query, (threshold_date,))
filas_afectadas = cursor.rowcount
if filas_afectadas == 0:
logging.info("No se encontraron sesiones para expirar.")
return filas_afectadas6.4. Eliminar Datos (Delete)
Más allá de la sintaxis estándar de DELETE, que opera idénticamente a UPDATE verificando el rowcount, la eliminación en SQLite tiene una implicación estructural a nivel de sistema de archivos.
Cuando eliminas filas con DELETE, SQLite no reduce el tamaño del archivo .db ni devuelve el espacio al sistema operativo. En su lugar, marca las páginas de memoria como “libres” para ser reutilizadas en futuras inserciones.
Mantenimiento (El comando VACUUM):
En sistemas donde ocurren eliminaciones masivas frecuentes y el espacio en disco es una restricción, es necesario programar la ejecución del comando VACUUM. Este comando reconstruye la base de datos entera, eliminando la fragmentación y liberando el espacio al OS.
def purgar_logs_antiguos(conn: sqlite3.Connection) -> None:
"""
Elimina registros obsoletos y desfragmenta el archivo de la base de datos.
"""
query_delete = "DELETE FROM logs_sistema WHERE nivel = 'DEBUG';"
try:
with conn:
cursor = conn.execute(query_delete)
logging.info(f"Purgados {cursor.rowcount} registros de log.")
# VACUUM no puede ejecutarse dentro de una transacción activa
# Debe ejecutarse en modo autocommit
conn.execute("VACUUM;")
logging.info("Base de datos desfragmentada y espacio liberado.")
except sqlite3.Error as e:
logging.error(f"Error en la purga de mantenimiento: {e}")