Skip to content

Commit 65d8142

Browse files
rnpridgeonRyan P
rnpridgeon
authored and
Ryan P
committed
Fix JSON examples and Avro, JSON serde argument ordering
1 parent a0d969f commit 65d8142

File tree

9 files changed

+45
-40
lines changed

9 files changed

+45
-40
lines changed

‎confluent_kafka/schema_registry/avro.py‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -119,11 +119,11 @@ class AvroSerializer(Serializer):
119119
See ``avro_producer.py`` in the examples directory for example usage.
120120
121121
Args:
122+
schema_str (str): Avro Schema declaration.
123+
122124
schema_registry_client (SchemaRegistryClient): Schema Registry
123125
client instance.
124126
125-
schema_str (str): Avro Schema declaration.
126-
127127
to_dict (callable, optional): Callable(object, SerializationContext) -> dict.
128128
Converts object to a dict.
129129
@@ -144,7 +144,7 @@ class AvroSerializer(Serializer):
144144
_default_conf = {'auto.register.schemas': True,
145145
'subject.name.strategy': topic_subject_name_strategy}
146146

147-
def __init__(self, schema_registry_client, schema_str,
147+
def __init__(self, schema_str, schema_registry_client,
148148
to_dict=None, conf=None):
149149
self._registry = schema_registry_client
150150
self._schema_id = None
@@ -256,11 +256,11 @@ class AvroDeserializer(Deserializer):
256256
directory for example usage.
257257
258258
Args:
259+
schema_str (str): Avro reader schema declaration.
260+
259261
schema_registry_client (SchemaRegistryClient): Confluent Schema Registry
260262
client instance.
261263
262-
schema_str (str): Avro reader schema declaration.
263-
264264
from_dict (callable, optional): Callable(dict, SerializationContext) -> object.
265265
Converts dict to an instance of some object.
266266
@@ -273,7 +273,7 @@ class AvroDeserializer(Deserializer):
273273
"""
274274
__slots__ = ['_reader_schema', '_registry', '_from_dict', '_writer_schemas']
275275

276-
def __init__(self, schema_registry_client, schema_str, from_dict=None):
276+
def __init__(self, schema_str, schema_registry_client, from_dict=None):
277277
self._registry = schema_registry_client
278278
self._writer_schemas = {}
279279

‎confluent_kafka/schema_registry/json_schema.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -96,11 +96,11 @@ class JSONSerializer(Serializer):
9696
and schema registration.
9797
9898
Args:
99+
schema_str (str): JSON Schema definition.
100+
99101
schema_registry_client (SchemaRegistryClient): Schema Registry
100102
client instance.
101103
102-
schema_str (str): JSON Schema definition.
103-
104104
to_dict (callable, optional): Callable(object, SerializationContext) -> dict.
105105
Converts object to a dict.
106106
@@ -121,7 +121,7 @@ class JSONSerializer(Serializer):
121121
_default_conf = {'auto.register.schemas': True,
122122
'subject.name.strategy': topic_subject_name_strategy}
123123

124-
def __init__(self, schema_registry_client, schema_str, to_dict=None,
124+
def __init__(self, schema_str, schema_registry_client, to_dict=None,
125125
conf=None):
126126
self._registry = schema_registry_client
127127
self._schema_id = None

‎examples/avro_consumer.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,9 @@ def main(args):
8383
sr_conf = {'url': args.schema_registry}
8484
schema_registry_client = SchemaRegistryClient(sr_conf)
8585

86-
avro_deserializer = AvroDeserializer(schema_registry_client, schema_str, dict_to_user)
86+
avro_deserializer = AvroDeserializer(schema_str,
87+
schema_registry_client,
88+
dict_to_user)
8789
string_deserializer = StringDeserializer('utf_8')
8890

8991
consumer_conf = {'bootstrap.servers': args.bootstrap_servers,

‎examples/avro_producer.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,9 @@ def main(args):
115115
schema_registry_conf = {'url': args.schema_registry}
116116
schema_registry_client = SchemaRegistryClient(schema_registry_conf)
117117

118-
avro_serializer = AvroSerializer(schema_registry_client, schema_str, user_to_dict)
118+
avro_serializer = AvroSerializer(schema_str,
119+
schema_registry_client,
120+
user_to_dict)
119121

120122
producer_conf = {'bootstrap.servers': args.bootstrap_servers,
121123
'key.serializer': StringSerializer('utf_8'),

‎examples/json_consumer.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ def __init__(self, name=None, favorite_number=None, favorite_color=None):
4444
self.favorite_color = favorite_color
4545

4646

47-
def dict_to_user(ctx, obj):
47+
def dict_to_user(obj, ctx):
4848
"""
4949
Converts object literal(dict) to a User instance.
5050

‎examples/json_producer.py‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -52,16 +52,16 @@ def __init__(self, name, address, favorite_number, favorite_color):
5252
self._address = address
5353

5454

55-
def user_to_dict(ctx, user):
55+
def user_to_dict(user, ctx):
5656
"""
5757
Returns a dict representation of a User instance for serialization.
5858
5959
Args:
60+
user (User): User instance.
61+
6062
ctx (SerializationContext): Metadata pertaining to the serialization
6163
operation.
6264
63-
user (User): User instance.
64-
6565
Returns:
6666
dict: Dict populated with user attributes to be serialized.
6767
@@ -127,7 +127,7 @@ def main(args):
127127
schema_registry_conf = {'url': args.schema_registry}
128128
schema_registry_client = SchemaRegistryClient(schema_registry_conf)
129129

130-
json_serializer = JSONSerializer(schema_registry_client, schema_str, user_to_dict)
130+
json_serializer = JSONSerializer(schema_str, schema_registry_client, user_to_dict)
131131

132132
producer_conf = {'bootstrap.servers': args.bootstrap_servers,
133133
'key.serializer': StringSerializer('utf_8'),

‎tests/integration/schema_registry/test_avro_serializers.py‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -72,9 +72,9 @@ def test_avro_record_serialization(kafka_cluster, load_file, avsc, data, record_
7272
sr = kafka_cluster.schema_registry()
7373

7474
schema_str = load_file(avsc)
75-
value_serializer = AvroSerializer(sr, schema_str)
75+
value_serializer = AvroSerializer(schema_str, sr)
7676

77-
value_deserializer = AvroDeserializer(sr, schema_str)
77+
value_deserializer = AvroDeserializer(schema_str, sr)
7878

7979
producer = kafka_cluster.producer(value_serializer=value_serializer)
8080

@@ -116,9 +116,9 @@ def test_delivery_report_serialization(kafka_cluster, load_file, avsc, data, rec
116116
sr = kafka_cluster.schema_registry()
117117
schema_str = load_file(avsc)
118118

119-
value_serializer = AvroSerializer(sr, schema_str)
119+
value_serializer = AvroSerializer(schema_str, sr)
120120

121-
value_deserializer = AvroDeserializer(sr, schema_str)
121+
value_deserializer = AvroDeserializer(schema_str, sr)
122122

123123
producer = kafka_cluster.producer(value_serializer=value_serializer)
124124

@@ -163,13 +163,13 @@ def test_avro_record_serialization_custom(kafka_cluster):
163163
sr = kafka_cluster.schema_registry()
164164

165165
user = User('Bowie', 47, 'purple')
166-
value_serializer = AvroSerializer(sr, User.schema_str,
166+
value_serializer = AvroSerializer(User.schema_str, sr,
167167
lambda user, ctx:
168168
dict(name=user.name,
169169
favorite_number=user.favorite_number,
170170
favorite_color=user.favorite_color))
171171

172-
value_deserializer = AvroDeserializer(sr, User.schema_str,
172+
value_deserializer = AvroDeserializer(User.schema_str, sr,
173173
lambda user_dict, ctx:
174174
User(**user_dict))
175175

‎tests/integration/schema_registry/test_json_serializers.py‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ def test_json_record_serialization(kafka_cluster, load_file):
9494
sr = kafka_cluster.schema_registry({'url': 'http://localhost:8081'})
9595

9696
schema_str = load_file("product.json")
97-
value_serializer = JSONSerializer(sr, schema_str)
97+
value_serializer = JSONSerializer(schema_str, sr)
9898
value_deserializer = JSONDeserializer(schema_str)
9999

100100
producer = kafka_cluster.producer(value_serializer=value_serializer)
@@ -142,7 +142,7 @@ def test_json_record_serialization_incompatible(kafka_cluster, load_file):
142142
sr = kafka_cluster.schema_registry({'url': 'http://localhost:8081'})
143143

144144
schema_str = load_file("product.json")
145-
value_serializer = JSONSerializer(sr, schema_str)
145+
value_serializer = JSONSerializer(schema_str, sr)
146146
producer = kafka_cluster.producer(value_serializer=value_serializer)
147147

148148
record = {"contractorId": 1,
@@ -170,7 +170,7 @@ def test_json_record_serialization_no_title(kafka_cluster, load_file):
170170

171171
with pytest.raises(ValueError,
172172
match="Missing required JSON schema annotation title"):
173-
JSONSerializer(sr, schema_str)
173+
JSONSerializer(schema_str, sr)
174174

175175

176176
def test_json_record_serialization_custom(kafka_cluster, load_file):
@@ -187,7 +187,7 @@ def test_json_record_serialization_custom(kafka_cluster, load_file):
187187
sr = kafka_cluster.schema_registry({'url': 'http://localhost:8081'})
188188

189189
schema_str = load_file("product.json")
190-
value_serializer = JSONSerializer(sr, schema_str,
190+
value_serializer = JSONSerializer(schema_str, sr,
191191
to_dict=_testProduct_to_dict)
192192
value_deserializer = JSONDeserializer(schema_str,
193193
from_dict=_testProduct_from_dict)
@@ -233,7 +233,7 @@ def test_json_record_deserialization_mismatch(kafka_cluster, load_file):
233233
schema_str = load_file("contractor.json")
234234
schema_str2 = load_file("product.json")
235235

236-
value_serializer = JSONSerializer(sr, schema_str)
236+
value_serializer = JSONSerializer(schema_str, sr)
237237
value_deserializer = JSONDeserializer(schema_str2)
238238

239239
producer = kafka_cluster.producer(value_serializer=value_serializer)

‎tests/schema_registry/test_avro_serializer.py‎

Lines changed: 14 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ def test_avro_serializer_config_auto_register_schemas():
3434
"""
3535
conf = {'url': TEST_URL}
3636
test_client = SchemaRegistryClient(conf)
37-
test_serializer = AvroSerializer(test_client, "string",
37+
test_serializer = AvroSerializer("string", test_client,
3838
conf={'auto.register.schemas': False})
3939
assert not test_serializer._auto_register
4040

@@ -47,7 +47,7 @@ def test_avro_serializer_config_auto_register_schemas_invalid():
4747
test_client = SchemaRegistryClient(conf)
4848

4949
with pytest.raises(ValueError, match="must be a boolean"):
50-
AvroSerializer(test_client, "string",
50+
AvroSerializer("string", test_client,
5151
conf={'auto.register.schemas': dict()})
5252

5353

@@ -60,7 +60,7 @@ def test_avro_serializer_config_auto_register_schemas_false(mock_schema_registry
6060
topic = "test-auto-register"
6161
subject = topic + '-key'
6262

63-
test_serializer = AvroSerializer(test_client, "string",
63+
test_serializer = AvroSerializer("string", test_client,
6464
conf={'auto.register.schemas': False})
6565

6666
test_serializer("test",
@@ -81,7 +81,7 @@ def test_avro_serializer_config_subject_name_strategy():
8181

8282
conf = {'url': TEST_URL}
8383
test_client = SchemaRegistryClient(conf)
84-
test_serializer = AvroSerializer(test_client, "int",
84+
test_serializer = AvroSerializer("int", test_client,
8585
conf={'subject.name.strategy':
8686
record_subject_name_strategy})
8787

@@ -96,7 +96,7 @@ def test_avro_serializer_config_subject_name_strategy_invalid():
9696
conf = {'url': TEST_URL}
9797
test_client = SchemaRegistryClient(conf)
9898
with pytest.raises(ValueError, match="must be callable"):
99-
AvroSerializer(test_client, "int",
99+
AvroSerializer("int", test_client,
100100
conf={'subject.name.strategy': dict()})
101101

102102

@@ -106,8 +106,8 @@ def test_avro_serializer_record_subject_name_strategy(load_avsc):
106106
"""
107107
conf = {'url': TEST_URL}
108108
test_client = SchemaRegistryClient(conf)
109-
test_serializer = AvroSerializer(test_client,
110-
load_avsc('basic_schema.avsc'),
109+
test_serializer = AvroSerializer(load_avsc('basic_schema.avsc'),
110+
test_client,
111111
conf={'subject.name.strategy':
112112
record_subject_name_strategy})
113113

@@ -123,7 +123,7 @@ def test_avro_serializer_record_subject_name_strategy_primitive(load_avsc):
123123
"""
124124
conf = {'url': TEST_URL}
125125
test_client = SchemaRegistryClient(conf)
126-
test_serializer = AvroSerializer(test_client, 'int',
126+
test_serializer = AvroSerializer('int', test_client,
127127
conf={'subject.name.strategy':
128128
record_subject_name_strategy})
129129

@@ -138,8 +138,8 @@ def test_avro_serializer_topic_record_subject_name_strategy(load_avsc):
138138
"""
139139
conf = {'url': TEST_URL}
140140
test_client = SchemaRegistryClient(conf)
141-
test_serializer = AvroSerializer(test_client,
142-
load_avsc('basic_schema.avsc'),
141+
test_serializer = AvroSerializer(load_avsc('basic_schema.avsc'),
142+
test_client,
143143
conf={'subject.name.strategy':
144144
topic_record_subject_name_strategy})
145145

@@ -155,7 +155,8 @@ def test_avro_serializer_topic_record_subject_name_strategy_primitive(load_avsc)
155155
"""
156156
conf = {'url': TEST_URL}
157157
test_client = SchemaRegistryClient(conf)
158-
test_serializer = AvroSerializer(test_client, 'int',
158+
test_serializer = AvroSerializer('int',
159+
test_client,
159160
conf={'subject.name.strategy':
160161
topic_record_subject_name_strategy})
161162

@@ -170,8 +171,8 @@ def test_avro_serializer_subject_name_strategy_default(load_avsc):
170171
"""
171172
conf = {'url': TEST_URL}
172173
test_client = SchemaRegistryClient(conf)
173-
test_serializer = AvroSerializer(test_client,
174-
load_avsc('basic_schema.avsc'))
174+
test_serializer = AvroSerializer(load_avsc('basic_schema.avsc'),
175+
test_client)
175176

176177
ctx = SerializationContext('test_subj', MessageField.VALUE)
177178
assert test_serializer._subject_name_func(

0 commit comments

Comments
 (0)