Перейти к содержанию

Валидатор качества

Kruma Quality Validator — Python-модуль валидации данных, который проверяет каждую запись на соответствие правилам из контракта и маршрутизирует их в production или Dead Letter Queue.

Версия: 1.0.1 | Python: >= 3.11 | Лицензия: Proprietary

Ключевые возможности

  • 10 типов правил

    not_null, unique, range, regex, enum, freshness, format, custom, sql, reference

  • Kafka-интеграция

    Автоматическая маршрутизация записей: .raw → .prod / .dlq

  • Безопасный sandbox

    Custom-выражения выполняются в изолированном окружении без доступа к ОС

  • Prometheus-метрики

    Pass rate, DLQ rate, латентность, ошибки по правилам

Как это работает

graph LR
    A[Kafka .raw] --> B[Quality<br/>Validator]
    B -->|Валидный| C[Kafka .prod]
    B -->|Невалидный| D[Kafka .dlq]
    D --> E[Алерты]
    C --> F[Потребители<br/>BI, ML, DWH]

    style B fill:#FF9800,color:#fff
    style D fill:#F44336,color:#fff
    style C fill:#4CAF50,color:#fff

Принцип работы

  1. Consumer читает сырые данные из топика {namespace}.{entity}.raw
  2. RuleEngine применяет все правила из quality_rules.yml
  3. Записи с ошибками severity error → DLQ (с полным контекстом ошибок)
  4. Валидные записи → Prod (для потребителей)
  5. Warnings логируются, но не блокируют запись
  6. Метрики экспортируются в Prometheus

Быстрый старт

Установка

pip install kruma-validator

Валидация записи

from kruma_validator import QualityValidator

# Создать валидатор из контракта
validator = QualityValidator.from_contract_path(
    "contracts/domains/sales/orders"
)

# Валидировать запись
record = {
    "order_id": "ord_abc123def456",
    "customer_id": "cust_xyz789",
    "total_amount": 150.50,
    "status": "confirmed",
    "items": [{"sku": "SKU001", "quantity": 2, "price": 75.25}],
    "items_count": 1,
    "created_at": "2026-01-23T10:30:00Z"
}

result = validator.validate(record)

if result.is_valid:
    print("Запись прошла валидацию")
else:
    for error in result.errors:
        print(f"Ошибка: {error.rule_name} — {error.message}")

Пакетная валидация

batch_result = validator.validate_batch(records)

print(f"Всего:      {batch_result.total_count}")
print(f"Валидных:   {batch_result.valid_count}")
print(f"Невалидных: {batch_result.invalid_count}")
print(f"Pass rate:  {batch_result.valid_count / batch_result.total_count:.1%}")

# Получить записи для Kafka
prod_records = batch_result.get_prod_records()  # → .prod
dlq_records = batch_result.get_dlq_records()    # → .dlq

Разделы документации

Изменения в v1.0.1

Breaking Change: именование топиков

С версии 1.0.1 используется единый формат с точками:

  • sales.orders.raw (без изменений)
  • sales.orders.prod (было: sales.orders_prod)
  • sales.orders.dlq (было: sales.orders_dlq)

Подробности — в истории изменений.