fix: improve response times of sync_dimension calls (#537)

* wip: attempt to run sync_dimensions functions async * fix: delegate sync_dimensions tasks to workers * fix(sage_intacct): extract all sync_dimensions POST handlers into async functions this reduces local response time to ≤ 1 second * fix(apps/fyle): extract all sync_dimensions POST handlers into async functions this reduces local response time to ≤ 1 second * fix: update tests * refactor: remove unused import Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> * test: add unit test for handle_refresh_dimensions * test: add fixture to test * fix(apps/sage_intacct): add back intacct creds check, helper logic refactor refactor: remove handle_refresh_dimensions and move logic to the `post()` handler refactor: update destination_synced_at within sync_dimensions * test: add back sync_dimensions and refresh_dimensions invalid creds tests * fix: dont call `async_update_workspace_name` when polling `workspaces/` * refactor: remove print statement * test: add unit test for `handle_refresh_dimensions` * test: add unit test for `check_interval_and_sync_dimension` --------- Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> Co-authored-by: anishfyle <[email protected]> (cherry picked from commit 715f85b)
fylein · Aug 14, 2024 · ae67b01 · ae67b01
1 parent 50c9045
commit ae67b01
Show file tree

Hide file tree

Showing 7 changed files with 113 additions and 65 deletions.
diff --git a/apps/fyle/helpers.py b/apps/fyle/helpers.py
@@ -8,7 +8,8 @@
 
 from django.conf import settings
 from django.db.models import Q
-from fyle_accounting_mappings.models import ExpenseAttribute
+from django_q.tasks import Chain
+from fyle_accounting_mappings.models import ExpenseAttribute, MappingSetting
 from rest_framework.exceptions import ValidationError
 
 from apps.fyle.models import ExpenseFilter, ExpenseGroup, ExpenseGroupSettings, Expense
@@ -129,22 +130,22 @@ def add_expense_id_to_expense_group_settings(workspace_id: int):
     expense_group_settings.save()
 
 
-def check_interval_and_sync_dimension(workspace: Workspace, fyle_credentials: FyleCredential) -> bool:
+def check_interval_and_sync_dimension(workspace_id, **kwargs) -> bool:
     """
     Check sync interval and sync dimension
-    :param workspace: Workspace Instance
-    :param fyle_credentials: Fyle credentials of an org
-    return: True/False based on sync
+    :param workspace_id: Workspace ID
     """
+
+    workspace = Workspace.objects.get(pk=workspace_id)
+    fyle_credentials = FyleCredential.objects.get(workspace_id=workspace.id)
+
     if workspace.source_synced_at:
         time_interval = datetime.now(timezone.utc) - workspace.source_synced_at
 
     if workspace.source_synced_at is None or time_interval.days > 0:
         sync_dimensions(fyle_credentials)
-        return True
-
-    return False
-
+        workspace.source_synced_at = datetime.now()
+        workspace.save(update_fields=['source_synced_at'])
 
 def sync_dimensions(fyle_credentials, is_export: bool = False):
     platform = PlatformConnector(fyle_credentials)
@@ -172,6 +173,33 @@ def sync_dimensions(fyle_credentials, is_export: bool = False):
         if projects_count != projects_expense_attribute_count:
             platform.projects.sync()
 
+def handle_refresh_dimensions(workspace_id):
+    workspace = Workspace.objects.get(id=workspace_id)
+    fyle_credentials = FyleCredential.objects.get(workspace_id=workspace.id)
+
+    mapping_settings = MappingSetting.objects.filter(workspace_id=workspace_id, import_to_fyle=True)
+    chain = Chain()
+
+    for mapping_setting in mapping_settings:
+        if mapping_setting.source_field in ['PROJECT', 'COST_CENTER'] or mapping_setting.is_custom:
+            chain.append(
+                'apps.mappings.imports.tasks.trigger_import_via_schedule',
+                int(workspace_id),
+                mapping_setting.destination_field,
+                mapping_setting.source_field,
+                mapping_setting.is_custom,
+                q_options={'cluster': 'import'}
+            )
+
+    if chain.length() > 0:
+        chain.run()
+
+
+    sync_dimensions(fyle_credentials)
+
+    workspace.source_synced_at = datetime.now()
+    workspace.save(update_fields=['source_synced_at'])
+
 
 def construct_expense_filter(expense_filter):
     constructed_expense_filter = {}

diff --git a/apps/fyle/views.py b/apps/fyle/views.py
@@ -5,7 +5,7 @@
 from rest_framework import generics
 from rest_framework.response import Response
 
-from django_q.tasks import Chain
+from django_q.tasks import Chain, async_task
 
 from django_filters.rest_framework import DjangoFilterBackend
 from fyle_accounting_mappings.models import ExpenseAttribute, MappingSetting
@@ -282,13 +282,14 @@ def post(self, request, *args, **kwargs):
         Sync data from Fyle
         """
         try:
+            # check if fyle credentials are present, and return 400 otherwise
             workspace = Workspace.objects.get(pk=kwargs['workspace_id'])
-            fyle_credentials = FyleCredential.objects.get(workspace_id=workspace.id)
+            FyleCredential.objects.get(workspace_id=workspace.id)
 
-            synced = check_interval_and_sync_dimension(workspace, fyle_credentials)
-            if synced:
-                workspace.source_synced_at = datetime.now()
-                workspace.save(update_fields=['source_synced_at'])
+            async_task(
+                'apps.fyle.helpers.check_interval_and_sync_dimension',
+                kwargs['workspace_id']
+            )
 
             return Response(
                 status=status.HTTP_200_OK
@@ -318,31 +319,11 @@ def post(self, request, *args, **kwargs):
         Sync data from Fyle
         """
         try:
+            # check if fyle credentials are present, and return 400 otherwise
             workspace = Workspace.objects.get(id=kwargs['workspace_id'])
-            fyle_credentials = FyleCredential.objects.get(workspace_id=workspace.id)
-
-            mapping_settings = MappingSetting.objects.filter(workspace_id=kwargs['workspace_id'], import_to_fyle=True)
-            chain = Chain()
-
-            for mapping_setting in mapping_settings:
-                if mapping_setting.source_field in ['PROJECT', 'COST_CENTER'] or mapping_setting.is_custom:
-                    chain.append(
-                        'apps.mappings.imports.tasks.trigger_import_via_schedule',
-                        int(kwargs['workspace_id']),
-                        mapping_setting.destination_field,
-                        mapping_setting.source_field,
-                        mapping_setting.is_custom,
-                        q_options={'cluster': 'import'}
-                    )
-
-            if chain.length() > 0:
-                chain.run()
-
-
-            sync_dimensions(fyle_credentials)
+            FyleCredential.objects.get(workspace_id=workspace.id)
 
-            workspace.source_synced_at = datetime.now()
-            workspace.save(update_fields=['source_synced_at'])
+            async_task('apps.fyle.helpers.handle_refresh_dimensions', kwargs['workspace_id'])
 
             return Response(
                 status=status.HTTP_200_OK

diff --git a/apps/sage_intacct/helpers.py b/apps/sage_intacct/helpers.py
@@ -40,24 +40,23 @@ def schedule_payment_sync(configuration: Configuration):
     )
 
 
-def check_interval_and_sync_dimension(workspace: Workspace, si_credentials: SageIntacctCredential) -> bool:
+def check_interval_and_sync_dimension(workspace_id, **kwargs) -> bool:
     """
     Check sync interval and sync dimensions
-    :param workspace: Workspace Instance
-    :param si_credentials: SageIntacctCredentials Instance
-
-    return: True/False based on sync
+    :param workspace_id: Workspace ID
     """
 
+    workspace = Workspace.objects.get(pk=workspace_id)
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace.id)
+
+
     if workspace.destination_synced_at:
         time_interval = datetime.now(timezone.utc) - workspace.source_synced_at
 
     if workspace.destination_synced_at is None or time_interval.days > 0:
-        sync_dimensions(si_credentials, workspace.id)
-        return True
-
-    return False
-
+        sync_dimensions(sage_intacct_credentials, workspace.id)
+        workspace.destination_synced_at = datetime.now()
+        workspace.save(update_fields=['destination_synced_at'])
 
 def is_dependent_field_import_enabled(workspace_id: int) -> bool:
     return DependentFieldSetting.objects.filter(workspace_id=workspace_id).exists()
@@ -67,16 +66,26 @@ def sync_dimensions(si_credentials: SageIntacctCredential, workspace_id: int, di
     sage_intacct_connection = import_string(
         'apps.sage_intacct.utils.SageIntacctConnector'
     )(si_credentials, workspace_id)
+
+    update_timestamp = False
     if not dimensions:
         dimensions = [
             'locations', 'customers', 'departments', 'tax_details', 'projects', 
             'expense_payment_types', 'classes', 'charge_card_accounts','payment_accounts', 
             'vendors', 'employees', 'accounts', 'expense_types', 'items', 'user_defined_dimensions', 'allocations'
         ]
+        update_timestamp = True
 
     for dimension in dimensions:
         try:
             sync = getattr(sage_intacct_connection, 'sync_{}'.format(dimension))
             sync()
         except Exception as exception:
             logger.info(exception)
+
+    if update_timestamp:
+        # Update destination_synced_at to current time only when full refresh happens
+        workspace = Workspace.objects.get(pk=workspace_id)
+
+        workspace.destination_synced_at = datetime.now()
+        workspace.save(update_fields=['destination_synced_at'])
diff --git a/apps/sage_intacct/views.py b/apps/sage_intacct/views.py
@@ -12,6 +12,8 @@
 
 from sageintacctsdk.exceptions import InvalidTokenError
 
+from django_q.tasks import async_task
+
 from apps.workspaces.models import SageIntacctCredential, Workspace, Configuration
 
 from .helpers import sync_dimensions, check_interval_and_sync_dimension
@@ -172,13 +174,12 @@ def post(self, request, *args, **kwargs):
 
         try:
             workspace = Workspace.objects.get(pk=kwargs['workspace_id'])
-            sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace.id)
-
-            synced = check_interval_and_sync_dimension(workspace, sage_intacct_credentials)
+            SageIntacctCredential.objects.get(workspace_id=workspace.id)
 
-            if synced:
-                workspace.destination_synced_at = datetime.now()
-                workspace.save(update_fields=['destination_synced_at'])
+            async_task(
+                'apps.sage_intacct.helpers.check_interval_and_sync_dimension',
+                kwargs['workspace_id'], 
+            )
 
             return Response(
                 status=status.HTTP_200_OK
@@ -203,17 +204,21 @@ def post(self, request, *args, **kwargs):
         """
         Sync data from sage intacct
         """
+        dimensions_to_sync = request.data.get('dimensions_to_sync', [])
+
         try:
-            dimensions_to_sync = request.data.get('dimensions_to_sync', [])
             workspace = Workspace.objects.get(pk=kwargs['workspace_id'])
-
             sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace.id)
-            sync_dimensions(sage_intacct_credentials, workspace.id, dimensions_to_sync)
 
-            # Update destination_synced_at to current time only when full refresh happens
-            if not dimensions_to_sync:
-                workspace.destination_synced_at = datetime.now()
-                workspace.save(update_fields=['destination_synced_at'])
+            # If only specified dimensions are to be synced, sync them synchronously
+            if dimensions_to_sync:
+                sync_dimensions(sage_intacct_credentials, workspace.id, dimensions_to_sync)
+            else:
+                async_task(
+                    'apps.sage_intacct.helpers.sync_dimensions',
+                    sage_intacct_credentials,
+                    workspace.id
+                )
 
             return Response(
                 status=status.HTTP_200_OK

diff --git a/apps/workspaces/views.py b/apps/workspaces/views.py
@@ -90,9 +90,10 @@ def get(self, request):
         """
         user = User.objects.get(user_id=request.user)
         org_id = request.query_params.get('org_id')
+        is_polling = request.query_params.get('is_polling', False)
         workspaces = Workspace.objects.filter(user__in=[user], fyle_org_id=org_id).all()
 
-        if workspaces:
+        if workspaces and not is_polling:
             async_task(
                 'apps.workspaces.tasks.async_update_workspace_name',
                 workspaces[0],

diff --git a/tests/test_fyle/test_helpers.py b/tests/test_fyle/test_helpers.py
@@ -723,3 +723,28 @@ def test_bulk_update_expenses(db):
             settings.INTACCT_INTEGRATION_APP_URL
         )
         assert expense.accounting_export_summary['id'] == expense.expense_id
+
+def test_handle_refresh_dimensions(db, mocker):
+    mocker.patch(
+        'apps.fyle.helpers.sync_dimensions',
+        return_value=None
+    )
+
+    res = handle_refresh_dimensions(workspace_id=1)
+    assert res == None
+
+    workspace = Workspace.objects.get(id=1)
+    assert workspace.source_synced_at != None
+
+def test_check_interval_and_sync_dimension(db, mocker):
+    mocker.patch(
+        'apps.fyle.helpers.sync_dimensions',
+        return_value=None
+    )
+
+    res = check_interval_and_sync_dimension(workspace_id=1)
+    assert res == None
+
+    workspace = Workspace.objects.get(id=1)
+    assert workspace.source_synced_at != None
+
diff --git a/tests/test_sageintacct/test_helpers.py b/tests/test_sageintacct/test_helpers.py
@@ -13,15 +13,14 @@ def test_schedule_payment_sync(db):
 def test_check_interval_and_sync_dimension(db):
     workspace_id = 1
 
-    intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
     workspace = Workspace.objects.get(id=workspace_id)
 
-    check_interval_and_sync_dimension(workspace, intacct_credentials)
+    check_interval_and_sync_dimension(workspace_id)
 
     workspace.destination_synced_at = None
     workspace.save()
 
-    check_interval_and_sync_dimension(workspace, intacct_credentials)
+    check_interval_and_sync_dimension(workspace_id)
 
 
 def test_is_dependent_field_import_enabled(db, create_dependent_field_setting):