DGS-21988 Fix transformation of nullable JSON props (#2032)

* DGS-21988 Fix transformation of nullable JSON props * Minor cleanup

Author: rayokota

src/confluent_kafka/schema_registry/common/json_schema.py

@@ -73,6 +73,14 @@ def transform(
  field_ctx = ctx.current_field()
  if field_ctx is not None:
  field_ctx.field_type = get_type(schema)
+ original_type = schema.get("type")
+ if isinstance(original_type, list) and len(original_type) > 0:
+ subschema = _validate_subtypes(schema, message, ref_registry)
+ try:
+ if subschema is not None:
+ return transform(ctx, subschema, ref_registry, ref_resolver, path, message, field_transform)
+ finally:
+ schema["type"] = original_type # restore original type
  all_of = schema.get("allOf")
  if all_of is not None:
  subschema = _validate_subschemas(all_of, message, ref_registry)
@@ -136,6 +144,22 @@ def _transform_field(
  ctx.exit_field()
 
 
+def _validate_subtypes(
+ schema: JsonSchema, message: JsonMessage, registry: Registry
+) -> Optional[JsonSchema]:
+ schema_type = schema.get("type")
+ if not isinstance(schema_type, list) or len(schema_type) == 0:
+ return None
+ for typ in schema_type:
+ schema["type"] = typ
+ try:
+ validate(instance=message, schema=schema, registry=registry)
+ return schema
+ except ValidationError:
+ pass
+ return None
+
+
 def _validate_subschemas(
  subschemas: List[JsonSchema],
  message: JsonMessage,

tests/schema_registry/_async/test_json_serdes.py

@@ -573,6 +573,72 @@ async def test_json_cel_field_transform():
  assert obj2 == newobj
 
 
+async def test_json_cel_field_transform_with_nullable():
+ conf = {'url': _BASE_URL}
+ client = AsyncSchemaRegistryClient.new_client(conf)
+ ser_conf = {'auto.register.schemas': False, 'use.latest.version': True}
+ schema = {
+ "type": "object",
+ "properties": {
+ "intField": {"type": "integer"},
+ "doubleField": {"type": "number"},
+ "stringField": {
+ "type": ["string", "null"],
+ "confluent:tags": ["PII"]
+ },
+ "booleanField": {"type": "boolean"},
+ "bytesField": {
+ "type": "string",
+ "contentEncoding": "base64",
+ "confluent:tags": ["PII"]
+ }
+ }
+ }
+
+ rule = Rule(
+ "test-cel",
+ "",
+ RuleKind.TRANSFORM,
+ RuleMode.WRITE,
+ "CEL_FIELD",
+ None,
+ None,
+ "name == 'stringField' ; value + '-suffix'",
+ None,
+ None,
+ False
+ )
+ await client.register_schema(_SUBJECT, Schema(
+ json.dumps(schema),
+ "JSON",
+ [],
+ None,
+ RuleSet(None, [rule])
+ ))
+
+ obj = {
+ 'intField': 123,
+ 'doubleField': 45.67,
+ 'stringField': 'hi',
+ 'booleanField': True,
+ 'bytesField': base64.b64encode(b'foobar').decode('utf-8'),
+ }
+ ser = await AsyncJSONSerializer(json.dumps(schema), client, conf=ser_conf)
+ ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+ obj_bytes = await ser(obj, ser_ctx)
+
+ obj2 = {
+ 'intField': 123,
+ 'doubleField': 45.67,
+ 'stringField': 'hi-suffix',
+ 'booleanField': True,
+ 'bytesField': base64.b64encode(b'foobar').decode('utf-8'),
+ }
+ deser = await AsyncJSONDeserializer(None, schema_registry_client=client)
+ newobj = await deser(obj_bytes, ser_ctx)
+ assert obj2 == newobj
+
+
 async def test_json_cel_field_transform_with_def():
  conf = {'url': _BASE_URL}
  client = AsyncSchemaRegistryClient.new_client(conf)

tests/schema_registry/_sync/test_json_serdes.py

@@ -573,6 +573,72 @@ def test_json_cel_field_transform():
  assert obj2 == newobj
 
 
+def test_json_cel_field_transform_with_nullable():
+ conf = {'url': _BASE_URL}
+ client = SchemaRegistryClient.new_client(conf)
+ ser_conf = {'auto.register.schemas': False, 'use.latest.version': True}
+ schema = {
+ "type": "object",
+ "properties": {
+ "intField": {"type": "integer"},
+ "doubleField": {"type": "number"},
+ "stringField": {
+ "type": ["string", "null"],
+ "confluent:tags": ["PII"]
+ },
+ "booleanField": {"type": "boolean"},
+ "bytesField": {
+ "type": "string",
+ "contentEncoding": "base64",
+ "confluent:tags": ["PII"]
+ }
+ }
+ }
+
+ rule = Rule(
+ "test-cel",
+ "",
+ RuleKind.TRANSFORM,
+ RuleMode.WRITE,
+ "CEL_FIELD",
+ None,
+ None,
+ "name == 'stringField' ; value + '-suffix'",
+ None,
+ None,
+ False
+ )
+ client.register_schema(_SUBJECT, Schema(
+ json.dumps(schema),
+ "JSON",
+ [],
+ None,
+ RuleSet(None, [rule])
+ ))
+
+ obj = {
+ 'intField': 123,
+ 'doubleField': 45.67,
+ 'stringField': 'hi',
+ 'booleanField': True,
+ 'bytesField': base64.b64encode(b'foobar').decode('utf-8'),
+ }
+ ser = JSONSerializer(json.dumps(schema), client, conf=ser_conf)
+ ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+ obj_bytes = ser(obj, ser_ctx)
+
+ obj2 = {
+ 'intField': 123,
+ 'doubleField': 45.67,
+ 'stringField': 'hi-suffix',
+ 'booleanField': True,
+ 'bytesField': base64.b64encode(b'foobar').decode('utf-8'),
+ }
+ deser = JSONDeserializer(None, schema_registry_client=client)
+ newobj = deser(obj_bytes, ser_ctx)
+ assert obj2 == newobj
+
+
 def test_json_cel_field_transform_with_def():
  conf = {'url': _BASE_URL}
  client = SchemaRegistryClient.new_client(conf)