Tu inventario pierde datos

Reto1 · La consulta que devuelve de más

Un parámetro ? representa un valor y recibe una tupla de argumentos. No concatena texto SQL. Escribir (sku,) incluye la coma necesaria para una tupla de un elemento. fetchone devuelve una fila o None. Convertimos la fila en un diccionario independiente. Los parámetros no sirven para nombres de tablas o columnas: esas partes deben ser estáticas o elegirse de una lista cerrada. El laboratorio contiene un SKU con apóstrofo para que la consulta correcta funcione también con él. Define buscar_sku(con,sku): devuelve dict con id,sku,nombre,stock,precio_centimos,version o None. Consulta parametrizada, sin escribir ni imprimir; el texto de entrada nunca cambia la población consultada.

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA

def buscar_sku(con, sku):
    # Predice, implementa y comprueba.
    pass

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA

def buscar_sku(con, sku):
    fila = con.execute('SELECT id,sku,nombre,stock,precio_centimos,version FROM productos WHERE sku=?', (sku,)).fetchone()
    return dict(fila) if fila is not None else None

Archivo de apoyo: laboratorio.py

Guárdalo con este nombre junto a main.py de este reto. No lo pegues dentro de main.py.

import sqlite3
SCHEMA = "\nCREATE TABLE IF NOT EXISTS productos(\n id INTEGER PRIMARY KEY,\n sku TEXT NOT NULL UNIQUE CHECK(length(trim(sku)) BETWEEN 1 AND 40),\n nombre TEXT NOT NULL CHECK(length(trim(nombre)) BETWEEN 1 AND 80),\n stock INTEGER NOT NULL CHECK(typeof(stock)='integer' AND stock>=0),\n precio_centimos INTEGER NOT NULL CHECK(typeof(precio_centimos)='integer' AND precio_centimos>=0),\n version INTEGER NOT NULL DEFAULT 1 CHECK(version>=1));\nCREATE TABLE IF NOT EXISTS movimientos(\n id INTEGER PRIMARY KEY, producto_id INTEGER NOT NULL REFERENCES productos(id),\n delta INTEGER NOT NULL CHECK(typeof(delta)='integer' AND delta!=0),\n motivo TEXT NOT NULL CHECK(length(trim(motivo)) BETWEEN 1 AND 100));\n"
def fresca(vacia=False):
    con = sqlite3.connect(':memory:', isolation_level='DEFERRED', autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL)
    con.row_factory = sqlite3.Row
    con.execute('PRAGMA foreign_keys=ON')
    con.executescript(SCHEMA)
    if not vacia:
        con.executemany('INSERT INTO productos(id,sku,nombre,stock,precio_centimos) VALUES(?,?,?,?,?)',
                        [(1,'A','Adaptador',5,250),(2,'B','Cable',0,150),(3,"O'X",'Conector',2,250)])
        con.commit()
    return con

Archivo de apoyo: apoyo.py

Guárdalo con este nombre junto a main.py de este reto. No lo pegues dentro de main.py.

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA

Comprueba: Consulta un SKU vacío y otro con comillas, punto y coma y espacios; ninguno debería romper la tabla.

Reto2 · Lo veía, pero no estaba guardado

Una inserción puede verse desde su conexión y seguir sin confirmar. Cerrar esa conexión descarta la transacción pendiente en el modelo DEFERRED. La evidencia de persistencia se obtiene abriendo otra conexión al mismo archivo. El código inicial olvida confirmar: pulsa Comprobar y observa el fallo «Se ve tras cerrar». Ejecutar solo define la función; no la llama ni imprime None. Añade el límite de transacción y repite. En el playground ruta es un archivo virtual temporal de esta ejecución; en local puedes conservarlo en una carpeta de prácticas. Define guardar_y_reabrir(ruta,sku): abre conexión, prepara esquema y guarda producto nuevo (nombre Nuevo,stock1,precio100) con parámetros y commit. Siempre cierra y devuelve la lectura desde otra conexión. SKU válido y no existente como precondición; duplicado propaga IntegrityError sin otra fila. Usa apoyos del reto.

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA
from apoyo import buscar_sku, conectar, crear_esquema, leer_archivo

def guardar_y_reabrir(ruta, sku):
    with closing(conectar(ruta)) as con:
        crear_esquema(con)
        con.execute('INSERT INTO productos(sku,nombre,stock,precio_centimos) VALUES(?,?,?,?)', (sku, 'Nuevo', 1, 100))
    return leer_archivo(ruta, sku)

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA
from apoyo import buscar_sku, conectar, crear_esquema, leer_archivo

def guardar_y_reabrir(ruta, sku):
    with closing(conectar(ruta)) as con:
        crear_esquema(con)
        with con:
            con.execute('INSERT INTO productos(sku,nombre,stock,precio_centimos) VALUES(?,?,?,?)', (sku, 'Nuevo', 1, 100))
    return leer_archivo(ruta, sku)

Archivo de apoyo: laboratorio.py

Guárdalo con este nombre junto a main.py de este reto. No lo pegues dentro de main.py.

import sqlite3
SCHEMA = "\nCREATE TABLE IF NOT EXISTS productos(\n id INTEGER PRIMARY KEY,\n sku TEXT NOT NULL UNIQUE CHECK(length(trim(sku)) BETWEEN 1 AND 40),\n nombre TEXT NOT NULL CHECK(length(trim(nombre)) BETWEEN 1 AND 80),\n stock INTEGER NOT NULL CHECK(typeof(stock)='integer' AND stock>=0),\n precio_centimos INTEGER NOT NULL CHECK(typeof(precio_centimos)='integer' AND precio_centimos>=0),\n version INTEGER NOT NULL DEFAULT 1 CHECK(version>=1));\nCREATE TABLE IF NOT EXISTS movimientos(\n id INTEGER PRIMARY KEY, producto_id INTEGER NOT NULL REFERENCES productos(id),\n delta INTEGER NOT NULL CHECK(typeof(delta)='integer' AND delta!=0),\n motivo TEXT NOT NULL CHECK(length(trim(motivo)) BETWEEN 1 AND 100));\n"
def fresca(vacia=False):
    con = sqlite3.connect(':memory:', isolation_level='DEFERRED', autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL)
    con.row_factory = sqlite3.Row
    con.execute('PRAGMA foreign_keys=ON')
    con.executescript(SCHEMA)
    if not vacia:
        con.executemany('INSERT INTO productos(id,sku,nombre,stock,precio_centimos) VALUES(?,?,?,?,?)',
                        [(1,'A','Adaptador',5,250),(2,'B','Cable',0,150),(3,"O'X",'Conector',2,250)])
        con.commit()
    return con

Archivo de apoyo: apoyo.py

Guárdalo con este nombre junto a main.py de este reto. No lo pegues dentro de main.py.

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA

def conectar(ruta, *, lectura=False):
    ubicacion = Path(ruta).resolve().as_uri() + '?mode=ro' if lectura else ruta
    con = sqlite3.connect(ubicacion, uri=lectura, timeout=1.0, isolation_level='DEFERRED', autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL)
    con.row_factory = sqlite3.Row
    con.execute('PRAGMA foreign_keys=ON')
    return con

def crear_esquema(con):
    if con.in_transaction:
        raise ValueError('esquema-requiere-conexion-sin-transaccion')
    con.executescript(SCHEMA)

def buscar_sku(con, sku):
    fila = con.execute('SELECT id,sku,nombre,stock,precio_centimos,version FROM productos WHERE sku=?', (sku,)).fetchone()
    return dict(fila) if fila is not None else None

def leer_archivo(ruta, sku):
    with closing(conectar(ruta, lectura=True)) as con:
        return buscar_sku(con, sku)

Comprueba: En tu equipo repite sin with con y cierra: compara los resultados en otro proceso.

Reto3 · Sale stock, pero falta el movimiento

Una salida de inventario cambia stock y añade auditoría. Si confirmas el UPDATE y después falla INSERT, queda una salida sin explicación. Usa una sola transacción y permite que la excepción salga del bloque para rollback. No llames a retirar dentro del bloque: esa función confirma su propio trabajo y rompería el límite de la operación compuesta. La prueba instala un trigger ficticio que rechaza el movimiento; tendrás que demostrar que también se conserva el stock. Define registrar_salida(con,sku,cantidad,motivo): valida cantidad positiva y motivo recortado1–100; en un with con hace UPDATE condicional e INSERT de delta negativo. Falta de producto/stock ValueError. Error INSERT propaga y revierte todo. True al confirmar; entrada sin transacción abierta. Rechaza una transacción ya abierta con ValueError, sin confirmarla ni deshacerla.

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA
from apoyo import exigir_transaccion_propia

def registrar_salida(con, sku, cantidad, motivo):
    # Predice, implementa y comprueba.
    pass

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA
from apoyo import exigir_transaccion_propia

def registrar_salida(con, sku, cantidad, motivo):
    exigir_transaccion_propia(con)
    if type(cantidad) is not int or cantidad <= 0:
        raise ValueError('cantidad-invalida')
    if not isinstance(motivo, str) or not 1 <= len(motivo.strip()) <= 100:
        raise ValueError('motivo-invalido')
    with con:
        cur = con.execute('UPDATE productos SET stock=stock-?,version=version+1 WHERE sku=? AND stock>=?', (cantidad, sku, cantidad))
        if cur.rowcount != 1:
            raise ValueError('stock-o-producto-no-disponible')
        identidad = con.execute('SELECT id FROM productos WHERE sku=?', (sku,)).fetchone()[0]
        con.execute('INSERT INTO movimientos(producto_id,delta,motivo) VALUES(?,?,?)', (identidad, -cantidad, motivo.strip()))
    return True

Archivo de apoyo: laboratorio.py

Guárdalo con este nombre junto a main.py de este reto. No lo pegues dentro de main.py.

import sqlite3
SCHEMA = "\nCREATE TABLE IF NOT EXISTS productos(\n id INTEGER PRIMARY KEY,\n sku TEXT NOT NULL UNIQUE CHECK(length(trim(sku)) BETWEEN 1 AND 40),\n nombre TEXT NOT NULL CHECK(length(trim(nombre)) BETWEEN 1 AND 80),\n stock INTEGER NOT NULL CHECK(typeof(stock)='integer' AND stock>=0),\n precio_centimos INTEGER NOT NULL CHECK(typeof(precio_centimos)='integer' AND precio_centimos>=0),\n version INTEGER NOT NULL DEFAULT 1 CHECK(version>=1));\nCREATE TABLE IF NOT EXISTS movimientos(\n id INTEGER PRIMARY KEY, producto_id INTEGER NOT NULL REFERENCES productos(id),\n delta INTEGER NOT NULL CHECK(typeof(delta)='integer' AND delta!=0),\n motivo TEXT NOT NULL CHECK(length(trim(motivo)) BETWEEN 1 AND 100));\n"
def fresca(vacia=False):
    con = sqlite3.connect(':memory:', isolation_level='DEFERRED', autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL)
    con.row_factory = sqlite3.Row
    con.execute('PRAGMA foreign_keys=ON')
    con.executescript(SCHEMA)
    if not vacia:
        con.executemany('INSERT INTO productos(id,sku,nombre,stock,precio_centimos) VALUES(?,?,?,?,?)',
                        [(1,'A','Adaptador',5,250),(2,'B','Cable',0,150),(3,"O'X",'Conector',2,250)])
        con.commit()
    return con

Archivo de apoyo: apoyo.py

Guárdalo con este nombre junto a main.py de este reto. No lo pegues dentro de main.py.

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA

def exigir_transaccion_propia(con):
    if con.in_transaction:
        raise ValueError('operacion-requiere-conexion-sin-transaccion')

Comprueba: Haz fallar el movimiento con un trigger en tu copia de laboratorio y observa stock antes y después.

Tu turno

Repite con un SKU con apóstrofo, cierra sin confirmar una inserción y provoca un fallo entre las dos escrituras. Explica qué dato se conserva y por qué.

import sqlite3
import os
from pathlib import Path
from contextlib import closing
from laboratorio import SCHEMA
from apoyo import exigir_transaccion_propia

def registrar_salida(con, sku, cantidad, motivo):
    exigir_transaccion_propia(con)
    if type(cantidad) is not int or cantidad <= 0:
        raise ValueError('cantidad-invalida')
    if not isinstance(motivo, str) or not 1 <= len(motivo.strip()) <= 100:
        raise ValueError('motivo-invalido')
    with con:
        cur = con.execute('UPDATE productos SET stock=stock-?,version=version+1 WHERE sku=? AND stock>=?', (cantidad, sku, cantidad))
        if cur.rowcount != 1:
            raise ValueError('stock-o-producto-no-disponible')
        identidad = con.execute('SELECT id FROM productos WHERE sku=?', (sku,)).fetchone()[0]
        con.execute('INSERT INTO movimientos(producto_id,delta,motivo) VALUES(?,?,?)', (identidad, -cantidad, motivo.strip()))
    return True

Explícalo con tus palabras

• ¿Qué demuestra el SKU con apóstrofo?
• ¿Leíste desde otra conexión después de cerrar?
• ¿Se conserva stock cuando falla el movimiento?

Repite con datos distintos antes de marcarlo como practicado. Esta es una autoevaluación, no una certificación.

Si quieres continuar

Tu primera base de datos con Python amplía el recorrido con CRUD, versiones, importación atómica, informes, copias, migración y un proyecto persistente. Estos tres retos y su guía siguen siendo gratuitos.