"""JSON Schema serializer helper.""" from confluent_kafka.schema_registry import SchemaRegistryClient from confluent_kafka.schema_registry.json_schema import JSONSerializer from kafka_utils.producer.helpers import get_context def build_json_schema_serializer( schema_registry_url, topic_name, schema_version, context='value', config=None ): """Build a JSON Schema serializer.""" schema_registry_client = SchemaRegistryClient({'url': schema_registry_url}) serializer_context = get_context(context) schema_subject = f'{topic_name}-{serializer_context}' schema = schema_registry_client.get_version(schema_subject, schema_version) schema_str = schema.schema.schema_str serializer_config = {'auto.register.schemas': False} if config: serializer_config.update(**config) return JSONSerializer(schema_str, schema_registry_client, conf=serializer_config)