fix: remove relations fields from the exported Excel data (#160)

.coveragerc

@@ -6,6 +6,7 @@ omit=fahari/config/*,fahari/**/migrations/*,scripts/*,fahari/**/tests/*
 [report]
 exclude_lines =
     # Have to re-enable the standard pragma
+    pragma: no branch
     pragma: nocover
     pragma: no cover
 

fahari/common/views/mixins/drf_mixins.py

@@ -5,6 +5,7 @@
 from openpyxl import Workbook
 from rest_framework import status
 from rest_framework.decorators import action
+from rest_framework.relations import ManyRelatedField, PrimaryKeyRelatedField
 from rest_framework.renderers import StaticHTMLRenderer
 from rest_framework.request import Request
 from rest_framework.response import Response
@@ -172,6 +173,9 @@ def retrieve_available_fields(self, request: Request) -> List[Dict[str, Any]]:
         def visit(fields: Dict[str, Any], parent_key="") -> List[Dict[str, Union[str, Any]]]:
             results = []
             for key, val in fields.items():
+                if isinstance(val, (ManyRelatedField, PrimaryKeyRelatedField)):
+                    # Skip primary key fields and many to many fields
+                    continue
                 new_key = parent_key + nested_entries_delimiter + key if parent_key else key
                 new_value = {
                     "icon": "fas fa-folder",

fahari/utils/excel_utils/drf_serializer_excel_io.py

@@ -21,6 +21,7 @@
 from openpyxl.utils.cell import get_column_letter
 from openpyxl.worksheet.worksheet import Worksheet
 from rest_framework.fields import Field
+from rest_framework.relations import ManyRelatedField, PrimaryKeyRelatedField
 from rest_framework.serializers import Serializer
 
 from fahari.common.serializers import AuditFieldsMixin
@@ -37,6 +38,27 @@
 _EXCEL_TYPES = (bool, int, float, str)
 
 
+def flatten_fields(fields: _NFD, nested_entries_delimiter: str) -> Dict[str, Field]:
+    """Flatten the nested fields in the given dict using the given delimiter.
+
+    This is needed as `DRFSerializerExcelIOTemplate` doesn't works with nested data.
+    """
+
+    # Visit each of the dump_fields and extract the field names to be used as
+    # column headers. If there are nested fields, flatten them.
+    def visit(source_fields: _NFD, parent_key="") -> Dict[str, Field]:
+        results: List[Tuple[str, Field]] = []
+        for key, val in source_fields.items():
+            new_key = parent_key + nested_entries_delimiter + key if parent_key else key
+            if isinstance(val, dict):
+                results.extend(visit(val, new_key).items())  # noqa
+            else:
+                results.append((new_key, val))
+        return OrderedDict(results)
+
+    return visit(fields)
+
+
 class DRFSerializerExcelIO(Generic[S, T], ExcelIO[DT]):
     """`ExcelIO` implementation powered by DRF `Serializer`."""
 
@@ -67,9 +89,7 @@ def context(self) -> Dict[str, Any]:
     def dump_data(self, data: DT, progress_callback: ProgressCallback = None) -> Workbook:
         dump_fields = self._pick_dump_fields()
         nested_entries_delimiter = self.get_nested_entries_delimiter()
-        template = self.get_template(
-            fields=self._flatten_fields(dump_fields, nested_entries_delimiter)
-        )
+        template = self.get_template(fields=flatten_fields(dump_fields, nested_entries_delimiter))
         wb = Workbook()
         template.render(
             self._clean_dump_data(data, dump_fields),
@@ -141,8 +161,10 @@ def get_template(self, *args, **kwargs) -> T:
 
         template_class = self.get_template_class()
         kwargs.setdefault("serializer", self.get_serializer())
-        if "fields" not in kwargs:
-            kwargs["fields"] = self.get_fields()  # This is expensive so only call it when needed
+        if "fields" not in kwargs:  # This is expensive so only call it when needed
+            kwargs["fields"] = flatten_fields(
+                self.get_fields(), self.get_nested_entries_delimiter()
+            )
 
         return template_class(*args, **kwargs)  # type: ignore
 
@@ -174,6 +196,11 @@ def visit(entry: Dict[str, Any], fields: _NFD, parent_key="") -> Dict[str, Any]:
         return [visit(entry, dump_fields) for entry in data]
 
     def _pick_dump_fields(self) -> _NFD:
+        """Return a dict consisting of only the selected dump fields.
+
+        If no dump fields have been provided, then returns all the fields as is.
+        """
+
         all_fields = self.get_fields()
         if not self._dump_fields:
             return all_fields
@@ -205,23 +232,6 @@ def visit_vertically(dump_fields: Sequence[str], fields: _NFD) -> _NFD:
 
         return visit_vertically(self._dump_fields, all_fields)
 
-    @staticmethod
-    def _flatten_fields(fields: _NFD, nested_entries_delimiter: str) -> Dict[str, Field]:
-
-        # Visit each of the dump_fields and extract the field names to be used as
-        # column headers. If there are nested fields, flatten them.
-        def visit(source_fields: _NFD, parent_key="") -> Dict[str, Field]:
-            results: List[Tuple[str, Field]] = []
-            for key, val in source_fields.items():
-                new_key = parent_key + nested_entries_delimiter + key if parent_key else key
-                if isinstance(val, dict):
-                    results.extend(visit(val, new_key).items())  # noqa
-                else:
-                    results.append((new_key, val))
-            return OrderedDict(results)
-
-        return visit(fields)
-
 
 class DRFSerializerExcelIOTemplate(Generic[S], ExcelIOTemplate[DT]):
     """A template that works with DRFSerializerExcelIO objects."""
@@ -263,10 +273,22 @@ def generate_input_template(
     ) -> None:
         raise NotImplementedError("`generate_input_template` must be implemented.")
 
-    def get_column_headers(self) -> Sequence[str]:
+    def get_column_headers(self, for_input=False) -> Sequence[str]:
         """Return the headers for each of the columns in the final exported file."""
 
-        return tuple(self.get_fields().keys())
+        if for_input:  # pragma: no branch
+            return tuple(self.get_fields().keys())
+
+        # If this is a dump, ignore write only, primary related and
+        # many to many fields.
+        fields = self.get_fields()
+        return tuple(
+            field_name
+            for field_name, field in fields.items()
+            if not (
+                field.write_only or isinstance(field, (PrimaryKeyRelatedField, ManyRelatedField))
+            )
+        )
 
     def get_fields(self) -> Dict[str, Field]:
         return self._fields
@@ -289,11 +311,25 @@ def read(self, workbook: Workbook, progress_callback: Optional[ProgressCallback]
     def render(
         self, data: DT, workbook: Workbook, progress_callback: Optional[ProgressCallback] = None
     ) -> None:
-        self._setup_workbook(workbook)
+        self._setup_workbook(workbook, False)
         self._setup_data_worksheet(workbook[self.DATA_WORKSHEET_NAME], False)
         self._dump_data(data, workbook[self.DATA_WORKSHEET_NAME])
         self._auto_size_columns(workbook[self.DATA_WORKSHEET_NAME])
 
+    def _dump_data(self, data: DT, worksheet: Worksheet) -> None:
+        fields = self.get_fields()
+        for entry in data:
+            worksheet.append(
+                self._coax_to_excel_value(value)
+                for value, field in zip(entry.values(), fields.values())
+                if not (
+                    # For a data dump, ignore write only, primary related
+                    # and many to many fields.
+                    field.write_only
+                    or isinstance(field, (PrimaryKeyRelatedField, ManyRelatedField))
+                )
+            )
+
     def _setup_data_worksheet(self, worksheet: Worksheet, for_input=False) -> None:
         header_row = self.get_column_headers()
 
@@ -303,12 +339,12 @@ def _setup_data_worksheet(self, worksheet: Worksheet, for_input=False) -> None:
         )
         self._freeze_column_headers(worksheet, len(header_row))
 
-    def _setup_workbook(self, workbook: Workbook) -> None:
+    def _setup_workbook(self, workbook: Workbook, for_input=False) -> None:
         work_sheets = workbook.sheetnames  # Remove existing worksheets
         for sheet_name in work_sheets:
             del workbook[sheet_name]
 
-        ws: Worksheet = workbook.create_sheet(title=self.DATA_WORKSHEET_NAME, index=0)
+        ws: Worksheet = workbook.create_sheet(title=self.DATA_WORKSHEET_NAME, index=0)  # noqa
         workbook.active = ws
         workbook.create_sheet(title=self.SCHEMA_WORKSHEET_NAME)
 
@@ -334,12 +370,6 @@ def _coax_to_excel_value(value: Any) -> Union[bool, float, int, str]:
             return value
         return "" if value is None else str(value)
 
-    @staticmethod
-    def _dump_data(data: DT, worksheet: Worksheet) -> None:
-        template = DRFSerializerExcelIOTemplate
-        for entry in data:
-            worksheet.append(template._coax_to_excel_value(value) for value in entry.values())
-
     @staticmethod
     def _freeze_column_headers(
         worksheet: Worksheet, no_of_column_headers: int, headers_row_index=1

fahari/utils/tests/test_drf_serializer_excel_io.py

@@ -15,6 +15,7 @@
     DRFSerializerExcelIO,
     DRFSerializerExcelIOTemplate,
 )
+from fahari.utils.excel_utils.drf_serializer_excel_io import flatten_fields
 
 
 class AuditSerializerExcelIOTestCase(TestCase):
@@ -154,15 +155,15 @@ def setUp(self) -> None:
     def test_correct_object_creation(self) -> None:
         """Assert that object creation and initialization produces the expected object."""
 
+        delimiter = self.excel_io.get_nested_entries_delimiter()
         excel_io_template = DRFSerializerExcelIOTemplate(
-            fields=self.excel_io.get_fields(), serializer=self.excel_io.get_serializer()
+            fields=flatten_fields(self.excel_io.get_fields(), delimiter),
+            serializer=self.excel_io.get_serializer(),
         )
 
         assert excel_io_template.get_column_headers() is not None
         assert excel_io_template.get_fields() is not None
-        assert excel_io_template.get_fields().keys() == self.excel_io.get_fields().keys()
         assert excel_io_template.get_serializer() is not None
-        assert self.excel_io.get_fields().keys() <= set(excel_io_template.get_column_headers())
 
     def test_generate_input_template(self) -> None:
         """Assert that input template generation works as expected.