[SchemaRegistry] avro serializer add exceptions

sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/_abstract_avro_serializer.py

@@ -0,0 +1,50 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from typing import BinaryIO, Dict, Mapping, Any, TypeVar, Union
+from abc import abstractmethod
+import avro
+
+ObjectType = TypeVar("ObjectType")
+
+class AbstractAvroObjectSerializer(object):
+    """
+    An Avro serializer used for serializing/deserializing an Avro RecordSchema.
+    """
+
+    @abstractmethod
+    def serialize(
+        self,
+        data,  # type: ObjectType
+        schema,  # type: Union[str, bytes, avro.schema.Schema]
+    ):
+        # type: (ObjectType, Union[str, bytes, avro.schema.Schema]) -> bytes
+        """Convert the provided value to it's binary representation and write it to the stream.
+        Schema must be a Avro RecordSchema:
+        https://avro.apache.org/docs/1.10.0/gettingstartedpython.html#Defining+a+schema
+        :param data: An object to serialize
+        :type data: ObjectType
+        :param schema: An Avro RecordSchema
+        :type schema: Union[str, bytes, avro.schema.Schema]
+        :returns: Encoded bytes
+        :rtype: bytes
+        """
+
+    @abstractmethod
+    def deserialize(
+        self,
+        data,  # type: Union[bytes, BinaryIO]
+        schema,  # type:  Union[str, bytes, avro.schema.Schema]
+    ):
+        # type: (Union[bytes, BinaryIO], Union[str, bytes, avro.schema.Schema]) -> ObjectType
+        """Read the binary representation into a specific type.
+        Return type will be ignored, since the schema is deduced from the provided bytes.
+        :param data: A stream of bytes or bytes directly
+        :type data: BinaryIO or bytes
+        :param schema: An Avro RecordSchema
+        :type schema: Union[str, bytes, avro.schema.Schema]
+        :returns: An instantiated object
+        :rtype: ObjectType
+        """

sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/_avro_serializer.py → sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/_apache_avro_serializer.py

@@ -1,28 +1,8 @@
-# --------------------------------------------------------------------------
-#
+# --------------------------------------------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
-#
-# The MIT License (MIT)
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the ""Software""), to
-# deal in the Software without restriction, including without limitation the
-# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
-# sell copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in
-# all copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
-# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
-# IN THE SOFTWARE.
-#
-# --------------------------------------------------------------------------
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
 try:
     from functools import lru_cache
 except ImportError:
@@ -32,10 +12,12 @@
 import avro
 from avro.io import DatumWriter, DatumReader, BinaryDecoder, BinaryEncoder
 
+from ._abstract_avro_serializer import AbstractAvroObjectSerializer
+
 ObjectType = TypeVar("ObjectType")
 
 
-class AvroObjectSerializer(object):
+class ApacheAvroObjectSerializer(AbstractAvroObjectSerializer):
 
     def __init__(self, codec=None):
         """A Avro serializer using avro lib from Apache.
@@ -44,13 +26,17 @@ def __init__(self, codec=None):
         self._writer_codec = codec
 
     @lru_cache(maxsize=128)
-    def _get_schema_writer(self, schema):   # pylint: disable=no-self-use
-        schema = avro.schema.parse(schema)
+    def parse_schema(self, schema):   # pylint: disable=no-self-use
+        return avro.schema.parse(schema)
+
+    @lru_cache(maxsize=128)
+    def get_schema_writer(self, schema):   # pylint: disable=no-self-use
+        schema = self.parse_schema(schema)
         return DatumWriter(schema)
 
     @lru_cache(maxsize=128)
-    def _get_schema_reader(self, schema):   # pylint: disable=no-self-use
-        schema = avro.schema.parse(schema)
+    def get_schema_reader(self, schema):   # pylint: disable=no-self-use
+        schema = self.parse_schema(schema)
         return DatumReader(writers_schema=schema)
 
     # pylint: disable=no-self-use
@@ -73,7 +59,7 @@ def serialize(
         if not schema:
             raise ValueError("Schema is required in Avro serializer.")
 
-        writer = self._get_schema_writer(str(schema))
+        writer = self.get_schema_writer(str(schema))
 
         stream = BytesIO()
         with stream:
@@ -100,7 +86,7 @@ def deserialize(
         if not hasattr(data, 'read'):
             data = BytesIO(data)
 
-        reader = self._get_schema_reader(str(schema))
+        reader = self.get_schema_reader(str(schema))
 
         with data:
             bin_decoder = BinaryDecoder(data)

sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/_schema_registry_avro_serializer.py

@@ -27,12 +27,16 @@
     from functools import lru_cache
 except ImportError:
     from backports.functools_lru_cache import lru_cache
+from ._apache_avro_serializer import ApacheAvroObjectSerializer as AvroObjectSerializer
 from io import BytesIO
 from typing import Any, Dict, Mapping
 
+from .exceptions import (
+    SchemaParseError,
+    SchemaSerializationError,
+    SchemaDeserializationError,
+)
 from ._constants import SCHEMA_ID_START_INDEX, SCHEMA_ID_LENGTH, DATA_START_INDEX
-from ._avro_serializer import AvroObjectSerializer
-from ._utils import parse_schema
 
 
 class AvroSerializer(object):
@@ -53,16 +57,18 @@ def __init__(self, **kwargs):
         # type: (Any) -> None
         try:
             self._schema_group = kwargs.pop("group_name")
-            self._schema_registry_client = kwargs.pop("client") # type: "SchemaRegistryClient"
+            self._schema_registry_client = kwargs.pop(
+                "client"
+            )  # type: "SchemaRegistryClient"
         except KeyError as e:
             raise TypeError("'{}' is a required keyword.".format(e.args[0]))
         self._avro_serializer = AvroObjectSerializer(codec=kwargs.get("codec"))
         self._auto_register_schemas = kwargs.get("auto_register_schemas", False)
         self._auto_register_schema_func = (
-                self._schema_registry_client.register_schema
-                if self._auto_register_schemas
-                else self._schema_registry_client.get_schema_properties
-            )
+            self._schema_registry_client.register_schema
+            if self._auto_register_schemas
+            else self._schema_registry_client.get_schema_properties
+        )
 
     def __enter__(self):
         # type: () -> SchemaRegistryAvroSerializer
@@ -131,10 +137,24 @@ def serialize(self, value, **kwargs):
         except KeyError as e:
             raise TypeError("'{}' is a required keyword.".format(e.args[0]))
 
-        cached_schema = parse_schema(raw_input_schema)
         record_format_identifier = b"\0\0\0\0"
-        schema_id = self._get_schema_id(cached_schema.fullname, str(cached_schema), **kwargs)
-        data_bytes = self._avro_serializer.serialize(value, cached_schema)
+
+        try:
+            cached_schema = self._avro_serializer.parse_schema(raw_input_schema)
+            schema_fullname = cached_schema.fullname
+        except Exception as e:
+            raise SchemaParseError("Cannot parse schema: {}".format(raw_input_schema), error=e)
+
+        schema_id = self._get_schema_id(
+            schema_fullname, str(cached_schema), **kwargs
+        )
+        try:
+            data_bytes = self._avro_serializer.serialize(value, cached_schema)
+        except Exception as e:
+            raise SchemaSerializationError(
+                "Cannot serialize value '{}' for schema: {}".format(value, str(cached_schema)),
+                error=e
+            )
 
         stream = BytesIO()
 
@@ -161,7 +181,13 @@ def deserialize(self, value, **kwargs):
         ].decode("utf-8")
         schema_definition = self._get_schema(schema_id, **kwargs)
 
-        dict_value = self._avro_serializer.deserialize(
-            value[DATA_START_INDEX:], schema_definition
-        )
+        try:
+            dict_value = self._avro_serializer.deserialize(
+                value[DATA_START_INDEX:], schema_definition
+            )
+        except Exception as e:
+            raise SchemaDeserializationError(
+                "Cannot deserialize value '{}' for schema: {}".format(value[DATA_START_INDEX], schema_definition),
+                error=e
+            )
         return dict_value

sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/aio/_schema_registry_avro_serializer_async.py

@@ -27,8 +27,12 @@
 from typing import Any, Dict, Mapping
 from ._async_lru import alru_cache
 from .._constants import SCHEMA_ID_START_INDEX, SCHEMA_ID_LENGTH, DATA_START_INDEX
-from .._avro_serializer import AvroObjectSerializer
-from .._utils import parse_schema
+from .._apache_avro_serializer import ApacheAvroObjectSerializer as AvroObjectSerializer
+from ..exceptions import (
+    SchemaParseError,
+    SchemaSerializationError,
+    SchemaDeserializationError,
+)
 
 
 class AvroSerializer(object):
@@ -126,11 +130,22 @@ async def serialize(self, value, **kwargs):
             raw_input_schema = kwargs.pop("schema")
         except KeyError as e:
             raise TypeError("'{}' is a required keyword.".format(e.args[0]))
-
-        cached_schema = parse_schema(raw_input_schema)
         record_format_identifier = b"\0\0\0\0"
-        schema_id = await self._get_schema_id(cached_schema.fullname, str(cached_schema), **kwargs)
-        data_bytes = self._avro_serializer.serialize(value, cached_schema)
+
+        try:
+            cached_schema = self._avro_serializer.parse_schema(raw_input_schema)
+            schema_fullname = cached_schema.fullname
+        except Exception as e:
+            raise SchemaParseError("Cannot parse schema: {}".format(raw_input_schema), error=e)
+
+        schema_id = await self._get_schema_id(schema_fullname, str(cached_schema), **kwargs)
+        try:
+            data_bytes = self._avro_serializer.serialize(value, cached_schema)
+        except Exception as e:
+            raise SchemaSerializationError(
+                "Cannot serialize value '{}' for schema: {}".format(value, str(cached_schema)),
+                error=e
+            )
 
         stream = BytesIO()
 
@@ -157,7 +172,13 @@ async def deserialize(self, value, **kwargs):
         ].decode("utf-8")
         schema_definition = await self._get_schema(schema_id, **kwargs)
 
-        dict_value = self._avro_serializer.deserialize(
-            value[DATA_START_INDEX:], schema_definition
-        )
+        try:
+            dict_value = self._avro_serializer.deserialize(
+                value[DATA_START_INDEX:], schema_definition
+            )
+        except Exception as e:
+            raise SchemaDeserializationError(
+                "Cannot deserialize value '{}' for schema: {}".format(value[DATA_START_INDEX], schema_definition),
+                error=e
+            )
         return dict_value

sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/exceptions.py

@@ -0,0 +1,13 @@
+from azure.core.exceptions import AzureError
+
+class SchemaParseError(AzureError):
+    def __init__(self, message, *args, **kwargs):
+        super(SchemaParseError, self).__init__(message, *args, **kwargs)
+
+class SchemaSerializationError(AzureError):
+    def __init__(self, message, *args, **kwargs):
+        super(SchemaSerializationError, self).__init__(message, *args, **kwargs)
+
+class SchemaDeserializationError(AzureError):
+    def __init__(self, message, *args, **kwargs):
+        super(SchemaDeserializationError, self).__init__(message, *args, **kwargs)

sdk/schemaregistry/azure-schemaregistry-avroserializer/tests/recordings/test_avro_serializer.test_basic_sr_avro_serializer_with_auto_register_schemas.yaml

@@ -22,18 +22,18 @@ interactions:
     uri: https://fake_resource.servicebus.windows.net/$schemagroups/fakegroup/schemas/example.avro.User?api-version=2020-09-01-preview
   response:
     body:
-      string: '{"id":"a9619ab12fb748ab9ec800c13850107e"}'
+      string: '{"id":"7b7cdb460f804bc8b13b7c1e924b446b"}'
     headers:
       content-type:
       - application/json
       date:
-      - Wed, 20 Oct 2021 19:04:59 GMT
+      - Fri, 22 Oct 2021 17:07:00 GMT
       location:
       - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/fakegroup/schemas/example.avro.User/versions/1?api-version=2020-09-01-preview
       schema-id:
-      - a9619ab12fb748ab9ec800c13850107e
+      - 7b7cdb460f804bc8b13b7c1e924b446b
       schema-id-location:
-      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/a9619ab12fb748ab9ec800c13850107e?api-version=2020-09-01-preview
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/7b7cdb460f804bc8b13b7c1e924b446b?api-version=2020-09-01-preview
       schema-version:
       - '1'
       schema-versions-location:
@@ -72,18 +72,18 @@ interactions:
     uri: https://fake_resource.servicebus.windows.net/$schemagroups/fakegroup/schemas/example.avro.User?api-version=2020-09-01-preview
   response:
     body:
-      string: '{"id":"a9619ab12fb748ab9ec800c13850107e"}'
+      string: '{"id":"7b7cdb460f804bc8b13b7c1e924b446b"}'
     headers:
       content-type:
       - application/json
       date:
-      - Wed, 20 Oct 2021 19:05:00 GMT
+      - Fri, 22 Oct 2021 17:07:01 GMT
       location:
       - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/fakegroup/schemas/example.avro.User/versions/1?api-version=2020-09-01-preview
       schema-id:
-      - a9619ab12fb748ab9ec800c13850107e
+      - 7b7cdb460f804bc8b13b7c1e924b446b
       schema-id-location:
-      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/a9619ab12fb748ab9ec800c13850107e?api-version=2020-09-01-preview
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/7b7cdb460f804bc8b13b7c1e924b446b?api-version=2020-09-01-preview
       schema-version:
       - '1'
       schema-versions-location:
@@ -111,21 +111,21 @@ interactions:
       User-Agent:
       - azsdk-python-azureschemaregistry/1.0.0b3 Python/3.9.0 (Windows-10-10.0.19041-SP0)
     method: GET
-    uri: https://fake_resource.servicebus.windows.net/$schemagroups/getSchemaById/a9619ab12fb748ab9ec800c13850107e?api-version=2020-09-01-preview
+    uri: https://fake_resource.servicebus.windows.net/$schemagroups/getSchemaById/7b7cdb460f804bc8b13b7c1e924b446b?api-version=2020-09-01-preview
   response:
     body:
       string: '{"type":"record","name":"User","namespace":"example.avro","fields":[{"type":"string","name":"name"},{"type":["int","null"],"name":"favorite_number"},{"type":["string","null"],"name":"favorite_color"}]}'
     headers:
       content-type:
       - application/json
       date:
-      - Wed, 20 Oct 2021 19:05:00 GMT
+      - Fri, 22 Oct 2021 17:07:01 GMT
       location:
       - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/fakegroup/schemas/example.avro.User/versions/1?api-version=2020-09-01-preview
       schema-id:
-      - a9619ab12fb748ab9ec800c13850107e
+      - 7b7cdb460f804bc8b13b7c1e924b446b
       schema-id-location:
-      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/a9619ab12fb748ab9ec800c13850107e?api-version=2020-09-01-preview
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/7b7cdb460f804bc8b13b7c1e924b446b?api-version=2020-09-01-preview
       schema-version:
       - '1'
       schema-versions-location: