Parseando el contrato con pydantic

Descripción

La lección 3 cerró con una advertencia: yaml.safe_load() convierte el YAML en un diccionario de Python, pero no garantiza absolutamente nada sobre su estructura — si a alguien se le olvida un campo, o escribe on_violation: "borrar_todo" en vez de un valor reconocido, ese diccionario roto pasa sin ningún aviso. Esta lección cierra esa brecha: construye DataContract y ColumnContract, dos clases de pydantic que definen, con precisión de tipos, cómo se ve un contrato válido — y demuestra, con un YAML roto a propósito, exactamente qué pasa cuando alguien no lo respeta.

Conexión con el módulo. Esta lección convierte el diccionario sin garantías de la lección 3 en un objeto de Python con tipos y validación real. La lección 5 toma ese objeto ya validado y lo convierte en un esquema de Pandera ejecutable.

Una analogía: el formulario que rechaza una fecha imposible antes de archivarla

Piensa en dos formularios de trámite. El primero es una hoja de papel en blanco: puedes escribir cualquier cosa en cualquier casillero, y el empleado que la recibe la archiva sin revisarla — el error, si lo hay, se descubre semanas después, cuando alguien intenta usar esa información y no tiene sentido. El segundo es un formulario digital con validación: si escribes 32 en el campo "día del mes", el sistema rechaza el envío en el acto, con un mensaje claro sobre cuál casillero está mal y por qué. yaml.safe_load(), por sí solo, es la hoja de papel en blanco — lee cualquier estructura que el YAML tenga, sin opinar sobre si tiene sentido. pydantic es el formulario digital: define, de antemano, qué forma exacta debe tener cada campo, y rechaza —con un mensaje preciso, campo por campo— cualquier documento que no la tenga.

Ejemplo trabajado: DataContract y ColumnContract

# contract.py
from typing import Literal
from pydantic import BaseModel, Field


class ColumnContract(BaseModel):
    name: str
    type: Literal["string", "float", "integer"]
    nullable: bool = True
    unique: bool = False
    minimum: float | None = None
    exclusive_minimum: float | None = None


class RowCountRange(BaseModel):
    min: int
    max: int


class SLAContract(BaseModel):
    freshness_hours: int
    row_count: RowCountRange


class DataContract(BaseModel):
    contract_version: str
    dataset: str
    owner: str
    description: str
    schema_: list[ColumnContract] = Field(alias="schema")
    sla: SLAContract
    on_violation: Literal["quarantine", "reject", "alert"]

Cuatro clases, cada una responsable de un nivel distinto del contrato. ColumnContract describe una sola entrada de la lista schema: — nota type: Literal["string", "float", "integer"]: no es un str cualquiera, es un tipo que solo acepta esos tres valores exactos, así que type: "boolean" (algo que este contrato nunca declaró soportar) fallaría de inmediato. minimum y exclusive_minimum son float | None = None — opcionales, porque no todas las columnas necesitan un límite numérico (order_id, de tipo string, nunca los usa). RowCountRange y SLAContract anidan un nivel más: sla.row_count.min y sla.row_count.max, reflejando exactamente la estructura anidada del YAML. Y DataContract junta todo, con un detalle de sintaxis que merece explicación aparte: schema_: list[ColumnContract] = Field(alias="schema").

Por qué schema_ con guion bajo, y no schema a secas

schema es una palabra reservada en el vocabulario de pydantic —el propio framework usa ese nombre internamente en versiones anteriores de su API—, así que nombrar el atributo schema a secas puede chocar con esa reserva interna. La solución de esta lección usa el mecanismo que pydantic ofrece exactamente para este caso: el atributo de Python se llama schema_ (con guion bajo, para no chocar con nada reservado), pero Field(alias="schema") le dice a pydantic que, al leer el diccionario de entrada, busque la llave schema (sin guion bajo, la que realmente existe en el YAML) y la asigne a ese atributo. model_validate(), el método que parsea el diccionario, respeta ese alias sin que tengas que renombrar nada en el archivo YAML mismo — orders_contract.yaml sigue diciendo schema:, tal como lo escribiste en la lección 3.

Ejemplo trabajado: parseando el contrato real

# contract.py -- continuacion, bloque ejecutable
if __name__ == "__main__":
    import yaml

    with open("orders_contract.yaml") as f:
        raw = yaml.safe_load(f)

    contract = DataContract.model_validate(raw)
    print(f"contract_version: {contract.contract_version}")
    print(f"dataset: {contract.dataset}")
    print(f"owner: {contract.owner}")
    print(f"columns: {[c.name for c in contract.schema_]}")
    for c in contract.schema_:
        print(f"  - {c.name}: type={c.type}, nullable={c.nullable}, unique={c.unique}, "
              f"minimum={c.minimum}, exclusive_minimum={c.exclusive_minimum}")
    print(f"sla.freshness_hours: {contract.sla.freshness_hours}")
    print(f"sla.row_count: min={contract.sla.row_count.min}, max={contract.sla.row_count.max}")
    print(f"on_violation: {contract.on_violation}")

Qué esperar. Al correr python3 contract.py, con orders_contract.yaml de la lección 3 en la misma carpeta, la salida es exactamente esta:

contract_version: 1.0.0
dataset: orders_s04
owner: kiosko-data-team
columns: ['order_id', 'unit_price', 'quantity']
  - order_id: type=string, nullable=False, unique=True, minimum=None, exclusive_minimum=None
  - unit_price: type=float, nullable=False, unique=False, minimum=0.0, exclusive_minimum=None
  - quantity: type=integer, nullable=False, unique=False, minimum=None, exclusive_minimum=0.0
sla.freshness_hours: 24
sla.row_count: min=5, max=20
on_violation: quarantine

DataContract.model_validate(raw) —el método de pydantic v2 que reemplaza al antiguo DataContract(**raw) cuando quieres pasar explícitamente un diccionario ya construido— recorre toda la estructura anidada, valida cada tipo, aplica los alias, y devuelve un objeto de Python con atributos tipados: contract.sla.row_count.min ya es un int real, no una cadena de texto que necesitarías convertir a mano. Compara esto contra raw["sla"]["row_count"]["min"] de la lección 3 — la misma información, pero ahora con la garantía de que, si esta línea se ejecutó sin lanzar ninguna excepción, el contrato completo respeta la forma que DataContract exige.

Ejemplo trabajado: rompiendo el contrato a propósito

Un contrato solo es útil si falla de forma ruidosa cuando alguien lo rompe — no si acepta en silencio cualquier cosa. Este archivo, guardado como orders_contract_broken.yaml, tiene dos errores deliberados: le falta el campo owner, y on_violation trae un valor que la clase nunca declaró aceptar:

# orders_contract_broken.yaml
contract_version: "1.0.0"
dataset: orders_s04
description: >
  Version rota a proposito: le falta 'owner' y 'on_violation' trae un
  valor que el contrato no reconoce.
schema:
  - name: order_id
    type: string
    nullable: false
    unique: true
  - name: unit_price
    type: float
    nullable: false
    minimum: 0
  - name: quantity
    type: integer
    nullable: false
    exclusive_minimum: 0
sla:
  freshness_hours: 24
  row_count:
    min: 5
    max: 20
on_violation: delete_silently

# parse_broken.py
import yaml
from pydantic import ValidationError
from contract import DataContract

with open("orders_contract_broken.yaml") as f:
    raw = yaml.safe_load(f)

try:
    contract = DataContract.model_validate(raw)
    print("El contrato roto paso la validacion (no esperado).")
except ValidationError as exc:
    print(f"ValidationError: {exc.error_count()} errores\n")
    print(exc)

Qué esperar. Al correr python3 parse_broken.py, la salida es exactamente esta:

ValidationError: 2 errores

2 validation errors for DataContract
owner
  Field required [type=missing, input_value={'contract_version': '1.0...ion': 'delete_silently'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.13/v/missing
on_violation
  Input should be 'quarantine', 'reject' or 'alert' [type=literal_error, input_value='delete_silently', input_type=str]
    For further information visit https://errors.pydantic.dev/2.13/v/literal_error

Lee este resultado con el mismo cuidado que ya entrenaste con SchemaErrors.failure_cases de Pandera en el módulo 2 — es la misma filosofía de diseño, aplicada a un problema distinto. ValidationError no se detiene en el primer error que encuentra (owner faltante) y listo; acumula los dos, exactamente como lazy=True de Pandera acumulaba las tres fallas de S04 en un solo reporte, en vez de detenerse en la primera. Cada error nombra el campo exacto (owner, on_violation), el tipo de problema (missing, literal_error), y en el segundo caso, hasta te dice cuáles son los únicos valores que sí acepta ('quarantine', 'reject' or 'alert') — información suficiente para corregir el contrato sin adivinar.

Diagrama: dos caminos posibles para el mismo YAML

flowchart TD
    A["orders_contract.yaml"] --> B["yaml.safe_load()"]
    B --> C{"DataContract.model_validate(raw)"}
    C -->|"estructura correcta"| D["objeto DataContract\ntipado y validado"]
    C -->|"falta un campo,\no un valor invalido"| E["ValidationError\ncon los campos exactos"]

Profundización: por qué esto importa más para un contrato que para cualquier otro archivo de configuración

Vale la pena preguntarse por qué esta guía dedica una lección completa a la validación de tipos de un archivo YAML, cuando otras configuraciones de Kiosko en guías anteriores del ecosistema nunca necesitaron algo así. La respuesta está en lo que un contrato hace con su propio contenido: contract_to_pandera_schema(), en la lección 5, va a leer contract.schema_[i].minimum y pasárselo directo a pa.Check.ge(...) — si ese valor resultara ser, por accidente, la cadena de texto "cero" en vez del número 0, el error no aparecería al leer el contrato, sino mucho más tarde, dentro de la lógica de Pandera, con un mensaje mucho menos claro sobre cuál fue la causa real. Validar el contrato con pydantic, en el momento en que se lee, mueve el punto de falla lo más cerca posible del error real — el mismo principio de "fallar rápido y con un mensaje claro" que ya aplicó lazy=True de Pandera en el módulo 2, ahora aplicado a la capa que hay antes de Pandera.

Errores comunes

Escribir schema como nombre de atributo, sin el alias. Qué pasa: alguien, copiando la estructura de esta lección, define schema: list[ColumnContract] directamente, sin Field(alias="schema") ni el guion bajo, y encuentra un comportamiento confuso o un error al definir la clase. Por qué pasa: schema parece, a primera vista, un nombre de atributo perfectamente razonable — no hay ninguna razón obvia para evitarlo, hasta que se choca con el uso interno de pydantic. Cómo detectarlo: si tu clase DataContract no se comporta como esperas al acceder a .schema, o pydantic se queja de un nombre reservado, revisa si nombraste el atributo schema a secas. Cómo corregirlo: sigue el patrón exacto de esta lección — nombra el atributo schema_ (con guion bajo) y usa Field(alias="schema") para que la lectura del YAML (que sí dice schema:, sin guion bajo) siga funcionando sin cambiar el archivo.

Usar DataContract(**raw) y asumir que se comporta exactamente igual que DataContract.model_validate(raw) en cualquier caso. Qué pasa: alguien, acostumbrado a construir objetos de Python desempaquetando un diccionario con **, usa DataContract(**raw) en vez de model_validate(raw) — y para un diccionario bien formado, como el raw de esta lección, el resultado es, de hecho, idéntico: mismos atributos, mismos valores, incluido el alias de schema_ resuelto correctamente. El problema aparece en el caso límite, no en el caso feliz. Por qué pasa: **raw es sintaxis nativa de Python —desempaquetar un diccionario como argumentos de palabra clave—, evaluada por el intérprete antes de que pydantic tenga oportunidad de intervenir. Si raw no es, por algún motivo, un diccionario válido (por ejemplo, si yaml.safe_load() devolviera None porque el archivo está vacío), **raw falla con un TypeError de Python puro, no con un ValidationError de pydantic. Cómo detectarlo: compara los dos mensajes de error para el mismo caso — DataContract(**None) lanza TypeError: contract.DataContract() argument after ** must be a mapping, not NoneType; DataContract.model_validate(None) lanza pydantic.ValidationError: Input should be a valid dictionary or instance of DataContract, un error mucho más claro y, sobre todo, capturable con el mismo except ValidationError que ya usa el resto de esta lección. Cómo corregirlo: usa model_validate() siempre que el origen del diccionario no esté garantizado —como un archivo YAML que alguien más podría dejar vacío por accidente—, para que cualquier problema, incluido un archivo vacío o mal leído, termine como el mismo tipo de error controlado que ya sabes manejar.

Capturar Exception en vez de ValidationError específicamente. Qué pasa: alguien escribe except Exception as exc: en vez de except ValidationError as exc: al parsear un contrato, y termina también capturando (y ocultando) errores completamente distintos —un archivo que no existe, un YAML con sintaxis rota— como si fueran el mismo tipo de problema. Por qué pasa: except Exception se siente como una forma "segura" de no dejar que nada se rompa el programa, pero esconde información valiosa sobre qué tipo de error ocurrió. Cómo detectarlo: si tu manejo de errores no puede distinguir "el YAML tiene mala sintaxis" de "el contrato no respeta la estructura de DataContract" de "el archivo no existe", tu except es demasiado amplio. Cómo corregirlo: captura ValidationError específicamente para errores de estructura del contrato (como hace esta lección), y deja que otros tipos de error —FileNotFoundError, yaml.YAMLError— se propaguen o se manejen por separado, con su propio mensaje específico.

Ejercicios

Ejercicio 1 — Rompe el contrato de una tercera forma: un tipo de columna inválido. Modifica orders_contract_broken.yaml (o créalo de nuevo) cambiando type: float de unit_price por type: number (un valor que Literal["string", "float", "integer"] no acepta). Corre parse_broken.py de nuevo y confirma el mensaje de error exacto.

schema.1.type
  Input should be 'string', 'float' or 'integer' [type=literal_error, input_value='number', input_type=str]

Ejercicio 2 — Confirma que un contrato con solo campos opcionales de más (extra) sigue pasando. Agrega un campo nuevo, no declarado en ninguna clase —por ejemplo, extra_note: "campo no declarado"— al nivel superior de orders_contract.yaml, y confirma si DataContract.model_validate(raw) lo acepta o lo rechaza.

Ejercicio 3 — Argumenta por qué ValidationError acumula todos los errores en vez de detenerse en el primero. En 2-3 frases, conecta este comportamiento de pydantic con el mismo patrón que ya viste en lazy=True de Pandera, en el módulo 2.

Resumen y siguiente paso

En esta lección construiste DataContract y ColumnContract, dos clases de pydantic que le dan al diccionario suelto de la lección 3 una garantía real de estructura: tipos correctos, campos obligatorios presentes, valores restringidos a los que el contrato declara aceptar. Parseaste el contrato real de S04 con model_validate(), y confirmaste, con un YAML roto a propósito en dos lugares distintos, que pydantic reporta ambos errores juntos, con el campo exacto y el motivo exacto de cada uno.

Antes de avanzar deberías poder: explicar por qué schema_ usa un alias en vez del nombre schema directo; y reproducir el ValidationError de dos errores corriendo parse_broken.py tú mismo.

Tienes un objeto DataContract completamente validado, con atributos tipados y accesibles sin necesitar volver a tocar el diccionario crudo. La lección 5 —el momento central de este módulo— toma ese objeto y lo convierte de vuelta en un esquema de Pandera ejecutable, confirmando que atrapa exactamente las mismas filas que OrdersSchema del módulo 2.

Recursos

• Pydantic — documentación oficial, BaseModel (la clase base de DataContract/ColumnContract, y el mecanismo de validación de campos). docs.pydantic.dev/latest/concepts/models. En inglés.
• Pydantic — documentación oficial, model_validate (el método que parsea un diccionario ya construido, usado en toda esta lección). docs.pydantic.dev/latest/concepts/models/#validating-data. En inglés.
• Pydantic — documentación oficial, alias de campos (Field(alias=...), la solución al conflicto de nombre con schema). docs.pydantic.dev/latest/concepts/alias. En inglés.
• Pydantic — documentación oficial, manejo de errores (ValidationError, la estructura de errores acumulados). docs.pydantic.dev/latest/errors/errors. En inglés.
• Módulo 2, lección 7, de esta misma guía — fuente del patrón lazy=True / reporte de errores acumulados, el mismo principio de diseño que esta lección aplica con pydantic. src/guides/data-reliability-and-governance-guide/workbook/module-02-declarative-data-quality-tests-with-pandera/es/07-validity-checks-and-reading-failure-cases.md. En español.
• DISEÑO de esta guía. src/guides/data-reliability-and-governance-guide/DISENO.md. En español.