Reintentos sin duplicados: idempotencia con Python y SQLite

Una automatización recibe una respuesta incierta y vuelve a ejecutar la tarea. Si el primer intento ya había terminado, puedes acabar con dos registros para una sola petición. La IA no resuelve esa ambigüedad por escribir «hecho»: necesitas identificar la operación y guardar su resultado.

Este ejemplo implementa idempotencia local con Python y SQLite: repetir la misma clave y el mismo texto devuelve el identificador anterior, sin insertar otra tarea. No llama a un modelo ni envía correos; sirve para probar el mecanismo antes de conectarlo a una automatización con efectos reales.

La clave identifica la operación, no el intento

El sistema que recibe la petición debe asignarle una clave estable, por ejemplo solicitud-001, y conservarla en todos sus reintentos. No encargues al modelo inventar una clave nueva en cada llamada. Dos solicitudes diferentes pueden llevar el mismo texto: aquí las distinguimos por clave, no por parecido semántico.

Guardamos una tarea y su recibo en la misma base de datos y dentro de una sola transacción. El recibo relaciona clave, huella del texto e identificador de la tarea. Si alguien reutiliza la clave con otro texto, la función rechaza el conflicto en lugar de sobrescribir lo anterior.

Ejemplo completo: una tarea local y un recibo

Guarda este código como idempotencia.py. Requiere Python 3.12 o posterior por el parámetro autocommit. La ejecución de esta pieza se ha probado en Python 3.14.9 y SQLite 3.53.1, en archivos temporales locales; no en tu infraestructura.

import hashlib
import sqlite3


def conectar(ruta):
    # Python 3.12+: transacciones SQL explícitas.
    db = sqlite3.connect(ruta, autocommit=True, timeout=2.0)
    db.execute("""CREATE TABLE IF NOT EXISTS tareas (
        id INTEGER PRIMARY KEY,
        texto TEXT NOT NULL
    )""")
    db.execute("""CREATE TABLE IF NOT EXISTS recibos (
        clave TEXT PRIMARY KEY NOT NULL,
        huella TEXT NOT NULL,
        tarea_id INTEGER NOT NULL
    )""")
    return db


def registrar(db, clave, texto):
    if not isinstance(clave, str) or not clave.strip() or len(clave) > 120:
        raise ValueError("Clave no válida")
    if not isinstance(texto, str) or not texto.strip() or len(texto) > 200:
        raise ValueError("Texto no válido")
    huella = hashlib.sha256(texto.encode("utf-8")).hexdigest()
    db.execute("BEGIN IMMEDIATE")
    try:
        previo = db.execute(
            "SELECT huella, tarea_id FROM recibos WHERE clave = ?", (clave,)
        ).fetchone()
        if previo:
            if previo[0] != huella:
                raise ValueError("Clave reutilizada con otro contenido")
            tarea_id = previo[1]
        else:
            cursor = db.execute("INSERT INTO tareas(texto) VALUES (?)", (texto,))
            tarea_id = cursor.lastrowid
            db.execute(
                "INSERT INTO recibos(clave, huella, tarea_id) VALUES (?, ?, ?)",
                (clave, huella, tarea_id),
            )
        db.execute("COMMIT")
        return tarea_id
    except BaseException:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise

El valor autocommit=True permite gestionar aquí las transacciones con instrucciones SQL explícitas. Usamos parámetros ? para los valores, no concatenación de texto en consultas. La documentación de sqlite3 de Python explica ambos mecanismos y el tiempo de espera ante bloqueos.

BEGIN IMMEDIATE intenta iniciar la transacción de escritura antes de consultar el recibo. SQLite admite un solo escritor simultáneo: otro escritor puede esperar o fallar por base ocupada. No significa que cualquier número de trabajadores vaya a terminar siempre sin errores; consulta las reglas oficiales de transacciones.

La huella SHA-256 se calcula sobre los bytes UTF-8 exactos del texto con hashlib. No es cifrado ni demuestra que el contenido sea correcto. En este ejemplo, añadir un espacio cambia la entrada: si necesitas normalización, defínela antes de producir la huella y mantenla estable.

Cómo comprobar el reintento

En una carpeta de pruebas nueva, guarda el bloque anterior y ejecuta este segundo archivo en la misma carpeta. No uses una base existente de producción.

from idempotencia import conectar, registrar

# Usa una ruta nueva para esta demostración.
db = conectar("demo-idempotencia.sqlite")
try:
    primero = registrar(db, "solicitud-001", "Revisar documento sintético A")
    segundo = registrar(db, "solicitud-001", "Revisar documento sintético A")
    total = db.execute("SELECT COUNT(*) FROM tareas").fetchone()[0]
    assert primero == segundo
    assert total == 1
    print("Mismo identificador:", primero == segundo, "Tareas:", total)
finally:
    db.close()

La comprobación exige un mismo identificador y una sola fila tras dos llamadas. El archivo conserva el recibo al cerrar la conexión: mientras mantengas esa base y esa clave, el siguiente intento puede recuperar el resultado. No borres recibos por comodidad sin definir antes cuánto tiempo pueden llegar reintentos.

Qué se ha probado realmente

Se han ejecutado diez pruebas locales con datos sintéticos. Además del primer registro y su reintento, se comprobó una respuesta descartada por el llamador después de confirmar la transacción, el rechazo de una clave con texto distinto, la reapertura del archivo y que una clave nueva permite otra tarea con el mismo texto. Esta última es una limitación deliberada, no un fallo oculto.

También se rechazaron cuatro entradas no válidas dentro de una prueba de validación, se provocó un fallo al insertar el recibo para comprobar que la tarea se revierte, y dos conexiones concurrentes devolvieron el mismo identificador con una sola tarea. Un escritor reteniendo la base provocó el error por bloqueo esperado sin insertar tareas.

Son diez escenarios comprobados en este entorno, no una certificación general. Descartar un retorno en el llamador simula la incertidumbre posterior al guardado; no se ha cortado una red real, matado un proceso ni probado recuperación de un disco. El script propaga los errores: no incorpora reintentos automáticos infinitos.

El límite importante: SQLite no confirma por ti un correo o una publicación

La tarea y el recibo de este ejemplo comparten transacción. Un correo, un cobro o una publicación en un servicio externo no la comparten. Si envías primero y guardas el recibo después, un fallo entre ambos pasos sigue dejando incertidumbre. No conviertas este ejemplo en una promesa de «exactamente una vez» para cualquier API.

Antes de conectar el mecanismo, comprueba si el destino acepta una clave de idempotencia propia o permite reconciliar el objeto ya creado. Si no puedes distinguir un intento fallido de un éxito sin respuesta, detén el reenvío y revisa el estado. No pruebes este patrón enviando mensajes, pagos o publicaciones reales sin autorización.

La idempotencia tampoco sustituye permisos, validación de datos ni revisión del resultado. Para comprobar la estructura de una respuesta antes de guardarla, consulta cómo validar el JSON de una IA con Python. Son controles distintos: uno comprueba la entrada; el otro evita repetir el efecto local de una operación identificada.

Contenido elaborado con asistencia de inteligencia artificial a partir de las fuentes indicadas.