refactor code

onadata/apps/main/models/meta_data.py

@@ -6,9 +6,12 @@
 from __future__ import unicode_literals
 
 import hashlib
+import importlib
+import json
 import logging
 import mimetypes
 import os
+from collections import OrderedDict
 from contextlib import closing
 
 from django.conf import settings
@@ -30,6 +33,7 @@
     safe_delete,
 )
 from onadata.libs.utils.common_tags import (
+    EXPORT_COLUMNS_REGISTER,
     GOOGLE_SHEET_DATA_TYPE,
     TEXTIT,
     TEXTIT_DETAILS,
@@ -96,13 +100,22 @@ def get_default_content_type():
     return content_object.id
 
 
-def unique_type_for_form(content_object, data_type, data_value=None, data_file=None):
+def unique_type_for_form(
+    content_object, data_type, data_value=None, data_file=None, extra_data=None
+):
     """
     Ensure that each metadata object has unique xform and data_type fields
 
     return the metadata object
     """
-    defaults = {"data_value": data_value} if data_value else {}
+    defaults = {}
+
+    if data_value:
+        defaults["data_value"] = data_value
+
+    if extra_data:
+        defaults["extra_data"] = extra_data
+
     content_type = ContentType.objects.get_for_model(content_object)
 
     if data_value is None and data_file is None:
@@ -111,7 +124,7 @@ def unique_type_for_form(content_object, data_type, data_value=None, data_file=N
             object_id=content_object.id, content_type=content_type, data_type=data_type
         ).first()
     else:
-        result, _created = MetaData.objects.update_or_create(
+        result, _ = MetaData.objects.update_or_create(
             object_id=content_object.id,
             content_type=content_type,
             data_type=data_type,
@@ -570,6 +583,27 @@ def instance_csv_imported_by(content_object, data_value=None):
         data_type = "imported_via_csv_by"
         return unique_type_for_form(content_object, data_type, data_value)
 
+    @staticmethod
+    def export_columns_register(content_object, data_value=None):
+        """Update or create export columns register for XForm."""
+        # Avoid cyclic import by using importlib
+        csv_builder = importlib.import_module("onadata.libs.utils.csv_builder")
+        ordered_columns = OrderedDict()
+        # pylint: disable=protected-access
+        csv_builder.CSVDataFrameBuilder._build_ordered_columns(
+            content_object._get_survey(), ordered_columns
+        )
+        serialized_columns = json.dumps(ordered_columns)
+        data_type = EXPORT_COLUMNS_REGISTER
+        extra_data = {
+            "merged_multiples": serialized_columns,
+            "split_multiples": serialized_columns,
+        }
+        data_value = "" if data_value is None else data_value
+        return unique_type_for_form(
+            content_object, data_type, data_value=data_value, extra_data=extra_data
+        )
+
 
 # pylint: disable=unused-argument,invalid-name
 def clear_cached_metadata_instance_object(

onadata/apps/viewer/models/data_dictionary.py

@@ -444,8 +444,8 @@ def create_or_update_export_register(sender, instance=None, created=False, **kwa
     """Create or update export columns register for the form"""
     # Avoid cyclic import by using importlib
     logger_tasks = importlib.import_module("onadata.apps.logger.tasks")
-    logger_tools = importlib.import_module("onadata.libs.utils.logger_tools")
-    logger_tools.update_or_create_export_register(instance)
+
+    MetaData.export_columns_register(instance)
 
     if not created:
         logger_tasks.reconstruct_xform_export_register_async.delay(instance.pk)

onadata/libs/utils/logger_tools.py

@@ -1605,39 +1605,13 @@ def register_instance_repeat_columns(instance: Instance) -> None:
     _register_instance_repeat_columns(instance, register)
 
 
-def update_or_create_export_register(xform: XForm) -> tuple[MetaData, bool]:
-    """Update or create export columns register for the form"""
-    # Avoid cyclic import by using importlib
-    csv_builder = importlib.import_module("onadata.libs.utils.csv_builder")
-    ordered_columns = OrderedDict()
-    # pylint: disable=protected-access
-    csv_builder.CSVDataFrameBuilder._build_ordered_columns(
-        xform._get_survey(), ordered_columns
-    )
-    serialized_columns = json.dumps(ordered_columns)
-    register, created = MetaData.objects.update_or_create(
-        content_type=ContentType.objects.get_for_model(xform),
-        object_id=xform.pk,
-        data_type=EXPORT_COLUMNS_REGISTER,
-        defaults={
-            "data_value": "",
-            "extra_data": {
-                "merged_multiples": serialized_columns,
-                "split_multiples": serialized_columns,
-            },
-        },
-    )
-
-    return register, created
-
-
 @transaction.atomic()
 def reconstruct_xform_export_register(xform: XForm) -> None:
     """Reconstruct the export columns register for an XForm
 
     :param xform: XForm object
     """
-    register, _ = update_or_create_export_register(xform)
+    register = MetaData.export_columns_register(xform)
     instance_qs = xform.instances.filter(deleted_at__isnull=True)
 
     for instance in queryset_iterator(instance_qs, chunksize=500):