Tests are only slightly working...

Author: fatimarahman

.python-version

@@ -1 +1 @@
-3.9.16
+3.9.16

CHANGELOG.md

@@ -1,4 +1,7 @@
 # Changelog
+## v1.1.6 6/26/24
+- Generalized Avro functions and separated encoding/decoding behavior.
+
 ## v1.1.5 6/6/24
 - Use executemany instead of execute when appropriate in RedshiftClient.execute_transaction
 

pyproject.toml

@@ -23,7 +23,7 @@ dependencies = []
 "Bug Tracker" = "https://github.com/NYPL/python-utils/issues"
 
 [project.optional-dependencies]
-avro-encoder = [
+avro-client = [
     "avro>=1.11.1",
     "requests>=2.28.1"
 ]
@@ -67,7 +67,7 @@ research-catalog-identifier-helper = [
     "requests>=2.28.1"
 ]
 development = [
-    "nypl_py_utils[avro-encoder,kinesis-client,kms-client,mysql-client,oauth2-api-client,postgresql-client,postgresql-pool-client,redshift-client,s3-client,config-helper,obfuscation-helper,research-catalog-identifier-helper]",
+    "nypl_py_utils[avro-client,kinesis-client,kms-client,mysql-client,oauth2-api-client,postgresql-client,postgresql-pool-client,redshift-client,s3-client,config-helper,obfuscation-helper,research-catalog-identifier-helper]",
     "flake8>=6.0.0",
     "freezegun>=1.2.2",
     "mock>=4.0.3",

src/nypl_py_utils/classes/avro_encoder.py src/nypl_py_utils/classes/avro_client.pysrc/nypl_py_utils/classes/avro_encoder.py renamed to src/nypl_py_utils/classes/avro_client.py

@@ -1,13 +1,14 @@
 import avro.schema
+import base64
 import requests
 
-from avro.errors import AvroException
+from avro.schema import AvroException
 from avro.io import BinaryDecoder, BinaryEncoder, DatumReader, DatumWriter
 from io import BytesIO
 from nypl_py_utils.functions.log_helper import create_log
 from requests.exceptions import JSONDecodeError, RequestException
 
-class AvroInterpreter:
+class AvroClient:
     """
     Base class for Avro schema interaction. Takes as input the
     Platform API endpoint from which to fetch the schema in JSON format.
@@ -32,7 +33,7 @@ def get_json_schema(self, platform_schema_url):
             self.logger.error(
                 'Failed to retrieve schema from {url}: {error}'.format(
                     url=platform_schema_url, error=e))
-            raise AvroInterpreterError(
+            raise AvroClientError(
                 'Failed to retrieve schema from {url}: {error}'.format(
                     url=platform_schema_url, error=e)) from None
 
@@ -43,12 +44,12 @@ def get_json_schema(self, platform_schema_url):
             self.logger.error(
                 'Retrieved schema is malformed: {errorType} {errorMessage}'
                 .format(errorType=type(e), errorMessage=e))
-            raise AvroInterpreterError(
+            raise AvroClientError(
                 'Retrieved schema is malformed: {errorType} {errorMessage}'
                 .format(errorType=type(e), errorMessage=e)) from None
 
 
-class AvroEncoder(AvroInterpreter):
+class AvroEncoder(AvroClient):
     """
     Class for encoding records using an Avro schema. Takes as input the
     Platform API endpoint from which to fetch the schema in JSON format.
@@ -71,7 +72,7 @@ def encode_record(self, record):
                 return output_stream.getvalue()
             except AvroException as e:
                 self.logger.error('Failed to encode record: {}'.format(e))
-                raise AvroInterpreterError(
+                raise AvroClientError(
                     'Failed to encode record: {}'.format(e)) from None
 
     def encode_batch(self, record_list):
@@ -95,37 +96,51 @@ def encode_batch(self, record_list):
                     output_stream.truncate(0)
                 except AvroException as e:
                     self.logger.error('Failed to encode record: {}'.format(e))
-                    raise AvroInterpreterError(
+                    raise AvroClientError(
                         'Failed to encode record: {}'.format(e)) from None
         return encoded_records
 
 
-class AvroDecoder(AvroInterpreter):
+class AvroDecoder(AvroClient):
     """
     Class for decoding records using an Avro schema. Takes as input the
     Platform API endpoint from which to fetch the schema in JSON format.
     """
 
-    def decode_record(self, record):
+    def decode_record(self, record, encoding="binary"):
         """
-        Decodes a single record represented as a byte string using the given
-        Avro schema.
+        Decodes a single record represented either as a byte or 
+        base64 string, using the given Avro schema.
 
         Returns a dictionary where each key is a field in the schema.
         """
         self.logger.debug('Decoding {rec} using {schema} schema'.format(
             rec=record, schema=self.schema.name))
+        
+        if encoding == "base64":
+            return self._decode_base64(record)
+        elif encoding == "binary":
+            return self._decode_binary(record)
+        else:
+            self.logger.error('Failed to decode record due to encoding type: {}'.format(encoding))
+            raise AvroClientError(
+                'Invalid encoding type: {}'.format(encoding))
+
+    def _decode_base64(self, record):
+        return base64.b64decode(record).decode('utf-8')
+
+    def _decode_binary(self, record):
         datum_reader = DatumReader(self.schema)
         with BytesIO(record) as input_stream:
             decoder = BinaryDecoder(input_stream)
             try:
                 return datum_reader.read(decoder)
             except Exception as e:
                 self.logger.error('Failed to decode record: {}'.format(e))
-                raise AvroInterpreterError(
+                raise AvroClientError(
                     'Failed to decode record: {}'.format(e)) from None
 
 
-class AvroInterpreterError(Exception):
+class AvroClientError(Exception):
     def __init__(self, message=None):
         self.message = message