refactor: post serialization simplified

.vscode/settings.json

@@ -3,5 +3,7 @@
         "."
     ],
     "python.testing.unittestEnabled": false,
-    "python.testing.pytestEnabled": true
+    "python.testing.pytestEnabled": true,
+    "editor.defaultFormatter": "ms-python.black-formatter",
+    "editor.formatOnSave": true
 }

config/data/schemas.py

@@ -1,6 +1,7 @@
 from marshmallow import fields, post_load
 
 from config.domain.entities import ConfigurationModel
+from core.decorators import to_model
 from core.schemas import CommonSchema
 
 
@@ -35,18 +36,9 @@ class GenreSchema(CommonSchema):
     name = fields.String(required=True)
     mediaId = fields.Integer(required=True)
 
-
+@to_model(ConfigurationModel)
 class ConfigurationSchema(CommonSchema):
     settings = fields.Nested(SettingsSchema(), required=True)
     image = fields.Nested(ImageSchema(), allow_none=True)
     navigation = fields.List(fields.Nested(nested=NavigationSchema(), allow_none=True))
     genres = fields.List(fields.Nested(nested=GenreSchema(), allow_none=True))
-
-    @post_load()
-    def __on_post_load(self, data, many, **kwargs) -> ConfigurationModel:
-        try:
-            model = ConfigurationModel.from_dict(data)
-            return model
-        except Exception as e:
-            self._logger.error(f"Conversion from dictionary failed", exc_info=e)
-            raise e

core/decorators.py

@@ -1,9 +1,40 @@
 import logging
-
+from typing import Dict, Type
+from serde import Model
 from uplink import error_handler
 
+from core.schemas import CommonSchema
+
 
 @error_handler(requires_consumer=True)
 def raise_api_error(consumer, exc_type=None, exc_val=None, exc_tb=None):
     logger = logging.getLogger("django")
-    logger.warning(f"API error occurred -> exc_type: {exc_type} exc_val: {exc_val} exc_tb: {exc_tb}")
+    logger.warning(
+        f"API error occurred -> exc_type: {exc_type} exc_val: {exc_val} exc_tb: {exc_tb}"
+    )
+
+
+def to_model(model_class: Type[Model]):
+    """
+    Decorator that adds functionality to schema classes to convert dict to model instance.
+
+    Args:
+        model_class: The model class to convert the dictionary data into
+    """
+
+    def decorator(schema_class: CommonSchema):
+        def on_post_serialization(
+            self: CommonSchema, data: Dict, many: bool, **kwargs
+        ) -> Model:
+            self._logger.debug(f"Executing post_load for {model_class.__name__}")
+            try:
+                model = model_class.from_dict(data)
+                return model
+            except Exception as e:
+                self._logger.error(f"Conversion from dictionary failed", exc_info=e)
+                raise e
+
+        schema_class._on_post_load = on_post_serialization
+        return schema_class
+
+    return decorator

core/schemas.py

@@ -3,7 +3,7 @@
 from logging import Logger
 from typing import Any, Optional, Dict, Union, cast
 
-from marshmallow import Schema
+from marshmallow import Schema, post_load
 from marshmallow.schema import SchemaMeta
 from marshmallow.types import StrSequenceOrSet
 
@@ -18,6 +18,7 @@ def __init__(self, *, only: Optional[StrSequenceOrSet] = None, exclude: StrSeque
         super().__init__(only=only, exclude=exclude, many=many, context=context, load_only=load_only,
                          dump_only=dump_only, partial=partial, unknown=unknown)
 
+    @post_load()
     def _on_post_load(self, data, many, **kwargs) -> Any:
         pass
 

core/tests/decorator_test.py

@@ -0,0 +1,58 @@
+import pytest
+import marshmallow
+import serde
+import serde.fields
+
+from core.decorators import to_model
+from core.schemas import CommonSchema
+
+class TestModel(serde.Model):
+    name = serde.fields.Str()
+    value = serde.fields.Int()
+
+@to_model(TestModel)
+class TestSchema(CommonSchema):
+    name = marshmallow.fields.String(required=True)
+    value = marshmallow.fields.Integer(required=True)
+
+
+class TestDecorator:
+    def test_successful_conversion(self):
+        test_data = {"name": "test", "value": 123}
+        schema = TestSchema()
+
+        result = schema.load(test_data)
+
+        assert isinstance(result, TestModel)
+        assert result.name == "test"
+        assert result.value == 123
+
+    def test_failed_conversion(self):
+        test_data = {"name": "test"}
+        schema = TestSchema()
+
+        with pytest.raises(Exception):
+            schema.load(test_data)
+
+    def test_multiple_schema_instances(self):
+        test_data = {"name": "test", "value": 123}
+        schema1 = TestSchema()
+        schema2 = TestSchema()
+
+        result1 = schema1.load(test_data)
+        result2 = schema2.load(test_data)
+
+        assert isinstance(result1, TestModel)
+        assert isinstance(result2, TestModel)
+        assert result1.name == result2.name
+        assert result1.value == result2.value
+
+    def test_successful_conversion_with_json(self):
+        test_data = '{"name": "test", "value": 123}'
+        schema = TestSchema()
+
+        result = schema.loads(test_data)
+
+        assert isinstance(result, TestModel)
+        assert result.name == "test"
+        assert result.value == 123

news/data/schemas.py

@@ -1,5 +1,6 @@
 from marshmallow import Schema, fields, post_load
 
+from core.decorators import to_model
 from news.domain.entities import NewsConnectionModel
 
 
@@ -13,17 +14,9 @@ class NewsSchema(Schema):
     published_on = fields.Integer(required=True, data_key="publishedOn")
     link = fields.String(required=True)
 
+@to_model(NewsConnectionModel)
 class NewsConnectionSchema(Schema):
     count = fields.Integer(required=True)
     first = fields.String(required=True)
     last = fields.String(required=True)
     data = fields.List(fields.Nested(NewsSchema), required=True)
-
-    @post_load()
-    def __on_post_load(self, data, many, **kwargs) -> NewsConnectionModel:
-        try:
-            model = NewsConnectionModel.from_dict(data)
-            return model
-        except Exception as e:
-            self._logger.error(f"Conversion from dictionary failed", exc_info=e)
-            raise e