#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2020 Confluent Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import struct as _struct
from enum import Enum
from typing import Any, Callable, Dict, List, Optional, Tuple
from confluent_kafka._types import HeadersType
from confluent_kafka.error import KafkaException
__all__ = [
'Deserializer',
'DeserializerBuilder',
'IntegerDeserializer',
'IntegerSerializer',
'DoubleDeserializer',
'DoubleSerializer',
'StringDeserializer',
'StringSerializer',
'MessageField',
'SerializationContext',
'SerializationError',
'Serializer',
'SerializerBuilder',
]
[docs]
class MessageField(str, Enum):
"""
Enum like object for identifying Message fields.
Attributes:
KEY (str): Message key
VALUE (str): Message value
"""
NONE = 'none'
KEY = 'key'
VALUE = 'value'
def __str__(self) -> str:
return str(self.value)
[docs]
class SerializationContext(object):
"""
SerializationContext provides additional context to the
serializer/deserializer about the data it's serializing/deserializing.
Args:
topic (str): Topic data is being produce to or consumed from.
field (MessageField): Describes what part of the message is
being serialized.
headers (list): List of message header tuples. Defaults to None.
"""
def __init__(self, topic: str, field: MessageField, headers: Optional[HeadersType] = None) -> None:
self.topic = topic
self.field = field
self.headers = headers
[docs]
class SerializationError(KafkaException):
"""Generic error from serializer package"""
pass
[docs]
class Serializer(object):
"""
Extensible class from which all Serializer implementations derive.
Serializers instruct Kafka clients on how to convert Python objects
to bytes.
See built-in implementations, listed below, for an example of how to
extend this class.
Note:
This class is not directly instantiable. The derived classes must be
used instead.
The following implementations are provided by this module.
Note:
Unless noted elsewhere all numeric types are signed and serialization
is big-endian.
.. list-table::
:header-rows: 1
* - Name
- Type
- Binary Format
* - DoubleSerializer
- float
- IEEE 764 binary64
* - IntegerSerializer
- int
- int32
* - StringSerializer
- unicode
- unicode(encoding)
"""
__slots__: List[str] = []
[docs]
def __call__(self, obj: Any, ctx: Optional[SerializationContext] = None) -> Optional[bytes]:
"""
Converts obj to bytes.
Args:
obj (object): object to be serialized
ctx (SerializationContext): Metadata pertaining to the serialization
operation
Raises:
SerializerError if an error occurs during serialization
Returns:
bytes if obj is not None, otherwise None
"""
raise NotImplementedError
[docs]
def set_cluster_id_resolver(self, resolver: Callable[[], Any]) -> None:
"""
Supply a callable returning the id of the Kafka cluster the client is
connected to.
*This API is experimental and subject to incompatible changes in future versions of the library.*
:py:class:`SerializingProducer` calls this once, right after it has
created the underlying client. The resolver may block for as long as
the client waits for broker metadata, so a serializer must not invoke
it during this call, only later when it actually needs the id;
concurrent invocations share a single wait. A serializer that
resolves subjects through the Schema Registry
*associated* subject name strategy keeps it and invokes it on the
first subject lookup, unless the cluster id was configured explicitly.
The default implementation does nothing.
Args:
resolver (callable): Callable returning the cluster id.
"""
pass
[docs]
def close(self) -> None:
"""
Release the resources this serializer created for itself.
*This API is experimental and subject to incompatible changes in future versions of the library.*
Called by :py:class:`SerializingProducer` when it is closed, for the
serializers it built from a builder. Resources handed to the
serializer by the application are left alone. The default
implementation does nothing.
"""
pass
[docs]
class Deserializer(object):
"""
Extensible class from which all Deserializer implementations derive.
Deserializers instruct Kafka clients on how to convert bytes to objects.
See built-in implementations, listed below, for an example of how to
extend this class.
Note:
This class is not directly instantiable. The derived classes must be
used instead.
The following implementations are provided by this module.
Note:
Unless noted elsewhere all numeric types are signed and
serialization is big-endian.
.. list-table::
:header-rows: 1
* - Name
- Type
- Binary Format
* - DoubleDeserializer
- float
- IEEE 764 binary64
* - IntegerDeserializer
- int
- int32
* - StringDeserializer
- unicode
- unicode(encoding)
"""
__slots__: List[str] = []
[docs]
def __call__(self, value: Optional[bytes], ctx: Optional[SerializationContext] = None) -> Any:
"""
Convert bytes to object
Args:
value (bytes): bytes to be deserialized
ctx (SerializationContext): Metadata pertaining to the serialization
operation
Raises:
SerializerError if an error occurs during deserialization
Returns:
object if data is not None, otherwise None
"""
raise NotImplementedError
[docs]
def set_cluster_id_resolver(self, resolver: Callable[[], Any]) -> None:
"""
Supply a callable returning the id of the Kafka cluster the client is
connected to.
*This API is experimental and subject to incompatible changes in future versions of the library.*
See :py:func:`Serializer.set_cluster_id_resolver`; the resolver is
supplied by :py:class:`DeserializingConsumer`. The default
implementation does nothing.
Args:
resolver (callable): Callable returning the cluster id.
"""
pass
[docs]
def close(self) -> None:
"""
Release the resources this deserializer created for itself.
*This API is experimental and subject to incompatible changes in future versions of the library.*
See :py:func:`Serializer.close`; called by
:py:class:`DeserializingConsumer`. The default implementation does
nothing.
"""
pass
[docs]
class SerializerBuilder(object):
"""
Extensible class from which all Serializer builders derive.
*This class is experimental and subject to incompatible API changes in future versions of the library.*
A builder defers construction of a :py:class:`Serializer` until the client
is created, which lets it take the client's own configuration into account
and lets the client own the serializer's lifecycle. Pass one to
:py:class:`SerializingProducer` through the ``key.serializer.builder`` or
``value.serializer.builder`` configuration property instead of constructing
a serializer yourself.
Builders for Protobuf, JSON Schema and Avro with Confluent Schema Registry
integration are supplied out-of-the-box in the
``confluent_kafka.schema_registry`` namespace.
Note:
This class is not directly instantiable. The derived classes must be
used instead.
"""
__slots__: List[str] = []
[docs]
def build(self, conf: Dict[str, Any], is_key: bool) -> Tuple[Serializer, Dict[str, Any]]:
"""
Build the serializer.
Args:
conf (dict): Client configuration. A builder may consume properties
of its own from it; whatever it does not consume must be
returned so the client can be configured with it.
is_key (bool): True when building the serializer for message keys,
False for message values.
Returns:
tuple: The built :py:class:`Serializer` and the configuration left
over for the client.
"""
raise NotImplementedError
[docs]
class DeserializerBuilder(object):
"""
Extensible class from which all Deserializer builders derive.
*This class is experimental and subject to incompatible API changes in future versions of the library.*
The deserializing counterpart of :py:class:`SerializerBuilder`. Pass one to
:py:class:`DeserializingConsumer` through the ``key.deserializer.builder``
or ``value.deserializer.builder`` configuration property.
Note:
This class is not directly instantiable. The derived classes must be
used instead.
"""
__slots__: List[str] = []
[docs]
def build(self, conf: Dict[str, Any], is_key: bool) -> Tuple[Deserializer, Dict[str, Any]]:
"""
Build the deserializer.
Args:
conf (dict): Client configuration. A builder may consume properties
of its own from it; whatever it does not consume must be
returned so the client can be configured with it.
is_key (bool): True when building the deserializer for message keys,
False for message values.
Returns:
tuple: The built :py:class:`Deserializer` and the configuration left
over for the client.
"""
raise NotImplementedError
[docs]
class DoubleSerializer(Serializer):
"""
Serializes float to IEEE 764 binary64.
See Also:
`DoubleSerializer Javadoc <https://docs.confluent.io/current/clients/javadocs/org/apache/kafka/common/serialization/DoubleSerializer.html>`_
""" # noqa: E501
[docs]
def __call__(self, obj: Optional[float], ctx: Optional[SerializationContext] = None) -> Optional[bytes]:
"""
Args:
obj (object): object to be serialized
ctx (SerializationContext): Metadata pertaining to the serialization
operation
Note:
None objects are represented as Kafka Null.
Raises:
SerializerError if an error occurs during serialization.
Returns:
IEEE 764 binary64 bytes if obj is not None, otherwise None
"""
if obj is None:
return None
try:
return _struct.pack('>d', obj)
except _struct.error as e:
raise SerializationError(str(e))
[docs]
class DoubleDeserializer(Deserializer):
"""
Deserializes float to IEEE 764 binary64.
See Also:
`DoubleDeserializer Javadoc <https://docs.confluent.io/current/clients/javadocs/org/apache/kafka/common/serialization/DoubleDeserializer.html>`_
""" # noqa: E501
[docs]
def __call__(self, value: Optional[bytes], ctx: Optional[SerializationContext] = None) -> Optional[float]:
"""
Deserializes float from IEEE 764 binary64 bytes.
Args:
value (bytes): bytes to be deserialized
ctx (SerializationContext): Metadata pertaining to the serialization
operation
Raises:
SerializerError if an error occurs during deserialization.
Returns:
float if data is not None, otherwise None
"""
if value is None:
return None
try:
return _struct.unpack('>d', value)[0]
except _struct.error as e:
raise SerializationError(str(e))
[docs]
class IntegerSerializer(Serializer):
"""
Serializes int to int32 bytes.
See Also:
`IntegerSerializer Javadoc <https://docs.confluent.io/current/clients/javadocs/org/apache/kafka/common/serialization/IntegerSerializer.html>`_
""" # noqa: E501
[docs]
def __call__(self, obj: Optional[int], ctx: Optional[SerializationContext] = None) -> Optional[bytes]:
"""
Serializes int as int32 bytes.
Args:
obj (object): object to be serialized
ctx (SerializationContext): Metadata pertaining to the serialization
operation
Note:
None objects are represented as Kafka Null.
Raises:
SerializerError if an error occurs during serialization
Returns:
int32 bytes if obj is not None, else None
"""
if obj is None:
return None
try:
return _struct.pack('>i', obj)
except _struct.error as e:
raise SerializationError(str(e))
[docs]
class IntegerDeserializer(Deserializer):
"""
Deserializes int to int32 bytes.
See Also:
`IntegerDeserializer Javadoc <https://docs.confluent.io/current/clients/javadocs/org/apache/kafka/common/serialization/IntegerDeserializer.html>`_
""" # noqa: E501
[docs]
def __call__(self, value: Optional[bytes], ctx: Optional[SerializationContext] = None) -> Optional[int]:
"""
Deserializes int from int32 bytes.
Args:
value (bytes): bytes to be deserialized
ctx (SerializationContext): Metadata pertaining to the serialization
operation
Raises:
SerializerError if an error occurs during deserialization.
Returns:
int if data is not None, otherwise None
"""
if value is None:
return None
try:
return _struct.unpack('>i', value)[0]
except _struct.error as e:
raise SerializationError(str(e))
[docs]
class StringSerializer(Serializer):
"""
Serializes unicode to bytes per the configured codec. Defaults to ``utf_8``.
Note:
None objects are represented as Kafka Null.
Args:
codec (str, optional): encoding scheme. Defaults to utf_8
See Also:
`Supported encodings <https://docs.python.org/3/library/codecs.html#standard-encodings>`_
`StringSerializer Javadoc <https://docs.confluent.io/current/clients/javadocs/org/apache/kafka/common/serialization/StringSerializer.html>`_
""" # noqa: E501
def __init__(self, codec: str = 'utf_8') -> None:
self.codec = codec
[docs]
def __call__(self, obj: Optional[str], ctx: Optional[SerializationContext] = None) -> Optional[bytes]:
"""
Serializes a str(py2:unicode) to bytes.
Compatibility Note:
Python 2 str objects must be converted to unicode objects.
Python 3 all str objects are already unicode objects.
Args:
obj (object): object to be serialized
ctx (SerializationContext): Metadata pertaining to the serialization
operation
Raises:
SerializerError if an error occurs during serialization.
Returns:
serialized bytes if obj is not None, otherwise None
"""
if obj is None:
return None
try:
return obj.encode(self.codec)
except UnicodeError as e:
raise SerializationError(str(e))
[docs]
class StringDeserializer(Deserializer):
"""
Deserializes a str(py2:unicode) from bytes.
Args:
codec (str, optional): encoding scheme. Defaults to utf_8
See Also:
`Supported encodings <https://docs.python.org/3/library/codecs.html#standard-encodings>`_
`StringDeserializer Javadoc <https://docs.confluent.io/current/clients/javadocs/org/apache/kafka/common/serialization/StringDeserializer.html>`_
""" # noqa: E501
def __init__(self, codec: str = 'utf_8') -> None:
self.codec = codec
[docs]
def __call__(self, value: Optional[bytes], ctx: Optional[SerializationContext] = None) -> Optional[str]:
"""
Serializes unicode to bytes per the configured codec. Defaults to ``utf_8``.
Compatibility Note:
Python 2 str objects must be converted to unicode objects by the
application prior to using this serializer.
Python 3 all str objects are already unicode objects.
Args:
value (bytes): bytes to be deserialized
ctx (SerializationContext): Metadata pertaining to the serialization
operation
Raises:
SerializerError if an error occurs during deserialization.
Returns:
unicode if data is not None, otherwise None
"""
if value is None:
return None
try:
return value.decode(self.codec)
except UnicodeError as e:
raise SerializationError(str(e))