Adds Expand class to support more complex controls in the future

rest_flex_fields/__init__.py

@@ -34,3 +34,4 @@
 from .utils import *
 from .serializers import FlexFieldsModelSerializer
 from .views import FlexFieldsModelViewSet
+from .expand import Expand

rest_flex_fields/expand.py

@@ -0,0 +1,116 @@
+import copy
+import importlib
+from inspect import isclass
+from typing import Any, Optional, Tuple, Type, Union
+
+from django.db import models
+from rest_framework import serializers
+from rest_framework.fields import empty
+
+
+def import_serializer_class(
+    path: str, class_name: str
+) -> Tuple[Optional[Type[serializers.Serializer]], Optional[str]]:
+    try:
+        module = importlib.import_module(path)
+    except ImportError:
+        return (
+            None,
+            f"No module found at path: {path} when trying to import {class_name}",
+        )
+
+    try:
+        return getattr(module, class_name), None
+    except AttributeError:
+        return None, f"No class {path} class found in module {class_name}"
+
+
+class Expand(object):
+    """
+    Simple class to track how a field gets expanded.
+    Can be used directly in `Meta.expandable_fields` to provide
+    typed interface when setting up fields.
+    """
+
+    _serializer_ref: Union[Type[serializers.Serializer], str]
+    many: bool
+    source: Optional[str]
+    prefetch_model: Optional[Type[models.Model]] = None
+    auto_prefetch: bool
+    extra: dict
+
+    def __init__(
+        self,
+        serializer_ref: Union[Type[serializers.Serializer], str],
+        *,
+        many: bool = empty,
+        source: Optional[str] = None,
+        auto_prefetch=False,
+        prefetch_model: Optional[Type[models.Model]] = None,
+        extra: Optional[dict] = None,
+    ) -> None:
+        self._serializer_ref = serializer_ref
+        self.many = many
+        self.source = source
+        self.auto_prefetch = auto_prefetch
+        self.prefetch_model = prefetch_model
+        self.extra = extra or {}
+
+    def resolve_string_import(self) -> Type[serializers.Serializer]:
+        assert isinstance(self._serializer_ref, str)
+
+        path_parts = self._serializer_ref.split(".")
+        class_name = path_parts.pop()
+        path = ".".join(path_parts)
+
+        serializer_class, error = import_serializer_class(path, class_name)
+
+        if error and not path.endswith(".serializers"):
+            serializer_class, error = import_serializer_class(
+                path + ".serializers", class_name
+            )
+
+        if serializer_class:
+            return serializer_class
+
+        raise Exception(error)
+
+    def get_serializer_class(self) -> Type[serializers.Serializer]:
+        if isinstance(self._serializer_ref, str):
+            return self.resolve_string_import()
+        elif isclass(self._serializer_ref) or issubclass(
+            self._serializer_ref, serializers.Field
+        ):
+            return self._serializer_ref
+        else:
+            raise Exception("Could not determine serializer class.")
+
+    def get_base_settings(self) -> dict:
+        settings = {}
+
+        if self.many != empty:
+            settings["many"] = self.many
+
+        if self.source is not None:
+            settings["source"] = self.source
+
+        settings.update(copy.deepcopy(self.extra))
+        return settings
+
+    @classmethod
+    def init_from_legacy_def(cls, field_def: Any) -> "Expand":
+        if isinstance(field_def, tuple) and len(field_def) == 2:
+            serializer_ref, settings = field_def
+            return Expand(
+                serializer_ref,
+                source=settings.get("source"),
+                many=settings.get("many", False),
+            )
+        elif isinstance(field_def, str):
+            return Expand(field_def)
+        elif isclass(field_def) and issubclass(field_def, serializers.Field):
+            return Expand(field_def)
+        else:
+            raise Exception(
+                f"Error: {field_def} could not be cast to an Expand instance."
+            )

rest_flex_fields/serializers.py

@@ -1,6 +1,4 @@
-import copy
-import importlib
-from typing import List, Optional, Tuple
+from typing import Dict, List, Optional
 
 from rest_framework import serializers
 
@@ -13,6 +11,7 @@
     RECURSIVE_EXPANSION_PERMITTED,
     split_levels,
 )
+from rest_flex_fields.expand import Expand
 
 
 class FlexFieldsSerializerMixin(object):
@@ -23,6 +22,7 @@ class FlexFieldsSerializerMixin(object):
     values with complex, nested serializations
     """
 
+    _expandable_fields: Dict[str, Expand]
     expandable_fields = {}
     maximum_expansion_depth: Optional[int] = None
     recursive_expansion_permitted: Optional[bool] = None
@@ -33,7 +33,7 @@ def __init__(self, *args, **kwargs):
         omit = list(kwargs.pop(OMIT_PARAM, []))
         parent = kwargs.pop("parent", None)
 
-        super(FlexFieldsSerializerMixin, self).__init__(*args, **kwargs)
+        super().__init__(*args, **kwargs)
 
         self.parent = parent
         self.expanded_fields = []
@@ -62,6 +62,26 @@ def __init__(self, *args, **kwargs):
             + self._flex_options_rep_only["omit"],
         }
 
+    def __init_subclass__(cls) -> None:
+        """
+        Cast expandable field values to Expand instances to simplify
+        instantiating the correct serializer later.
+        """
+        expandable_fields = {}
+
+        if hasattr(cls, "Meta") and hasattr(cls.Meta, "expandable_fields"):
+            expandable_fields = cls.Meta.expandable_fields
+        elif hasattr(cls, "expandable_fields"):
+            expandable_fields = cls.expandable_fields
+
+        for field_name, field_spec in expandable_fields.items():
+            if isinstance(field_spec, Expand):
+                continue
+
+            expandable_fields[field_name] = Expand.init_from_legacy_def(field_spec)
+
+        cls._expandable_fields = expandable_fields
+
     def get_maximum_expansion_depth(self) -> Optional[int]:
         """
         Defined at serializer level or based on MAXIMUM_EXPANSION_DEPTH setting
@@ -115,19 +135,10 @@ def _make_expanded_field_serializer(
         """
         Returns an instance of the dynamically created nested serializer.
         """
-        field_options = self._expandable_fields[name]
+        expand_spec = self._expandable_fields[name]
 
-        if isinstance(field_options, tuple):
-            serializer_class = field_options[0]
-            settings = copy.deepcopy(field_options[1]) if len(field_options) > 1 else {}
-        else:
-            serializer_class = field_options
-            settings = {}
-
-        if type(serializer_class) == str:
-            serializer_class = self._get_serializer_class_from_lazy_string(
-                serializer_class
-            )
+        serializer_class = expand_spec.get_serializer_class()
+        settings = expand_spec.get_base_settings()
 
         if issubclass(serializer_class, serializers.Serializer):
             settings["context"] = self.context
@@ -146,39 +157,6 @@ def _make_expanded_field_serializer(
 
         return serializer_class(**settings)
 
-    def _get_serializer_class_from_lazy_string(self, full_lazy_path: str):
-        path_parts = full_lazy_path.split(".")
-        class_name = path_parts.pop()
-        path = ".".join(path_parts)
-        serializer_class, error = self._import_serializer_class(path, class_name)
-
-        if error and not path.endswith(".serializers"):
-            serializer_class, error = self._import_serializer_class(
-                path + ".serializers", class_name
-            )
-
-        if serializer_class:
-            return serializer_class
-
-        raise Exception(error)
-
-    def _import_serializer_class(
-        self, path: str, class_name: str
-    ) -> Tuple[Optional[str], Optional[str]]:
-        try:
-            module = importlib.import_module(path)
-        except ImportError:
-            return (
-                None,
-                "No module found at path: %s when trying to import %s"
-                % (path, class_name),
-            )
-
-        try:
-            return getattr(module, class_name), None
-        except AttributeError:
-            return None, "No class %s class found in module %s" % (path, class_name)
-
     def _get_fields_names_to_remove(
         self,
         current_fields: List[str],
@@ -258,16 +236,6 @@ def _get_expanded_field_names(
 
         return accum
 
-    @property
-    def _expandable_fields(self) -> dict:
-        """It's more consistent with DRF to declare the expandable fields
-        on the Meta class, however we need to support both places
-        for legacy reasons."""
-        if hasattr(self, "Meta") and hasattr(self.Meta, "expandable_fields"):
-            return self.Meta.expandable_fields
-
-        return self.expandable_fields
-
     def _get_query_param_value(self, field: str) -> List[str]:
         """
         Only allowed to examine query params if it's the root serializer.

tests/test_flex_fields_model_serializer.py

@@ -44,25 +44,25 @@ def test_clean_fields(self):
 
     def test_get_expanded_field_names_if_all(self):
         serializer = FlexFieldsModelSerializer()
-        serializer.expandable_fields = {"cat": "field", "dog": "field"}
+        serializer._expandable_fields = {"cat": "field", "dog": "field"}
         result = serializer._get_expanded_field_names("*", [], [], {})
         self.assertEqual(result, ["cat", "dog"])
 
     def test_get_expanded_names_but_not_omitted(self):
         serializer = FlexFieldsModelSerializer()
-        serializer.expandable_fields = {"cat": "field", "dog": "field"}
+        serializer._expandable_fields = {"cat": "field", "dog": "field"}
         result = serializer._get_expanded_field_names(["cat", "dog"], ["cat"], [], {})
         self.assertEqual(result, ["dog"])
 
     def test_get_expanded_names_but_only_sparse(self):
         serializer = FlexFieldsModelSerializer()
-        serializer.expandable_fields = {"cat": "field", "dog": "field"}
+        serializer._expandable_fields = {"cat": "field", "dog": "field"}
         result = serializer._get_expanded_field_names(["cat"], [], ["cat"], {})
         self.assertEqual(result, ["cat"])
 
     def test_get_expanded_names_including_omitted_when_defer_to_next_level(self):
         serializer = FlexFieldsModelSerializer()
-        serializer.expandable_fields = {"cat": "field", "dog": "field"}
+        serializer._expandable_fields = {"cat": "field", "dog": "field"}
         result = serializer._get_expanded_field_names(
             ["cat"], ["cat"], [], {"cat": ["age"]}
         )