Revert "cost-centers import rewrite (#345)"

apps/fyle/views.py

@@ -1,5 +1,5 @@
 from django.db.models import Q
-from datetime import datetime
+from datetime import datetime, timezone
 
 from rest_framework.views import status
 from rest_framework import generics
@@ -344,13 +344,14 @@ def post(self, request, *args, **kwargs):
             chain = Chain()
 
             for mapping_setting in mapping_settings:
-                if mapping_setting.source_field in ['PROJECT', 'COST_CENTER']:
+                if mapping_setting.source_field == 'PROJECT':
                     chain.append(
-                        'apps.mappings.imports.tasks.trigger_import_via_schedule',
+                        'apps.mappings.imports.tasks.trigger_projects_import_via_schedule',
                         int(kwargs['workspace_id']),
-                        mapping_setting.destination_field,
-                        mapping_setting.source_field
+                        mapping_setting.destination_field
                     )
+                elif mapping_setting.source_field == 'COST_CENTER':
+                    chain.append('apps.mappings.tasks.auto_create_cost_center_mappings', int(kwargs['workspace_id']))
                 elif mapping_setting.is_custom:
                     chain.append('apps.mappings.tasks.async_auto_create_custom_field_mappings',
                                 int(kwargs['workspace_id']))

apps/mappings/imports/modules/cost_centers.py

@@ -1,56 +0,0 @@
-from datetime import datetime
-from typing import List
-from apps.mappings.imports.modules.base import Base
-from fyle_accounting_mappings.models import DestinationAttribute
-
-
-class CostCenter(Base):
-    """
-    Class for CostCenter module
-    """
-    def __init__(self, workspace_id: int, destination_field: str, sync_after: datetime):
-        super().__init__(
-            workspace_id=workspace_id,
-            source_field='COST_CENTER',
-            destination_field=destination_field,
-            platform_class_name='cost_centers',
-            sync_after=sync_after
-        )
-
-    def trigger_import(self):
-        """
-        Trigger import for CostCenter module
-        """
-        self.check_import_log_and_start_import()
-
-    # remove the is_auto_sync_status_allowed parameter
-    def construct_fyle_payload(
-        self,
-        paginated_destination_attributes: List[DestinationAttribute],
-        existing_fyle_attributes_map: object,
-        is_auto_sync_status_allowed: bool
-    ):
-        """
-        Construct Fyle payload for CostCenter module
-        :param paginated_destination_attributes: List of paginated destination attributes
-        :param existing_fyle_attributes_map: Existing Fyle attributes map
-        :param is_auto_sync_status_allowed: Is auto sync status allowed
-        :return: Fyle payload
-        """
-        payload = []
-
-        for attribute in paginated_destination_attributes:
-            cost_center = {
-                'name': attribute.value,
-                'is_enabled': True if attribute.active is None else attribute.active,
-                'description': 'Cost Center - {0}, Id - {1}'.format(
-                    attribute.value,
-                    attribute.destination_id
-                )
-            }
-
-            # Create a new cost-center if it does not exist in Fyle
-            if attribute.value.lower() not in existing_fyle_attributes_map:
-                payload.append(cost_center)
-
-        return payload

apps/mappings/imports/queues.py

@@ -2,6 +2,13 @@
 from fyle_accounting_mappings.models import MappingSetting
 from apps.workspaces.models import Configuration
 
+
+IMPORT_TASK_TARGET_MAP = {
+    'PROJECT': 'trigger_projects_import_via_schedule',
+    'CATEGORY': 'trigger_categories_import_via_schedule',
+}
+
+
 def chain_import_fields_to_fyle(workspace_id):
     """
     Chain import fields to Fyle
@@ -19,19 +26,17 @@ def chain_import_fields_to_fyle(workspace_id):
             destination_field = 'ACCOUNT'
 
         chain.append(
-            'apps.mappings.imports.tasks.trigger_import_via_schedule',
+            'apps.mappings.imports.tasks.{}'.format(IMPORT_TASK_TARGET_MAP['CATEGORY']),
             workspace_id,
-            destination_field,
-            'CATEGORY'
+            destination_field
         )
 
     for mapping_setting in mapping_settings:
-        if mapping_setting.source_field in ['PROJECT', 'COST_CENTER']:
+        if mapping_setting.source_field in IMPORT_TASK_TARGET_MAP:
             chain.append(
-               'apps.mappings.imports.tasks.trigger_import_via_schedule',
+                'apps.mappings.imports.tasks.{}'.format(IMPORT_TASK_TARGET_MAP[mapping_setting.source_field]),
                 workspace_id,
-                mapping_setting.destination_field,
-                mapping_setting.source_field
+                mapping_setting.destination_field
             )
 
     if chain.length() > 0:

apps/mappings/imports/tasks.py

@@ -1,24 +1,27 @@
 from apps.mappings.models import ImportLog
 from apps.mappings.imports.modules.projects import Project
 from apps.mappings.imports.modules.categories import Category
-from apps.mappings.imports.modules.cost_centers import CostCenter
 
-SOURCE_FIELD_CLASS_MAP = {
-    'PROJECT': Project,
-    'CATEGORY': Category,
-    'COST_CENTER': CostCenter,
-}
 
-def trigger_import_via_schedule(workspace_id: int, destination_field: str, source_field: str):
+# TODO: Add a common function later when we add more modules
+def trigger_projects_import_via_schedule(workspace_id: int, destination_field: str):
     """
-    Trigger import via schedule
+    Trigger projects import via schedule
     :param workspace_id: Workspace id
     :param destination_field: Destination field
-    :param source_field: Type of attribute (e.g., 'PROJECT', 'CATEGORY', 'COST_CENTER')
     """
-    import_log = ImportLog.objects.filter(workspace_id=workspace_id, attribute_type=source_field).first()
+    import_log = ImportLog.objects.filter(workspace_id=workspace_id, attribute_type='PROJECT').first()
     sync_after = import_log.last_successful_run_at if import_log else None
+    project = Project(workspace_id, destination_field, sync_after)
+    project.trigger_import()
 
-    module_class = SOURCE_FIELD_CLASS_MAP[source_field]
-    item = module_class(workspace_id, destination_field, sync_after)
-    item.trigger_import()
+def trigger_categories_import_via_schedule(workspace_id: int, destination_field: str):
+    """
+    Trigger Categories import via schedule
+    :param workspace_id: Workspace id
+    :param destination_field: Destination field
+    """
+    import_log = ImportLog.objects.filter(workspace_id=workspace_id, attribute_type='CATEGORY').first()
+    sync_after = import_log.last_successful_run_at if import_log else None
+    category = Category(workspace_id, destination_field, sync_after)
+    category.trigger_import()

apps/mappings/signals.py

@@ -15,6 +15,7 @@
 from fyle.platform.exceptions import WrongParamsError
 
 from apps.mappings.tasks import (
+    schedule_cost_centers_creation,
     schedule_fyle_attributes_creation,
     upload_attributes_to_fyle
 )
@@ -113,7 +114,7 @@ def run_post_mapping_settings_triggers(sender, instance: MappingSetting, **kwarg
 
 
     if instance.source_field == 'COST_CENTER':
-        new_schedule_or_delete_fyle_import_tasks(configuration, instance)
+        schedule_cost_centers_creation(instance.import_to_fyle, int(instance.workspace_id))
 
     if instance.is_custom:
         schedule_fyle_attributes_creation(int(instance.workspace_id))

apps/mappings/tasks.py

@@ -248,6 +248,125 @@ def sync_sage_intacct_attributes(sageintacct_attribute_type: str, workspace_id:
         sage_intacct_connection.sync_user_defined_dimensions()
 
 
+def create_fyle_cost_centers_payload(sageintacct_attributes: List[DestinationAttribute], existing_fyle_cost_centers: list):
+    """
+    Create Fyle Cost Centers Payload from SageIntacct Objects
+    :param workspace_id: Workspace integer id
+    :param sageintacct_attributes: SageIntacct Objects
+    :param fyle_attribute: Fyle Attribute
+    :return: Fyle Cost Centers Payload
+    """
+    fyle_cost_centers_payload = []
+
+    for si_attribute in sageintacct_attributes:
+        if si_attribute.value not in existing_fyle_cost_centers:
+            fyle_cost_centers_payload.append({
+                'name': si_attribute.value,
+                'is_enabled': True if si_attribute.active is None else si_attribute.active,
+                'description': 'Cost Center - {0}, Id - {1}'.format(
+                    si_attribute.value,
+                    si_attribute.destination_id
+                )
+            })
+
+    return fyle_cost_centers_payload
+
+
+def post_cost_centers_in_batches(platform: PlatformConnector, workspace_id: int, sageintacct_attribute_type: str):
+    existing_cost_center_names = ExpenseAttribute.objects.filter(
+        attribute_type='COST_CENTER', workspace_id=workspace_id).values_list('value', flat=True)
+
+    si_attributes_count = DestinationAttribute.objects.filter(
+        attribute_type=sageintacct_attribute_type, workspace_id=workspace_id).count()
+
+    page_size = 200
+
+    for offset in range(0, si_attributes_count, page_size):
+        limit = offset + page_size
+        paginated_si_attributes = DestinationAttribute.objects.filter(
+            attribute_type=sageintacct_attribute_type, workspace_id=workspace_id).order_by('value', 'id')[offset:limit]
+
+        paginated_si_attributes = remove_duplicates(paginated_si_attributes)
+
+        fyle_payload: List[Dict] = create_fyle_cost_centers_payload(
+            paginated_si_attributes, existing_cost_center_names)
+
+        if fyle_payload:
+            platform.cost_centers.post_bulk(fyle_payload)
+            platform.cost_centers.sync()
+
+        Mapping.bulk_create_mappings(paginated_si_attributes, 'COST_CENTER', sageintacct_attribute_type, workspace_id)
+
+
+def auto_create_cost_center_mappings(workspace_id: int):
+    """
+    Create Cost Center Mappings
+    """
+    try:
+        fyle_credentials: FyleCredential = FyleCredential.objects.get(workspace_id=workspace_id)
+
+        platform = PlatformConnector(fyle_credentials=fyle_credentials)
+
+        mapping_setting = MappingSetting.objects.get(
+            source_field='COST_CENTER', import_to_fyle=True, workspace_id=workspace_id
+        )
+
+        platform.cost_centers.sync()
+
+        sync_sage_intacct_attributes(mapping_setting.destination_field, workspace_id)
+
+        post_cost_centers_in_batches(platform, workspace_id, mapping_setting.destination_field)
+
+    except (SageIntacctCredential.DoesNotExist, InvalidTokenError):
+        logger.info('Invalid Token or Sage Intacct credentials does not exist - %s', workspace_id)
+
+    except FyleInvalidTokenError:
+        logger.info('Invalid Token for fyle')
+
+    except InternalServerError:
+        logger.error('Internal server error while importing to Fyle')
+
+    except NoPrivilegeError:
+        logger.info('Insufficient permission to access the requested module')
+
+    except WrongParamsError as exception:
+        logger.error(
+            'Error while creating cost centers workspace_id - %s in Fyle %s %s',
+            workspace_id, exception.message, {'error': exception.response}
+        )
+
+    except Exception:
+        error = traceback.format_exc()
+        error = {
+            'error': error
+        }
+        logger.exception(
+            'Error while creating cost centers workspace_id - %s error: %s',
+            workspace_id, error
+        )
+
+
+def schedule_cost_centers_creation(import_to_fyle, workspace_id):
+    if import_to_fyle:
+        schedule, _ = Schedule.objects.update_or_create(
+            func='apps.mappings.tasks.auto_create_cost_center_mappings',
+            args='{}'.format(workspace_id),
+            defaults={
+                'schedule_type': Schedule.MINUTES,
+                'minutes': 24 * 60,
+                'next_run': datetime.now()
+            }
+        )
+    else:
+        schedule: Schedule = Schedule.objects.filter(
+            func='apps.mappings.tasks.auto_create_cost_center_mappings',
+            args='{}'.format(workspace_id)
+        ).first()
+
+        if schedule:
+            schedule.delete()
+
+
 def construct_custom_field_placeholder(source_placeholder: str, fyle_attribute: str, existing_attribute: Dict):
     new_placeholder = None
     placeholder = None

apps/sage_intacct/utils.py

@@ -502,6 +502,7 @@ def sync_classes(self):
         count = self.connection.classes.count()
         if count <= SYNC_UPPER_LIMIT['classes']:
             classes = self.connection.classes.get_all(field='STATUS', value='active', fields=['NAME', 'CLASSID'])
+
             class_attributes = []
 
             for _class in classes:

scripts/python/create-update-new-category-import.py

@@ -20,12 +20,12 @@
                         'next_run':datetime.now()
                     }
                 )
-        categories = Configuration.objects.filter(import_categories=True).values_list('workspace_id', flat=True)
-        project = MappingSetting.objects.filter(source_field='PROJECT', import_to_fyle=True).values_list('workspace_id', flat=True)
-        unique_workspace_ids = list(set(categories) | set(project))
-        total_count = len(unique_workspace_ids)
+        categories_count = Configuration.objects.filter(import_categories=True).count()
         schedule_count = Schedule.objects.filter(func='apps.mappings.imports.queues.chain_import_fields_to_fyle').count()
-        print("total_count: {}".format(total_count))
+        project_count = MappingSetting.objects.filter(source_field='PROJECT', import_to_fyle=True).count()
+        #make the sanity check a bit more clear
+        print("categoreis_count: {}".format(categories_count))
+        print("project_count: {}".format(project_count))
         print("schedule_count: {}".format(schedule_count))
         raise Exception("This is a sanity check")
 except Exception as e:

tests/sql_fixtures/reset_db_fixtures/reset_db.sql

@@ -2,8 +2,8 @@
 -- PostgreSQL database dump
 --
 
--- Dumped from database version 15.4 (Debian 15.4-1.pgdg120+1)
--- Dumped by pg_dump version 15.4 (Debian 15.4-1.pgdg100+1)
+-- Dumped from database version 15.3 (Debian 15.3-1.pgdg120+1)
+-- Dumped by pg_dump version 15.3 (Debian 15.3-1.pgdg100+1)
 
 SET statement_timeout = 0;
 SET lock_timeout = 0;