Producer Onboarding Guide¶

from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
import json

# Конфигурация
schema_registry_conf = {'url': 'http://schema-registry:8081'}
kafka_conf = {
    'bootstrap.servers': 'kafka:9092',
    'client.id': 'your-service-name'
}

# Загрузка схемы из контракта
with open('path/to/contract.yaml') as f:
    contract = yaml.safe_load(f)

# Создание producer
schema_registry_client = SchemaRegistryClient(schema_registry_conf)
avro_serializer = AvroSerializer(
    schema_registry_client,
    schema_str=json.dumps(contract_to_avro_schema(contract))
)

producer = Producer(kafka_conf)

def send_data(record: dict):
    """Отправка данных согласно контракту"""
    topic = f"{contract['metadata']['namespace']}.{contract['metadata']['name']}.raw"

    producer.produce(
        topic=topic,
        value=avro_serializer(record, SerializationContext(topic, MessageField.VALUE)),
        callback=delivery_report
    )
    producer.flush()

import requests
import io
import avro.schema
import avro.io

API_GATEWAY_URL = "https://data-gateway.company.ru"

def serialize_to_avro(record: dict, schema_json: str) -> bytes:
    """Сериализация записи в Avro"""
    schema = avro.schema.parse(schema_json)
    writer = avro.io.DatumWriter(schema)
    bytes_writer = io.BytesIO()
    encoder = avro.io.BinaryEncoder(bytes_writer)
    writer.write(record, encoder)
    return bytes_writer.getvalue()

def send_data(record: dict, topic: str, contract_version: str, schema_json: str):
    """
    Отправка через API Gateway.

    ⚠️ Важно:
    - API Gateway НЕ валидирует данные
    - Данные принимаются "as-is" в Kafka .raw
    - Валидация происходит потом в Quality Validator
    - Поэтому API Gateway быстрый и не блокирует издателя
    """

    # Сериализация в Avro
    avro_payload = serialize_to_avro(record, schema_json)

    response = requests.post(
        f"{API_GATEWAY_URL}/api/1.0/{topic}",
        data=avro_payload,  # Avro bytes
        headers={
            "Content-Type": "application/avro",
            "version": contract_version,
            "batch": "1",  # количество записей в батче
        },
        cert=("/secure/certs/{topic}.crt", "/secure/certs/{topic}.key"),
        verify="/secure/certs/ca.crt"
    )

    if response.status_code != 201:
        raise Exception(f"Failed to send data: {response.status_code} {response.text}")

    return response

import pandas as pd
from datetime import datetime

def export_batch(df: pd.DataFrame, domain: str, entity: str):
    """Экспорт batch данных"""
    timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
    filename = f"{domain}_{entity}_{timestamp}.parquet"

    # Опциональная client-side валидация (для раннего обнаружения проблем)
    # Основная валидация всё равно происходит в Quality Validator
    # validate_against_contract(df, f"contracts/domains/{domain}/{entity}/contract.yaml")

    # Сохранение
    df.to_parquet(f"/data/outbox/{filename}")