cost-centers import rewrite

apps/fyle/views.py

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

@@ -0,0 +1,56 @@
+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,13 +2,6 @@
 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
@@ -26,17 +19,19 @@ def chain_import_fields_to_fyle(workspace_id):
             destination_field = 'ACCOUNT'
 
         chain.append(
-            'apps.mappings.imports.tasks.{}'.format(IMPORT_TASK_TARGET_MAP['CATEGORY']),
+            'apps.mappings.imports.tasks.trigger_import_via_schedule',
             workspace_id,
-            destination_field
+            destination_field,
+            'CATEGORY'
         )
 
     for mapping_setting in mapping_settings:
-        if mapping_setting.source_field in IMPORT_TASK_TARGET_MAP:
+        if mapping_setting.source_field in ['PROJECT', 'COST_CENTER']:
             chain.append(
-                'apps.mappings.imports.tasks.{}'.format(IMPORT_TASK_TARGET_MAP[mapping_setting.source_field]),
+               'apps.mappings.imports.tasks.trigger_import_via_schedule',
                 workspace_id,
-                mapping_setting.destination_field
+                mapping_setting.destination_field,
+                mapping_setting.source_field
             )
 
     if chain.length() > 0:

apps/mappings/imports/tasks.py

@@ -1,27 +1,24 @@
 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,
+}
 
-# TODO: Add a common function later when we add more modules
-def trigger_projects_import_via_schedule(workspace_id: int, destination_field: str):
+def trigger_import_via_schedule(workspace_id: int, destination_field: str, source_field: str):
     """
-    Trigger projects import via schedule
+    Trigger 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='PROJECT').first()
+    import_log = ImportLog.objects.filter(workspace_id=workspace_id, attribute_type=source_field).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()
 
-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()
+    module_class = SOURCE_FIELD_CLASS_MAP[source_field]
+    item = module_class(workspace_id, destination_field, sync_after)
+    item.trigger_import()

apps/mappings/signals.py

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

apps/mappings/tasks.py

@@ -248,125 +248,6 @@ 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,7 +502,6 @@ 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_count = Configuration.objects.filter(import_categories=True).count()
+        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)
         schedule_count = Schedule.objects.filter(func='apps.mappings.imports.queues.chain_import_fields_to_fyle').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("total_count: {}".format(total_count))
         print("schedule_count: {}".format(schedule_count))
         raise Exception("This is a sanity check")
 except Exception as e:

scripts/python/create-update-new-cost-centers-import.py

@@ -0,0 +1,34 @@
+from django.db import transaction
+from datetime import datetime
+from django_q.models import Schedule
+from apps.workspaces.models import Configuration
+from fyle_accounting_mappings.models import MappingSetting
+existing_import_enabled_schedules = Schedule.objects.filter(
+    func__in=['apps.mappings.tasks.auto_create_cost_center_mappings']
+).values('args')
+try:
+    with transaction.atomic():
+        for schedule in existing_import_enabled_schedules:
+            mapping_setting = MappingSetting.objects.filter(source_field='COST_CENTER', workspace_id=schedule['args'], import_to_fyle=True).first()
+            if mapping_setting:
+                Schedule.objects.update_or_create(
+                    func='apps.mappings.imports.queues.chain_import_fields_to_fyle',
+                    args=schedule['args'],
+                    defaults={
+                        'schedule_type': Schedule.MINUTES,
+                        'minutes':24 * 60,
+                        '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)
+        cost_center = MappingSetting.objects.filter(source_field='COST_CENTER', import_to_fyle=True).values_list('workspace_id', flat=True)
+        unique_states = list(set(categories) | set(project) | set(cost_center))
+        tot_count = len(unique_states)
+        schedule_count = Schedule.objects.filter(func='apps.mappings.imports.queues.chain_import_fields_to_fyle').count()
+        #make the sanity check a bit more clear
+        print("tot_count: {}".format(tot_count))
+        print("schedule_count: {}".format(schedule_count))
+        raise Exception("This is a sanity check")
+except Exception as e:
+    print(e)

tests/sql_fixtures/reset_db_fixtures/reset_db.sql

@@ -2,8 +2,8 @@
 -- PostgreSQL database dump
 --
 
--- 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)
+-- 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)
 
 SET statement_timeout = 0;
 SET lock_timeout = 0;