[fix] Fixes by @coderabbitai

Author: pandafy

openwisp_users/management/commands/export_users.py

@@ -3,6 +3,7 @@
 from django.contrib.auth import get_user_model
 from django.core.exceptions import ObjectDoesNotExist
 from django.core.management.base import BaseCommand, CommandError
+from django.db.models.manager import BaseManager
 from django.db.models.query import QuerySet
 from django.utils.translation import gettext_lazy as _
 
@@ -21,6 +22,10 @@ def normalize_field(field):
 class Command(BaseCommand):
     help = _("Exports user data to a CSV file")
 
+    def _normalize_value(self, value):
+        """Convert None to empty string, otherwise stringify the value."""
+        return "" if value is None else str(value)
+
     def add_arguments(self, parser):
         parser.add_argument(
             "--exclude-fields",
@@ -73,7 +78,7 @@ def handle(self, *args, **options):
                 row = []
                 for field in fields:
                     val = self._get_field_value(user, field)
-                    row.append(val if val is not None else "")
+                    row.append(val)
                 csv_writer.writerow(row)
         self.stdout.write(
             self.style.SUCCESS(
@@ -98,7 +103,7 @@ def serialize_related(self, manager, subfields):
             row = []
             for f in subfields:
                 val = self._get_nested_attr(obj, f)
-                row.append("" if val is None else str(val))
+                row.append(self._normalize_value(val))
             rows.append(row)
         if not rows:
             return ""
@@ -126,17 +131,15 @@ def _get_nested_attr(self, obj, attr_path):
                 # missing attribute or intermediate raises -> None sentinel
                 return None
             # Detect querysets/related managers robustly.
-            if (isinstance(current, QuerySet) or hasattr(current, "all")) and i < len(
-                parts
-            ) - 1:
+            if (isinstance(current, (QuerySet, BaseManager))) and (i < len(parts) - 1):
                 remaining_path = ".".join(parts[i + 1 :])
                 # We use current.all() instead of current.iterator() to utilize
                 # the prefetch_related queryset cache. The iterator() method
                 # would bypass the cache and cause additional queries.
                 values = []
                 for item in current.all():
                     v = self._get_nested_attr(item, remaining_path)
-                    values.append("" if v is None else str(v))
+                    values.append(self._normalize_value(v))
                 return ",".join(values)
         return current
 
@@ -148,19 +151,23 @@ def _get_field_value(self, user, field):
         # Priority: callable > fields > name
         if callable_fn is not None:
             try:
-                return callable_fn(user)
+                val = callable_fn(user)
             except Exception as e:
                 func_name = getattr(callable_fn, "__name__", repr(callable_fn))
                 raise CommandError(
                     _(
                         "Error calling function {func_name!r} for field '{name}': {e}"
                     ).format(func_name=func_name, name=name, e=e)
                 )
+            return self._normalize_value(val)
         if subfields is not None:
             attr = self._get_nested_attr(user, name)
             if attr is None:
                 return ""
-            if isinstance(attr, QuerySet) or hasattr(attr, "all"):
+            if isinstance(attr, (QuerySet, BaseManager)):
                 return self.serialize_related(attr, subfields)
-            return ",".join(str(self._get_nested_attr(attr, f)) for f in subfields)
-        return self._get_nested_attr(user, name)
+            return ",".join(
+                self._normalize_value(self._get_nested_attr(attr, f)) for f in subfields
+            )
+        val = self._get_nested_attr(user, name)
+        return self._normalize_value(val)

openwisp_users/tests/test_commands.py

@@ -139,7 +139,7 @@ def test_related_fields(self):
             csv_reader = csv.reader(temp_file)
             csv_data = list(csv_reader)
 
-        # 3 user and 1 header
+        # 1 user and 1 header
         self.assertEqual(len(csv_data), 2)
         # When fields are ["id", "auth_token.key"] the expected headers
         # are the literal keys used to identify columns in the CSV.
@@ -264,4 +264,4 @@ class FakeUser:
             intermediate = FakeIntermediate()
 
         result = Command()._get_field_value(FakeUser(), "intermediate.sub_field")
-        self.assertEqual(result, None)
+        self.assertEqual(result, "")