[SchemaRegistry] Avro API changes (Azure#23573)

* combine into one exception

* update version + changelog

* remove preamble back compat

* add details ivar to exception

* log all cache info

* add py.typed file

* manifest file

* update to beta version

* fix SRAvro mypy

* update EH version + return type of __message_content__

* remove callbacks

* rename auto_register_schemas

* remove exceptions module

* pylint

* errors inherit from ValueError

* remove typos

* fix MessageContent samples and return value

* mypy/lint

* add InvalidContentError

* adams comments

* lint

Author: swathipil

sdk/eventhub/azure-eventhub/CHANGELOG.md

@@ -1,5 +1,11 @@
 # Release History
 
+## 5.9.0b3 (Unreleased)
+
+### Other Changes
+
+- Updated `EventData` internals for interoperability with the Schema Registry Avro Encoder library.
+
 ## 5.9.0b2 (2022-03-09)
 
 ### Breaking Changes

sdk/eventhub/azure-eventhub/azure/eventhub/_common.py

@@ -18,6 +18,7 @@
     TYPE_CHECKING,
     cast,
 )
+from typing_extensions import TypedDict
 
 import six
 
@@ -60,6 +61,7 @@
 if TYPE_CHECKING:
     import datetime
 
+MessageContent = TypedDict("MessageContent", {"content": bytes, "content_type": str})
 PrimitiveTypes = Optional[Union[
     int,
     float,
@@ -182,16 +184,17 @@ def __str__(self):
         event_str += " }"
         return event_str
 
-    def __message_content__(self) -> Dict:
+    def __message_content__(self) -> MessageContent:
         if self.body_type != AmqpMessageBodyType.DATA:
             raise TypeError('`body_type` must be `AmqpMessageBodyType.DATA`.')
         content = bytearray()
         for c in self.body: # type: ignore
             content += c   # type: ignore
-        return {"content": bytes(content), "content_type": self.content_type}
+        content_type = cast(str, self.content_type)
+        return {"content": bytes(content), "content_type": content_type}
 
     @classmethod
-    def from_message_content(cls, content: bytes, content_type: str) -> "EventData":
+    def from_message_content(cls, content: bytes, content_type: str, **kwargs: Any) -> "EventData": # pylint: disable=unused-argument
         """
         Creates an EventData object given content type and a content value to be set as body.
 

sdk/eventhub/azure-eventhub/azure/eventhub/_version.py

@@ -3,4 +3,4 @@
 # Licensed under the MIT License.
 # ------------------------------------
 
-VERSION = "5.9.0b2"
+VERSION = "5.9.0b3"

sdk/schemaregistry/azure-schemaregistry-avroencoder/CHANGELOG.md

@@ -1,5 +1,20 @@
 # Release History
 
+## 1.0.0b3 (Unreleased)
+
+### Breaking Changes
+
+- `auto_register_schemas` keyword in the sync and async `AvroEncoder` constructors has been renamed `auto_register`.
+- `SchemaParseError`, `SchemaEncodeError`, and `SchemaDecodeError` have been replaced with `InvalidContentError` and `InvalidSchemaError`. The errors have been added under the `azure.schemaregistry.encoder.avroencoder` namespace.
+- The `exceptions` module in `azure.schemaregistry.encoder.avroencoder` has been removed.
+- The `encode` method on the sync and async `AvroEncoder` only allows subtypes of the `MessageType` protocol as values to the `message_type` optional parameter, rather than any callable that has the method signature `(content: bytes, content_type: str, **kwargs: Any)`.
+
+### Other Changes
+
+- This release and future releases will not backward compatibility support for decoding data that was encoded with the AvroSerializer.
+- The `encode` and `decode` methods on `AvroEncoder` support the following message models:
+  - `azure.eventhub.EventData` in `azure-eventhub==5.9.0b3`
+
 ## 1.0.0b2 (2022-03-09)
 
 ### Features Added

sdk/schemaregistry/azure-schemaregistry-avroencoder/MANIFEST.in

@@ -4,3 +4,4 @@ include azure/__init__.py
 include azure/schemaregistry/__init__.py
 recursive-include tests *.py
 recursive-include samples *.py
+include azure/schemaregistry/encoder/avroencoder/py.typed

sdk/schemaregistry/azure-schemaregistry-avroencoder/README.md

@@ -70,7 +70,7 @@ content type with schema ID. Uses [SchemaRegistryClient][schemaregistry_client]
 
 Support has been added to certain Azure Messaging SDK model classes for interoperability with the `AvroEncoder`. These models are subtypes of the `MessageType` protocol defined under the `azure.schemaregistry.encoder.avroencoder` namespace. Currently, the supported model classes are:
 
-- `azure.eventhub.EventData` for `azure-eventhub==5.9.0b2`
+- `azure.eventhub.EventData` for `azure-eventhub==5.9.0b3`
 
 ### Message format
 
@@ -86,7 +86,7 @@ If a message type that follows the MessageType protocol is provided to the encod
   - `avro/binary` is the format indicator
   - `<schema ID>` is the hexadecimal representation of GUID, same format and byte order as the string from the Schema Registry service.
 
-If message type or callback function is not provided, and by default, the encoder will create the following dict:
+If message type is not provided, and by default, the encoder will create the following dict:
 `{"content": <Avro encoded payload>, "content_type": 'avro/binary+<schema ID>' }`
 
 ## Examples
@@ -101,7 +101,7 @@ The following sections provide several code snippets covering some of the most c
 ### Encoding
 
 Use `AvroEncoder.encode` method to encode dict content with the given Avro schema.
-The method will use a schema previously registered to the Schema Registry service and keep the schema cached for future encoding usage. It is also possible to avoid pre-registering the schema to the service and automatically register with the `encode` method by instantiating the `AvroEncoder` with the keyword argument `auto_register_schemas=True`.
+The method will use a schema previously registered to the Schema Registry service and keep the schema cached for future encoding usage. It is also possible to avoid pre-registering the schema to the service and automatically register with the `encode` method by instantiating the `AvroEncoder` with the keyword argument `auto_register=True`.
 
 ```python
 import os
@@ -202,7 +202,7 @@ definition = """
 }"""
 
 schema_registry_client = SchemaRegistryClient(fully_qualified_namespace, token_credential)
-avro_encoder = AvroEncoder(client=schema_registry_client, group_name=group_name, auto_register_schemas=True)
+avro_encoder = AvroEncoder(client=schema_registry_client, group_name=group_name, auto_register=True)
 
 eventhub_producer = EventHubProducerClient.from_connection_string(
     conn_str=eventhub_connection_str,

sdk/schemaregistry/azure-schemaregistry-avroencoder/azure/schemaregistry/encoder/avroencoder/__init__.py

@@ -29,10 +29,13 @@
 
 from ._schema_registry_avro_encoder import AvroEncoder  # pylint: disable=import-error
 from ._message_protocol import MessageType, MessageContent  # pylint: disable=import-error
+from ._exceptions import InvalidContentError, InvalidSchemaError # pylint: disable=import-error
 
 
 __all__ = [
     "AvroEncoder",
     "MessageType",
-    "MessageContent"
+    "MessageContent",
+    "InvalidContentError",
+    "InvalidSchemaError"
 ]

sdk/schemaregistry/azure-schemaregistry-avroencoder/azure/schemaregistry/encoder/avroencoder/_abstract_avro_encoder.py

@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-from typing import BinaryIO, TypeVar, Union, Optional
+from typing import BinaryIO, TypeVar, Union, Any
 from abc import abstractmethod
 
 ObjectType = TypeVar("ObjectType")
@@ -51,9 +51,7 @@ def encode(
     def decode(
         self,
         content: Union[bytes, BinaryIO],
-        schema: str,
-        *,
-        readers_schema: Optional[str]
+        reader: Any
     ):
         """Read the binary representation into a specific type.
         Return type will be ignored, since the schema is deduced from the provided bytes.

sdk/schemaregistry/azure-schemaregistry-avroencoder/azure/schemaregistry/encoder/avroencoder/_apache_avro_encoder.py

@@ -4,10 +4,10 @@
 # --------------------------------------------------------------------------------------------
 
 from functools import lru_cache
-from typing import BinaryIO, Union, TypeVar
+from typing import BinaryIO, Union, TypeVar, cast
 from io import BytesIO
-import avro
-from avro.io import DatumWriter, DatumReader, BinaryDecoder, BinaryEncoder
+import avro # type: ignore
+from avro.io import DatumWriter, DatumReader, BinaryDecoder, BinaryEncoder  # type: ignore
 
 from ._abstract_avro_encoder import AbstractAvroObjectEncoder   # pylint: disable=import-error
 
@@ -36,7 +36,7 @@ def get_schema_writer(self, schema):   # pylint: disable=no-self-use
         return DatumWriter(schema)
 
     @lru_cache(maxsize=128)
-    def get_schema_reader(self, schema, readers_schema):   # pylint: disable=no-self-use
+    def get_schema_reader(self, schema, readers_schema=None):   # pylint: disable=no-self-use
         schema = self.parse_schema(schema)
         if readers_schema:
             readers_schema = self.parse_schema(readers_schema)
@@ -73,9 +73,7 @@ def encode(
     def decode(
         self,
         content,  # type: Union[bytes, BinaryIO]
-        schema,  # type:  Union[str, bytes, avro.schema.Schema]
-        *,
-        readers_schema=None,  # type:  Optional[Union[str, bytes, avro.schema.Schema]]
+        reader,  # type: DatumReader
     ) -> ObjectType:
         """Read the binary representation into a specific type.
         Return type will be ignored, since the schema is deduced from the provided bytes.
@@ -89,11 +87,10 @@ def decode(
         :rtype: ObjectType
         """
         if not hasattr(content, 'read'):
+            content = cast(bytes, content)
             content = BytesIO(content)
 
-        reader = self.get_schema_reader(schema, readers_schema)
-
-        with content:
+        with content:   # type: ignore
             bin_decoder = BinaryDecoder(content)
             decoded_content = reader.read(bin_decoder)
 

sdk/schemaregistry/azure-schemaregistry-avroencoder/azure/schemaregistry/encoder/avroencoder/_exceptions.py

@@ -0,0 +1,53 @@
+# --------------------------------------------------------------------------
+#
+# 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.
+#
+# --------------------------------------------------------------------------
+
+class InvalidSchemaError(ValueError):
+    """Error during schema validation.
+    :param str message: The message object stringified as 'message' attribute
+    :keyword error: The original exception, if any.
+
+    :ivar str message: A stringified version of the message parameter
+    :ivar dict details: The error details related to the schema. Depending on the error,
+     this may include information like: `schema_id`, `schema_definition`, `message_content`.
+    """
+    def __init__(self, message, *args, **kwargs):
+        self.message = str(message)
+        self.details = kwargs.pop("details", {})
+        super(InvalidSchemaError, self).__init__(self.message, *args)
+
+class InvalidContentError(ValueError):
+    """Error during encoding or decoding content with a schema.
+    :param str message: The message object stringified as 'message' attribute
+    :keyword error: The original exception, if any.
+
+    :ivar str message: A stringified version of the message parameter
+    :ivar dict details: The error details. Depending on the error, this may include information like:
+        `schema_id`, `schema_definition`, `message_content`.
+    """
+    def __init__(self, message, *args, **kwargs):
+        self.message = str(message)
+        self.details = kwargs.pop("details", {})
+        super(InvalidContentError, self).__init__(self.message, *args)