"""Avro Deserializer."""

from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer as ConfluentAvroDeserializer
from confluent_kafka.serialization import SerializationContext

from kafka_utils.consumer.deserializer.base import BaseDeserializer
from kafka_utils.consumer.deserializer.helpers import read_schema_id_from_message


class AvroDeserializer(BaseDeserializer):
    """Avro Deserializer."""

    def __init__(self, schema_registry_url, **kwargs):
        """Init."""
        super().__init__()
        self.schema_registry_client = SchemaRegistryClient({'url': schema_registry_url})
        self.deserializer = ConfluentAvroDeserializer(self.schema_registry_client, **kwargs)

    def deserialize(self, value, topic, context):
        """Deserialize."""
        serialization_context = SerializationContext(topic, context)
        return self.deserializer(value, serialization_context)

    @staticmethod
    def get_schema_id(value):
        """Get the schema ID from the serialized message."""
        return read_schema_id_from_message(value)