Place SerializationContext first in Serializer/SubjectNameStrategy APIs

Author: rnpridgeon

confluent_kafka/deserializing_consumer.py

@@ -30,6 +30,7 @@ class DeserializingConsumer(_ConsumerImpl):
     capabilities.
 
     Note:
+
         The DeserializingConsumer is an experimental API and subject to change.
 
     .. versionadded:: 1.4.0
@@ -38,6 +39,7 @@ class DeserializingConsumer(_ConsumerImpl):
         DeserializingConsumer on how to convert the message payload bytes to objects.
 
     Note:
+
         All configured callbacks are served from the application queue upon
         calling :py:func:`DeserializingConsumer.poll`
 
@@ -52,11 +54,11 @@ class DeserializingConsumer(_ConsumerImpl):
     | group.id*          | str             | All clients sharing the same group.id belong to the |
     |                    |                 | same group.                                         |
     +--------------------+-----------------+-----------------------------------------------------+
-    |                    |                 | Callable(bytes, SerializationContext) -> obj        |
+    |                    |                 | Callable(SerializationContext, bytes) -> obj        |
     | key.deserializer   | callable        |                                                     |
     |                    |                 | Serializer used for message keys.                   |
     +--------------------+-----------------+-----------------------------------------------------+
-    |                    |                 | Callable(bytes, SerializationContext) -> obj        |
+    |                    |                 | Callable(SerializationContext, bytes) -> obj        |
     | value.deserializer | callable        |                                                     |
     |                    |                 | Serializer used for message values.                 |
     +--------------------+-----------------+-----------------------------------------------------+
@@ -72,15 +74,16 @@ class DeserializingConsumer(_ConsumerImpl):
     |                    |                 |                                                     |
     |                    |                 | Callback for statistics. This callback is           |
     | stats_cb           | callable        | added to the application queue every                |
-    |                    |                 |``statistics.interval.ms`` (configured separately).  |
+    |                    |                 | ``statistics.interval.ms`` (configured separately). |
     |                    |                 | The function argument is a JSON formatted str       |
     |                    |                 | containing statistics data.                         |
     +--------------------+-----------------+-----------------------------------------------------+
     |                    |                 | Callable(ThrottleEvent)                             |
     | throttle_cb        | callable        |                                                     |
     |                    |                 | Callback for throttled request reporting.           |
     +--------------------+-----------------+-----------------------------------------------------+
-    .. _See Client CONFIGURATION.md for a complete list of configuration properties.
+
+    .. _See Client CONFIGURATION.md for a complete list of configuration properties:
         https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
 
     Args:
@@ -127,7 +130,7 @@ def poll(self, timeout=-1):
         value = msg.value()
         if self._value_deserializer is not None:
             try:
-                value = self._value_deserializer(value, ctx)
+                value = self._value_deserializer(ctx, value)
             except SerializationError as se:
                 raise ConsumeError(KafkaError._VALUE_DESERIALIZATION,
                                    reason=se.message,
@@ -136,7 +139,7 @@ def poll(self, timeout=-1):
         if self._key_deserializer is not None:
             try:
                 ctx.field = MessageField.KEY
-                key = self._key_deserializer(key, ctx)
+                key = self._key_deserializer(ctx, key)
             except SerializationError as se:
                 raise ConsumeError(KafkaError._KEY_DESERIALIZATION,
                                    reason=se.message,

confluent_kafka/error.py

@@ -42,10 +42,8 @@ def __init__(self, error, reason=None, message=None):
         self.message = message
 
     def __repr__(self):
-        return '{klass}(error={error})'.format(
-            klass=self.__class__.__name__,
-            error=self.reason
-        )
+        return str(self)
 
     def __str__(self):
-        return self.reason
+        return "{} (Error code {})".format(self.reason,
+                                           self.error_code)

confluent_kafka/schema_registry/__init__.py

@@ -33,7 +33,7 @@
            "record_subject_name_strategy"]
 
 
-def topic_subject_name_strategy(record_name, ctx):
+def topic_subject_name_strategy(ctx, record_name):
     """
     Constructs a subject name in the form of {topic}-key|value.
 
@@ -47,7 +47,7 @@ def topic_subject_name_strategy(record_name, ctx):
     return ctx.topic + "-" + ctx.field
 
 
-def topic_record_subject_name_strategy(record_name, ctx):
+def topic_record_subject_name_strategy(ctx, record_name):
     """
     Constructs a subject name in the form of {topic}-{record_name}.
 
@@ -61,7 +61,7 @@ def topic_record_subject_name_strategy(record_name, ctx):
     return ctx.topic + "-" + record_name
 
 
-def record_subject_name_strategy(record_name, ctx):
+def record_subject_name_strategy(ctx, record_name):
     """
     Constructs a subject name in the form of {record_name}.
 
@@ -75,7 +75,7 @@ def record_subject_name_strategy(record_name, ctx):
     return record_name
 
 
-def reference_subject_name_strategy(schema_ref, ctx):
+def reference_subject_name_strategy(ctx, schema_ref):
     """
     Constructs a subject reference name in the form of {reference name}.
 

confluent_kafka/schema_registry/avro.py

@@ -125,7 +125,7 @@ class AvroSerializer(Serializer):
 
         schema_str (str): Avro Schema declaration.
 
-        to_dict (callable, optional): Callable(object, SerializationContext) -> dict.
+        to_dict (callable, optional): Callable(SerializationContext, object) -> dict.
             Converts object to a dict.
 
         conf (dict): AvroSerializer configuration.
@@ -189,16 +189,18 @@ def __init__(self, schema_registry_client, schema_str,
         self._schema_name = schema_name
         self._parsed_schema = parsed_schema
 
-    def __call__(self, obj, ctx):
+    def __call__(self, ctx, obj):
         """
         Serializes an object to the Confluent Schema Registry's Avro binary
         format.
 
         Args:
-            obj (object): object instance to serializes.
             ctx (SerializationContext): Metadata pertaining to the serialization
                 operation.
 
+            obj (object): object instance to serializes.
+
+
         Note:
             None objects are represented as Kafka Null.
 
@@ -212,7 +214,7 @@ def __call__(self, obj, ctx):
         if obj is None:
             return None
 
-        subject = self._subject_name_func(self._schema_name, ctx)
+        subject = self._subject_name_func(ctx, self._schema_name)
 
         # Check to ensure this schema has been registered under subject_name.
         if self._auto_register and subject not in self._known_subjects:
@@ -229,7 +231,7 @@ def __call__(self, obj, ctx):
             self._known_subjects.add(subject)
 
         if self.to_dict is not None:
-            value = self.to_dict(obj, ctx)
+            value = self.to_dict(ctx, obj)
         else:
             value = obj
 
@@ -261,7 +263,7 @@ class AvroDeserializer(Deserializer):
 
         schema_str (str): Avro reader schema declaration.
 
-        from_dict (callable, optional): Callable(dict, SerializationContext) -> object.
+        from_dict (callable, optional): Callable(SerializationContext, dict) -> object.
             Converts dict to an instance of some object.
 
     .. _Schema declaration:
@@ -281,20 +283,20 @@ def __init__(self, schema_registry_client, schema_str, from_dict=None):
 
         if from_dict is not None and not callable(from_dict):
             raise ValueError("from_dict must be callable with the signature"
-                             " from_dict(dict, SerializationContext) -> object")
+                             " from_dict(SerializationContext, dict) -> object")
         self.from_dict = from_dict
 
-    def __call__(self, value, ctx):
+    def __call__(self, ctx, value):
         """
         Decodes a Confluent Schema Registry formatted Avro bytes
         to an object.
 
         Arguments:
-            value (bytes): bytes
-
             ctx (SerializationContext): Metadata pertaining to the serialization
                 operation.
 
+            value (bytes): bytes
+
         Raises:
             SerializerError if an error occurs ready data.
 
@@ -332,6 +334,6 @@ def __call__(self, value, ctx):
                                          self._reader_schema)
 
             if self.from_dict is not None:
-                return self.from_dict(obj_dict, ctx)
+                return self.from_dict(ctx, obj_dict)
 
             return obj_dict

confluent_kafka/schema_registry/error.py

@@ -42,7 +42,7 @@ def __init__(self, http_status_code, error_code, error_message):
         self.error_message = error_message
 
     def __repr__(self):
-        return self.__str__()
+        return str(self)
 
     def __str__(self):
         return "{} (HTTP status code {}, SR code {})".format(self.error_message,