Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/confluent_kafka/deserializing_consumer.py: 20%

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3# 

4# Copyright 2020 Confluent Inc. 

5# 

6# Licensed under the Apache License, Version 2.0 (the "License"); 

7# you may not use this file except in compliance with the License. 

8# You may obtain a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, software 

13# distributed under the License is distributed on an "AS IS" BASIS, 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

15# See the License for the specific language governing permissions and 

16# limitations under the License. 

17# 

18 

19from confluent_kafka.cimpl import Consumer as _ConsumerImpl 

20from .error import (ConsumeError, 

21 KeyDeserializationError, 

22 ValueDeserializationError) 

23from .serialization import (SerializationContext, 

24 MessageField) 

25 

26 

27class DeserializingConsumer(_ConsumerImpl): 

28 """ 

29 A high level Kafka consumer with deserialization capabilities. 

30 

31 `This class is experimental and likely to be removed, or subject to incompatible API 

32 changes in future versions of the library. To avoid breaking changes on upgrading, we 

33 recommend using deserializers directly.` 

34 

35 Derived from the :py:class:`Consumer` class, overriding the :py:func:`Consumer.poll` 

36 method to add deserialization capabilities. 

37 

38 Additional configuration properties: 

39 

40 +-------------------------+---------------------+-----------------------------------------------------+ 

41 | Property Name | Type | Description | 

42 +=========================+=====================+=====================================================+ 

43 | | | Callable(bytes, SerializationContext) -> obj | 

44 | ``key.deserializer`` | callable | | 

45 | | | Deserializer used for message keys. | 

46 +-------------------------+---------------------+-----------------------------------------------------+ 

47 | | | Callable(bytes, SerializationContext) -> obj | 

48 | ``value.deserializer`` | callable | | 

49 | | | Deserializer used for message values. | 

50 +-------------------------+---------------------+-----------------------------------------------------+ 

51 

52 Deserializers for string, integer and double (:py:class:`StringDeserializer`, :py:class:`IntegerDeserializer` 

53 and :py:class:`DoubleDeserializer`) are supplied out-of-the-box in the ``confluent_kafka.serialization`` 

54 namespace. 

55 

56 Deserializers for Protobuf, JSON Schema and Avro (:py:class:`ProtobufDeserializer`, :py:class:`JSONDeserializer` 

57 and :py:class:`AvroDeserializer`) with Confluent Schema Registry integration are supplied out-of-the-box 

58 in the ``confluent_kafka.schema_registry`` namespace. 

59 

60 See Also: 

61 - The :ref:`Configuration Guide <pythonclient_configuration>` for in depth information on how to configure the client. 

62 - `CONFIGURATION.md <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md>`_ for a comprehensive set of configuration properties. 

63 - `STATISTICS.md <https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md>`_ for detailed information on the statistics provided by stats_cb 

64 - The :py:class:`Consumer` class for inherited methods. 

65 

66 Args: 

67 conf (dict): DeserializingConsumer configuration. 

68 

69 Raises: 

70 ValueError: if configuration validation fails 

71 """ # noqa: E501 

72 

73 def __init__(self, conf): 

74 conf_copy = conf.copy() 

75 self._key_deserializer = conf_copy.pop('key.deserializer', None) 

76 self._value_deserializer = conf_copy.pop('value.deserializer', None) 

77 

78 super(DeserializingConsumer, self).__init__(conf_copy) 

79 

80 def poll(self, timeout=-1): 

81 """ 

82 Consume messages and calls callbacks. 

83 

84 Args: 

85 timeout (float): Maximum time to block waiting for message(Seconds). 

86 

87 Returns: 

88 :py:class:`Message` or None on timeout 

89 

90 Raises: 

91 KeyDeserializationError: If an error occurs during key deserialization. 

92 

93 ValueDeserializationError: If an error occurs during value deserialization. 

94 

95 ConsumeError: If an error was encountered while polling. 

96 """ 

97 

98 msg = super(DeserializingConsumer, self).poll(timeout) 

99 

100 if msg is None: 

101 return None 

102 

103 if msg.error() is not None: 

104 raise ConsumeError(msg.error(), kafka_message=msg) 

105 

106 ctx = SerializationContext(msg.topic(), MessageField.VALUE, msg.headers()) 

107 value = msg.value() 

108 if self._value_deserializer is not None: 

109 try: 

110 value = self._value_deserializer(value, ctx) 

111 except Exception as se: 

112 raise ValueDeserializationError(exception=se, kafka_message=msg) 

113 

114 key = msg.key() 

115 ctx.field = MessageField.KEY 

116 if self._key_deserializer is not None: 

117 try: 

118 key = self._key_deserializer(key, ctx) 

119 except Exception as se: 

120 raise KeyDeserializationError(exception=se, kafka_message=msg) 

121 

122 msg.set_key(key) 

123 msg.set_value(value) 

124 return msg 

125 

126 def consume(self, num_messages=1, timeout=-1): 

127 """ 

128 :py:func:`Consumer.consume` not implemented, use 

129 :py:func:`DeserializingConsumer.poll` instead 

130 """ 

131 

132 raise NotImplementedError