Refactor imports supplier (#318)

* Added integrations_imports submodule, made changes in settings

* Resource Project: Added new flow for the Project Import

* Typo Fix

* Resource Project: removed old import logic for Project

* Resource Cost_Center: refactored imports

* Resource Custom Expense Field: refactored imports

* Resource Tax Group: refactored imports

* Resource Category: refactored imports

* Added RetryException handler and bumped the sdk versions (#306)

* Set tasks limit to 1L (#317)

* Set tasks limit to 1L

* set timeout

* Modified the import schedule creation condition

---------

Co-authored-by: ruuushhh <[email protected]>

apps/fyle/tasks.py

@@ -4,7 +4,10 @@
 from typing import Dict, List
 
 from django.db import transaction
-from fyle.platform.exceptions import InvalidTokenError as FyleInvalidTokenError
+from fyle.platform.exceptions import (
+    InvalidTokenError as FyleInvalidTokenError,
+    RetryException,
+)
 from fyle_integrations_platform_connector import PlatformConnector
 
 from apps.fyle.actions import create_generator_and_post_in_batches
@@ -146,6 +149,13 @@ def async_create_expense_groups(
 
     except FyleInvalidTokenError:
         logger.info("Invalid Token for Fyle")
+
+    except RetryException:
+        logger.info("Fyle Retry Exception occured in workspace_id: %s", workspace_id)
+        task_log.detail = {"message": "Fyle Retry Exception"}
+        task_log.status = TaskLogStatusEnum.FATAL
+        task_log.save()
+
     except Exception:
         error = traceback.format_exc()
         task_log.detail = {"error": error}

apps/mappings/exceptions.py

@@ -1,7 +1,13 @@
 import logging
 import traceback
 
-from fyle.platform.exceptions import InternalServerError, InvalidTokenError, PlatformError, WrongParamsError
+from fyle.platform.exceptions import (
+    InternalServerError,
+    InvalidTokenError,
+    PlatformError,
+    WrongParamsError,
+    RetryException
+)
 from xerosdk.exceptions import InvalidGrant
 from xerosdk.exceptions import InvalidTokenError as XeroInvalidTokenError
 from xerosdk.exceptions import UnsuccessfulAuthentication
@@ -38,6 +44,10 @@ def new_fn(workspace_id, *args):
                 error["response"] = exception.response
                 error["alert"] = True
 
+            except RetryException as exception:
+                error["message"] = "Retry exception"
+                error["response"] = exception.__dict__
+
             except InternalServerError as exception:
                 error["message"] = "Internal server error while importing to Fyle"
                 error["response"] = exception.__dict__

apps/mappings/queue.py

@@ -128,6 +128,7 @@ def construct_tasks_and_chain_import_fields_to_fyle(workspace_id: int):
         'mapping_settings': [],
         'credentials': credentials,
         'sdk_connection_string': 'apps.xero.utils.XeroConnector',
+        'custom_properties': None
     }
 
     # For now adding only for PROJECT
@@ -153,6 +154,14 @@ def construct_tasks_and_chain_import_fields_to_fyle(workspace_id: int):
             'charts_of_accounts': workspace_general_settings.charts_of_accounts
         }
 
+    if workspace_general_settings.import_suppliers_as_merchants:
+        task_settings['custom_properties'] = {
+            'func': 'apps.mappings.tasks.auto_create_suppliers_as_merchants',
+            'args': {
+                'workspace_id': workspace_id
+            }
+        }
+
     if mapping_settings:
         for mapping_setting in mapping_settings:
             if mapping_setting.source_field in ALLOWED_SOURCE_FIELDS or mapping_setting.is_custom:

apps/mappings/schedules.py

@@ -24,6 +24,7 @@ def new_schedule_or_delete_fyle_import_tasks(
         or workspace_general_settings_instance.import_customers
         or workspace_general_settings_instance.import_tax_codes
         or workspace_general_settings_instance.import_categories
+        or workspace_general_settings_instance.import_suppliers_as_merchants
     ):
         Schedule.objects.update_or_create(
             func='apps.mappings.queue.construct_tasks_and_chain_import_fields_to_fyle',

apps/mappings/tasks.py

@@ -1,7 +1,7 @@
 import logging
 from typing import List
+from datetime import datetime, timedelta, timezone
 
-from django_q.tasks import Chain
 from fyle_accounting_mappings.models import Mapping
 from fyle_integrations_platform_connector import PlatformConnector
 
@@ -10,39 +10,11 @@
 from apps.tasks.models import Error
 from apps.workspaces.models import FyleCredential, WorkspaceGeneralSettings, XeroCredentials
 from apps.xero.utils import XeroConnector
+from fyle_integrations_imports.models import ImportLog
 
 logger = logging.getLogger(__name__)
 logger.level = logging.INFO
 
-DEFAULT_FYLE_CATEGORIES = [
-    "Activity",
-    "Train",
-    "Fuel",
-    "Snacks",
-    "Office Supplies",
-    "Utility",
-    "Entertainment",
-    "Others",
-    "Mileage",
-    "Food",
-    "Per Diem",
-    "Bus",
-    "Internet",
-    "Taxi",
-    "Courier",
-    "Hotel",
-    "Professional Services",
-    "Phone",
-    "Office Party",
-    "Flight",
-    "Software",
-    "Parking",
-    "Toll Charge",
-    "Tax",
-    "Training",
-    "Unspecified",
-]
-
 
 def resolve_expense_attribute_errors(
     source_attribute_type: str,
@@ -93,6 +65,34 @@ def async_auto_map_employees(workspace_id: int):
 
 
 def auto_create_suppliers_as_merchants(workspace_id):
+    logger.info("| Importing Suppliers as Merchant to Fyle | Content: {{WORKSPACE_ID: {}}}".format(workspace_id))
+
+    import_log, is_created = ImportLog.objects.get_or_create(
+        workspace_id=workspace_id,
+        attribute_type="MERCHANT",
+        defaults={
+            'status': 'IN_PROGRESS'
+        }
+    )
+
+    sync_after = None
+
+    if not is_created:
+        sync_after = import_log.last_successful_run_at if import_log.last_successful_run_at else None
+
+    time_difference = datetime.now() - timedelta(minutes=30)
+    offset_aware_time_difference = time_difference.replace(tzinfo=timezone.utc)
+
+    if (import_log.status == 'IN_PROGRESS' and not is_created) \
+        or (sync_after and (sync_after > offset_aware_time_difference)):
+        return
+
+    else:
+        import_log.status = 'IN_PROGRESS'
+        import_log.processed_batches_count = 0
+        import_log.total_batches_count = 0
+        import_log.save()
+
     fyle_credentials: FyleCredential = FyleCredential.objects.get(
         workspace_id=workspace_id
     )
@@ -106,24 +106,9 @@ def auto_create_suppliers_as_merchants(workspace_id):
     if merchant_names:
         fyle_connection.merchants.post(merchant_names, skip_existing_merchants=True)
 
-
-def auto_import_and_map_fyle_fields(workspace_id):
-    """
-    Auto import and map fyle fields
-    """
-    workspace_general_settings: WorkspaceGeneralSettings = (
-        WorkspaceGeneralSettings.objects.get(workspace_id=workspace_id)
-    )
-
-    chain = Chain()
-
-    if workspace_general_settings.import_suppliers_as_merchants:
-        chain.append(
-            "apps.mappings.tasks.auto_create_suppliers_as_merchants", workspace_id,
-            q_options={
-                'cluster': 'import'
-            }
-        )
-
-    if chain.length() > 0:
-        chain.run()
+    import_log.status = 'COMPLETE'
+    import_log.last_successful_run_at = datetime.now()
+    import_log.error_log = []
+    import_log.total_batches_count = 0
+    import_log.processed_batches_count = 0
+    import_log.save()

apps/workspaces/apis/import_settings/triggers.py

@@ -3,7 +3,6 @@
 from django.db.models import Q
 from fyle_accounting_mappings.models import MappingSetting
 
-from apps.mappings.helpers import schedule_or_delete_fyle_import_tasks
 from apps.workspaces.models import WorkspaceGeneralSettings
 from apps.mappings.schedules import new_schedule_or_delete_fyle_import_tasks
 
@@ -36,7 +35,6 @@ def post_save_workspace_general_settings(
                 destination_field="CUSTOMER",
             ).delete()
 
-        schedule_or_delete_fyle_import_tasks(workspace_general_settings_instance)
         new_schedule_or_delete_fyle_import_tasks(
             workspace_general_settings_instance=workspace_general_settings_instance,
             mapping_settings=self.__mapping_settings,

apps/xero/actions.py

@@ -42,9 +42,6 @@ def refersh_xero_dimension(workspace_id):
     xero_credentials = XeroCredentials.get_active_xero_credentials(
         workspace_id=workspace_id
     )
-    xero_credentials = XeroCredentials.get_active_xero_credentials(
-        workspace_id=workspace_id
-    )
     xero_connector = get_xero_connector(workspace_id=workspace_id)
 
     mapping_settings = MappingSetting.objects.filter(

fyle_xero_api/settings.py

@@ -115,9 +115,11 @@
 
 Q_CLUSTER = {
     "name": "fyle_xero_api",
-    "save_limit": 0,
+    # The number of tasks will be stored in django q tasks
+    "save_limit": 100000,
     "retry": 14400,
-    "timeout": 3600,
+    # 15 mins
+    'timeout': 900,
     "catch_up": False,
     "workers": 4,
     # How many tasks are kept in memory by a single cluster.
@@ -138,7 +140,8 @@
     'ALT_CLUSTERS': {
         'import': {
             'retry': 14400,
-            'timeout': 3600
+            # 15 mins
+            'timeout': 900,
         },
     }
 }

requirements.txt

@@ -17,9 +17,9 @@ djangorestframework==3.11.2
 django-sendgrid-v5==1.2.0
 enum34==1.1.10
 future==0.18.2
-fyle==0.35.0
+fyle==0.36.1
 fyle-accounting-mappings==1.26.2
-fyle-integrations-platform-connector==1.36.1
+fyle-integrations-platform-connector==1.36.3
 fyle-rest-auth==1.7.0
 gevent==23.9.1
 gunicorn==20.1.0