Muchos problemas de mantenimiento aparecen cuando una misma función lee datos, valida reglas, calcula resultados, guarda información y muestra mensajes. Ese código puede funcionar, pero se vuelve difícil de probar y de modificar.
En este tema separaremos cuatro responsabilidades: entrada, reglas de negocio, persistencia y salida. El objetivo no es crear una arquitectura compleja, sino ordenar el código para que cada parte tenga un motivo claro para cambiar.
Para este refactoring usaremos una separación práctica:
Cuando estas responsabilidades están separadas, las reglas pueden probarse sin archivos, consola, base de datos ni red.
Crea el archivo src/importador_pedidos.py con este código inicial:
import csv
def importar_pedidos(ruta_csv):
total_general = 0
pedidos_guardados = []
with open(ruta_csv, newline="", encoding="utf-8") as archivo:
lector = csv.DictReader(archivo)
for fila in lector:
cantidad = int(fila["cantidad"])
precio = float(fila["precio"])
if cantidad <= 0:
print(f"Pedido inválido: {fila['cliente']}")
continue
if precio <= 0:
print(f"Pedido inválido: {fila['cliente']}")
continue
subtotal = cantidad * precio
if subtotal >= 10000:
descuento = subtotal * 0.10
else:
descuento = 0
total = subtotal - descuento
pedido = {
"cliente": fila["cliente"],
"producto": fila["producto"],
"cantidad": cantidad,
"precio": precio,
"total": total,
}
pedidos_guardados.append(pedido)
print(f"Pedido guardado para {pedido['cliente']}: {pedido['total']}")
total_general += total
print(f"Total importado: {total_general}")
return pedidos_guardadosEl código lee el archivo, parsea datos, valida reglas, calcula descuentos, guarda en una lista y escribe en consola dentro de la misma función.
La función tiene varias razones para cambiar. Si cambia el formato del CSV, cambia esta función. Si cambia la regla del descuento, también. Si mañana guardamos en una base de datos, vuelve a cambiar.
Además, probar la regla de descuento exige preparar un archivo CSV y revisar salidas de consola. La prueba termina verificando demasiado al mismo tiempo.
Antes de separar responsabilidades, protegemos el comportamiento observable. Crea tests/test_importador_pedidos.py:
from src.importador_pedidos import importar_pedidos
def test_importa_pedidos_validos_desde_csv(tmp_path):
archivo = tmp_path / "pedidos.csv"
archivo.write_text(
"cliente,producto,cantidad,precio\n"
"Ana,Teclado,2,3000\n"
"Luis,Monitor,1,12000\n",
encoding="utf-8",
)
pedidos = importar_pedidos(archivo)
assert pedidos == [
{
"cliente": "Ana",
"producto": "Teclado",
"cantidad": 2,
"precio": 3000.0,
"total": 6000.0,
},
{
"cliente": "Luis",
"producto": "Monitor",
"cantidad": 1,
"precio": 12000.0,
"total": 10800.0,
},
]python -m pytest tests/test_importador_pedidos.pyLa entrada convierte texto en datos del dominio. Podemos extraer una función que transforme una fila del CSV en un diccionario con tipos correctos:
def parsear_fila_pedido(fila):
return {
"cliente": fila["cliente"],
"producto": fila["producto"],
"cantidad": int(fila["cantidad"]),
"precio": float(fila["precio"]),
}Esta función todavía no aplica reglas de negocio. Solo traduce datos externos a una estructura interna.
El parseo puede probarse sin archivos si recibe una fila como diccionario:
from src.importador_pedidos import parsear_fila_pedido
def test_parsea_fila_de_pedido():
fila = {
"cliente": "Ana",
"producto": "Teclado",
"cantidad": "2",
"precio": "3000",
}
pedido = parsear_fila_pedido(fila)
assert pedido == {
"cliente": "Ana",
"producto": "Teclado",
"cantidad": 2,
"precio": 3000.0,
}La validación y el cálculo del total pertenecen al negocio. Podemos expresarlos con funciones puras:
def pedido_es_valido(pedido):
return pedido["cantidad"] > 0 and pedido["precio"] > 0
def calcular_total_pedido(pedido):
subtotal = pedido["cantidad"] * pedido["precio"]
if subtotal >= 10000:
return subtotal * 0.90
return subtotalEstas funciones no leen archivos, no guardan datos y no muestran mensajes.
Ahora las pruebas se concentran en las decisiones importantes:
from src.importador_pedidos import calcular_total_pedido, pedido_es_valido
def test_pedido_es_valido_si_cantidad_y_precio_son_positivos():
pedido = {"cantidad": 2, "precio": 3000.0}
assert pedido_es_valido(pedido) is True
def test_pedido_no_es_valido_si_cantidad_es_cero():
pedido = {"cantidad": 0, "precio": 3000.0}
assert pedido_es_valido(pedido) is False
def test_calcula_total_sin_descuento():
pedido = {"cantidad": 2, "precio": 3000.0}
assert calcular_total_pedido(pedido) == 6000.0
def test_calcula_total_con_descuento():
pedido = {"cantidad": 1, "precio": 12000.0}
assert calcular_total_pedido(pedido) == 10800.0python -m pytest tests/test_importador_pedidos.pyEl caso de uso coordina las piezas. Recibe pedidos ya parseados y delega guardado y salida a colaboradores externos:
def procesar_pedidos(pedidos, repositorio, salida):
total_general = 0
pedidos_guardados = []
for pedido in pedidos:
if not pedido_es_valido(pedido):
salida.informar_pedido_invalido(pedido["cliente"])
continue
pedido_procesado = {
**pedido,
"total": calcular_total_pedido(pedido),
}
repositorio.guardar(pedido_procesado)
salida.informar_pedido_guardado(pedido_procesado)
pedidos_guardados.append(pedido_procesado)
total_general += pedido_procesado["total"]
salida.informar_total(total_general)
return pedidos_guardadosEsta función no sabe si los pedidos vienen de CSV, API o formulario. Tampoco sabe si se guardan en memoria, archivo o base de datos.
Para mantener el ejemplo simple, usaremos un repositorio en memoria. En una aplicación real podría ser una base de datos:
class RepositorioPedidosEnMemoria:
def __init__(self):
self.pedidos = []
def guardar(self, pedido):
self.pedidos.append(pedido)La persistencia tiene una interfaz pequeña: guardar un pedido procesado.
La salida también puede aislarse. Este objeto escribe en consola, pero el caso de uso no depende de print directamente:
class SalidaConsola:
def informar_pedido_invalido(self, cliente):
print(f"Pedido inválido: {cliente}")
def informar_pedido_guardado(self, pedido):
print(f"Pedido guardado para {pedido['cliente']}: {pedido['total']}")
def informar_total(self, total):
print(f"Total importado: {total}")Si más adelante la salida debe ir a logs, JSON o una interfaz web, cambiaremos esta clase sin tocar la regla de negocio.
La lectura del CSV queda reducida a producir una lista de pedidos parseados:
def leer_pedidos_csv(ruta_csv):
with open(ruta_csv, newline="", encoding="utf-8") as archivo:
lector = csv.DictReader(archivo)
return [parsear_fila_pedido(fila) for fila in lector]Esta función pertenece al borde de entrada. Su responsabilidad termina cuando entrega datos internos.
La función original puede conservarse como punto de entrada público, pero ahora solo conecta piezas:
def importar_pedidos(ruta_csv):
pedidos = leer_pedidos_csv(ruta_csv)
repositorio = RepositorioPedidosEnMemoria()
salida = SalidaConsola()
return procesar_pedidos(pedidos, repositorio, salida)Esto mantiene compatibilidad con el código que ya llamaba a importar_pedidos.
import csv
def parsear_fila_pedido(fila):
return {
"cliente": fila["cliente"],
"producto": fila["producto"],
"cantidad": int(fila["cantidad"]),
"precio": float(fila["precio"]),
}
def pedido_es_valido(pedido):
return pedido["cantidad"] > 0 and pedido["precio"] > 0
def calcular_total_pedido(pedido):
subtotal = pedido["cantidad"] * pedido["precio"]
return subtotal * 0.90 if subtotal >= 10000 else subtotal
class RepositorioPedidosEnMemoria:
def __init__(self):
self.pedidos = []
def guardar(self, pedido):
self.pedidos.append(pedido)
class SalidaConsola:
def informar_pedido_invalido(self, cliente):
print(f"Pedido inválido: {cliente}")
def informar_pedido_guardado(self, pedido):
print(f"Pedido guardado para {pedido['cliente']}: {pedido['total']}")
def informar_total(self, total):
print(f"Total importado: {total}")
def procesar_pedidos(pedidos, repositorio, salida):
total_general = 0
pedidos_guardados = []
for pedido in pedidos:
if not pedido_es_valido(pedido):
salida.informar_pedido_invalido(pedido["cliente"])
continue
pedido_procesado = {**pedido, "total": calcular_total_pedido(pedido)}
repositorio.guardar(pedido_procesado)
salida.informar_pedido_guardado(pedido_procesado)
pedidos_guardados.append(pedido_procesado)
total_general += pedido_procesado["total"]
salida.informar_total(total_general)
return pedidos_guardados
def leer_pedidos_csv(ruta_csv):
with open(ruta_csv, newline="", encoding="utf-8") as archivo:
lector = csv.DictReader(archivo)
return [parsear_fila_pedido(fila) for fila in lector]
def importar_pedidos(ruta_csv):
pedidos = leer_pedidos_csv(ruta_csv)
repositorio = RepositorioPedidosEnMemoria()
salida = SalidaConsola()
return procesar_pedidos(pedidos, repositorio, salida)El caso de uso puede probarse con colaboradores falsos:
from src.importador_pedidos import procesar_pedidos
class RepositorioFalso:
def __init__(self):
self.pedidos = []
def guardar(self, pedido):
self.pedidos.append(pedido)
class SalidaFalsa:
def __init__(self):
self.eventos = []
def informar_pedido_invalido(self, cliente):
self.eventos.append(("invalido", cliente))
def informar_pedido_guardado(self, pedido):
self.eventos.append(("guardado", pedido["cliente"]))
def informar_total(self, total):
self.eventos.append(("total", total))
def test_procesa_pedidos_validos_e_invalidos():
repositorio = RepositorioFalso()
salida = SalidaFalsa()
pedidos = [
{"cliente": "Ana", "producto": "Teclado", "cantidad": 2, "precio": 3000.0},
{"cliente": "Luis", "producto": "Mouse", "cantidad": 0, "precio": 1000.0},
]
resultado = procesar_pedidos(pedidos, repositorio, salida)
assert resultado == [{
"cliente": "Ana",
"producto": "Teclado",
"cantidad": 2,
"precio": 3000.0,
"total": 6000.0,
}]
assert repositorio.pedidos == resultado
assert salida.eventos == [
("guardado", "Ana"),
("invalido", "Luis"),
("total", 6000.0),
]python -m pytest tests/test_importador_pedidos.pyDespués del refactoring, cada prueba puede enfocarse en una responsabilidad:
El resultado es código más explícito y con menor costo de cambio.
Refactoriza una función que lee usuarios desde un CSV, valida el email, guarda usuarios válidos y muestra un resumen. Separa al menos estas piezas:
Escribe primero pruebas para la validación y para el procesamiento sin usar archivos reales.
Antes de continuar, verifica que puedes explicar estos puntos:
En este tema separamos un flujo mezclado en responsabilidades claras. La lógica de negocio quedó aislada de archivos, consola y persistencia, y el caso de uso pasó a coordinar piezas pequeñas.
En el próximo tema refactorizaremos manejo de errores y excepciones sin ocultar fallos.