feat: Real-time exports, phase 2 - Intacct

apps/fyle/helpers.py

@@ -487,17 +487,21 @@ def get_fund_source(workspace_id: int) -> list[str]:
     return fund_source
 
 
-def handle_import_exception(task_log: TaskLog) -> None:
+def handle_import_exception(task_log: TaskLog | None) -> None:
     """
     Handle import exception
     :param task_log: task log
     :return: None
     """
     error = traceback.format_exc()
-    task_log.detail = {'error': error}
-    task_log.status = 'FATAL'
-    task_log.save()
-    logger.error('Something unexpected happened workspace_id: %s %s', task_log.workspace_id, task_log.detail)
+    if task_log:
+        task_log.detail = {'error': error}
+        task_log.status = 'FATAL'
+        task_log.updated_at = datetime.now()
+        task_log.save(update_fields=['detail', 'status', 'updated_at'])
+        logger.error('Something unexpected happened workspace_id: %s %s', task_log.workspace_id, task_log.detail)
+    else:
+        logger.error('Something unexpected happened %s', error)
 
 
 def assert_valid_request(workspace_id:int, fyle_org_id:str) -> None:
@@ -633,3 +637,12 @@ class Meta:
         model = Expense
         fields = ['org_id', 'is_skipped', 'updated_at__gte', 'updated_at__lte']
         or_fields = ['expense_number', 'employee_name', 'employee_email', 'claim_number']
+
+
+def update_task_log_post_import(task_log: TaskLog, status: str, message: str = None, error: str = None) -> None:
+    """Helper function to update task log status and details"""
+    if task_log:
+        task_log.status = status
+        task_log.detail = {"message": message} if message else {"error": error}
+        task_log.updated_at = datetime.now()
+        task_log.save(update_fields=['status', 'detail', 'updated_at'])

apps/fyle/signals.py

@@ -4,11 +4,16 @@
 from django.dispatch import receiver
 from django.core.exceptions import ValidationError
 
+from django_q.tasks import async_task
+
+from fyle_accounting_library.fyle_platform.enums import FundSourceEnum, ExpenseImportSourceEnum, ExpenseStateEnum
+
 from apps.fyle.tasks import re_run_skip_export_rule
 from apps.sage_intacct.dependent_fields import create_dependent_custom_field_in_fyle
 
 from apps.fyle.helpers import connect_to_platform
-from apps.fyle.models import DependentFieldSetting, ExpenseFilter
+from apps.fyle.models import DependentFieldSetting, ExpenseFilter, ExpenseGroupSettings
+from apps.workspaces.models import Configuration
 
 logger = logging.getLogger(__name__)
 logger.level = logging.INFO
@@ -70,3 +75,25 @@ def run_post_save_expense_filters(sender: type[ExpenseFilter], instance: Expense
         except Exception as e:
             logger.error(f'Error while processing expense filter for workspace: {instance.workspace.id} - {str(e)}')
             raise ValidationError('Failed to process expense filter')
+
+
+@receiver(pre_save, sender=ExpenseGroupSettings)
+def run_pre_save_expense_group_setting_triggers(sender: type[ExpenseGroupSettings], instance: ExpenseGroupSettings, **kwargs) -> None:
+    """
+    Run pre save expense group setting triggers
+    """
+    existing_expense_group_setting = ExpenseGroupSettings.objects.filter(
+        workspace_id=instance.workspace_id
+    ).first()
+
+    if existing_expense_group_setting:
+        configuration = Configuration.objects.filter(workspace_id=instance.workspace_id).first()
+        if configuration:
+            # TODO: move these async_tasks to maintenance worker later
+            if configuration.reimbursable_expenses_object and existing_expense_group_setting.expense_state != instance.expense_state and existing_expense_group_setting.expense_state == ExpenseStateEnum.PAID and instance.expense_state == ExpenseStateEnum.PAYMENT_PROCESSING:
+                logger.info(f'Reimbursable expense state changed from {existing_expense_group_setting.expense_state} to {instance.expense_state} for workspace {instance.workspace_id}, so pulling the data from Fyle')
+                async_task('apps.fyle.tasks.create_expense_groups', workspace_id=instance.workspace_id, fund_source=[FundSourceEnum.PERSONAL], task_log=None, imported_from=ExpenseImportSourceEnum.CONFIGURATION_UPDATE)
+
+            if configuration.corporate_credit_card_expenses_object and existing_expense_group_setting.ccc_expense_state != instance.ccc_expense_state and existing_expense_group_setting.ccc_expense_state == ExpenseStateEnum.PAID and instance.ccc_expense_state == ExpenseStateEnum.APPROVED:
+                logger.info(f'Corporate credit card expense state changed from {existing_expense_group_setting.ccc_expense_state} to {instance.ccc_expense_state} for workspace {instance.workspace_id}, so pulling the data from Fyle')
+                async_task('apps.fyle.tasks.create_expense_groups', workspace_id=instance.workspace_id, fund_source=[FundSourceEnum.CCC], task_log=None, imported_from=ExpenseImportSourceEnum.CONFIGURATION_UPDATE)

apps/fyle/tasks.py

@@ -13,6 +13,7 @@
     InternalServerError,
     InvalidTokenError
 )
+from fyle_accounting_library.fyle_platform.branding import feature_configuration
 from fyle_accounting_library.fyle_platform.helpers import get_expense_import_states, filter_expenses_based_on_state
 from fyle_accounting_library.fyle_platform.enums import ExpenseImportSourceEnum
 
@@ -22,7 +23,8 @@
     LastExportDetail,
     Workspace,
     FyleCredential,
-    Configuration
+    Configuration,
+    WorkspaceSchedule
 )
 from apps.fyle.models import (
     Expense,
@@ -34,7 +36,8 @@
     get_fund_source,
     get_source_account_type,
     handle_import_exception,
-    construct_expense_filter_query
+    construct_expense_filter_query,
+    update_task_log_post_import
 )
 from apps.fyle.actions import (
     mark_expenses_as_skipped,
@@ -86,7 +89,7 @@ def schedule_expense_group_creation(workspace_id: int) -> None:
     async_task('apps.fyle.tasks.create_expense_groups', workspace_id, fund_source, task_log)
 
 
-def create_expense_groups(workspace_id: int, fund_source: list[str], task_log: TaskLog,  imported_from: ExpenseImportSourceEnum) -> None:
+def create_expense_groups(workspace_id: int, fund_source: list[str], task_log: TaskLog | None, imported_from: ExpenseImportSourceEnum) -> None:
     """
     Create expense groups
     :param task_log: Task log object
@@ -97,8 +100,8 @@ def create_expense_groups(workspace_id: int, fund_source: list[str], task_log: T
         with transaction.atomic():
             workspace = Workspace.objects.get(pk=workspace_id)
 
-            last_synced_at = workspace.last_synced_at
-            ccc_last_synced_at = workspace.ccc_last_synced_at
+            last_synced_at = workspace.last_synced_at if imported_from != ExpenseImportSourceEnum.CONFIGURATION_UPDATE else None
+            ccc_last_synced_at = workspace.ccc_last_synced_at if imported_from != ExpenseImportSourceEnum.CONFIGURATION_UPDATE else None
             fyle_credentials = FyleCredential.objects.get(workspace_id=workspace_id)
 
             expense_group_settings = ExpenseGroupSettings.objects.get(workspace_id=workspace_id)
@@ -140,61 +143,38 @@ def create_expense_groups(workspace_id: int, fund_source: list[str], task_log: T
             if workspace.ccc_last_synced_at or len(expenses) != reimbursable_expense_count:
                 workspace.ccc_last_synced_at = datetime.now()
 
-            workspace.save()
+            if imported_from != ExpenseImportSourceEnum.CONFIGURATION_UPDATE:
+                workspace.save()
 
             group_expenses_and_save(expenses, task_log, workspace, imported_from=imported_from)
 
     except NoPrivilegeError:
         logger.info('Invalid Fyle Credentials / Admin is disabled')
-        task_log.detail = {
-            'message': 'Invalid Fyle Credentials / Admin is disabled'
-        }
-        task_log.status = 'FAILED'
-        task_log.save()
+        update_task_log_post_import(task_log, 'FAILED', message='Invalid Fyle Credentials / Admin is disabled')
 
     except FyleCredential.DoesNotExist:
         logger.info('Fyle credentials not found %s', workspace_id)
-        task_log.detail = {
-            'message': 'Fyle credentials do not exist in workspace'
-        }
-        task_log.status = 'FAILED'
-        task_log.save()
+        update_task_log_post_import(task_log, 'FAILED', message='Fyle credentials do not exist in workspace')
 
     except RetryException:
         logger.info('Fyle Retry Exception occured in workspace_id: %s', workspace_id)
-        task_log.detail = {
-            'message': 'Fyle Retry Exception occured'
-        }
-        task_log.status = 'FATAL'
-        task_log.save()
+        update_task_log_post_import(task_log, 'FATAL', message='Fyle Retry Exception occured')
 
     except InvalidTokenError:
         logger.info('Invalid Token for Fyle')
-        task_log.detail = {
-            'message': 'Invalid Token for Fyle'
-        }
-        task_log.status = 'FAILED'
-        task_log.save()
+        update_task_log_post_import(task_log, 'FAILED', message='Invalid Token for Fyle')
 
     except InternalServerError:
         logger.info('Fyle Internal Server Error occured in workspace_id: %s', workspace_id)
-        task_log.detail = {
-            'message': 'Fyle Internal Server Error occured'
-        }
-        task_log.status = 'FAILED'
-        task_log.save()
+        update_task_log_post_import(task_log, 'FAILED', message='Fyle Internal Server Error occured')
 
     except Exception:
         error = traceback.format_exc()
-        task_log.detail = {
-            'error': error
-        }
-        task_log.status = 'FATAL'
-        task_log.save()
+        update_task_log_post_import(task_log, 'FATAL', error=error)
         logger.exception('Something unexpected happened workspace_id: %s %s', task_log.workspace_id, task_log.detail)
 
 
-def group_expenses_and_save(expenses: list[dict], task_log: TaskLog, workspace: Workspace, imported_from: ExpenseImportSourceEnum = None) -> None:
+def group_expenses_and_save(expenses: list[dict], task_log: TaskLog | None, workspace: Workspace, imported_from: ExpenseImportSourceEnum = None) -> None:
     """
     Group expenses and save
     :param expenses: Expenses
@@ -237,8 +217,10 @@ def group_expenses_and_save(expenses: list[dict], task_log: TaskLog, workspace:
             except Exception:
                 logger.error('Error posting accounting export summary for workspace_id: %s', workspace.id)
 
-    task_log.status = 'COMPLETE'
-    task_log.save()
+    if task_log:
+        task_log.status = 'COMPLETE'
+        task_log.updated_at = datetime.now()
+        task_log.save(update_fields=['status', 'updated_at'])
 
 
 def import_and_export_expenses(report_id: str, org_id: str, is_state_change_event: bool, report_state: str = None, imported_from: ExpenseImportSourceEnum = None) -> None:
@@ -282,8 +264,17 @@ def import_and_export_expenses(report_id: str, org_id: str, is_state_change_even
         expense_groups = ExpenseGroup.objects.filter(expenses__id__in=[expense_ids], workspace_id=workspace.id, exported_at__isnull=True).distinct('id').values('id')
         expense_group_ids = [expense_group['id'] for expense_group in expense_groups]
 
-        if len(expense_group_ids) and not is_state_change_event:
-            export_to_intacct(workspace.id, None, expense_group_ids, triggered_by=imported_from)
+        if len(expense_group_ids):
+            if is_state_change_event:
+                # Trigger export immediately for customers who have enabled real time export
+                is_real_time_export_enabled = WorkspaceSchedule.objects.filter(workspace_id=workspace.id, is_real_time_export_enabled=True).exists()
+
+                # Don't allow real time export if it's not supported for the branded app / setting not enabled
+                if not is_real_time_export_enabled or not feature_configuration.feature.real_time_export_1hr_orgs:
+                    return
+
+            logger.info(f'Exporting expenses for workspace {workspace.id} with expense group ids {expense_group_ids}, triggered by {imported_from}')
+            export_to_intacct(workspace_id=workspace.id, expense_group_ids=expense_group_ids, triggered_by=imported_from)
 
     except Configuration.DoesNotExist:
         logger.info('Configuration does not exist for workspace_id: %s', workspace.id)

apps/fyle/views.py

@@ -73,7 +73,7 @@ def post(self, request: Request, *args, **kwargs) -> Response:
             fund_source.append('CCC')
 
         create_expense_groups(
-            kwargs['workspace_id'],
+            workspace_id=kwargs['workspace_id'],
             fund_source=fund_source,
             task_log=task_log,
             imported_from=ExpenseImportSourceEnum.DASHBOARD_SYNC
@@ -512,7 +512,7 @@ def post(self, request: Request, *args, **kwargs) -> Response:
         """
         task_log, fund_source = get_task_log_and_fund_source(kwargs['workspace_id'])
 
-        create_expense_groups(kwargs['workspace_id'], fund_source, task_log, imported_from=ExpenseImportSourceEnum.DASHBOARD_SYNC)
+        create_expense_groups(workspace_id=kwargs['workspace_id'], fund_source=fund_source, task_log=task_log, imported_from=ExpenseImportSourceEnum.DASHBOARD_SYNC)
 
         return Response(
             status=status.HTTP_200_OK

apps/internal/tasks.py

@@ -67,7 +67,7 @@ def re_export_stuck_exports() -> None:
                 export_expense_group_ids = list(expense_groups.filter(workspace_id=workspace_id).values_list('id', flat=True))
                 if export_expense_group_ids and len(export_expense_group_ids) < 200:
                     logger.info('Re-triggering export for expense group %s since no 1 hour schedule for workspace  %s', export_expense_group_ids, workspace_id)
-                    export_to_intacct(workspace_id, 'AUTO', export_expense_group_ids, triggered_by=ExpenseImportSourceEnum.INTERNAL)
+                    export_to_intacct(workspace_id=workspace_id, expense_group_ids=export_expense_group_ids, triggered_by=ExpenseImportSourceEnum.INTERNAL)
                 else:
                     logger.info('Skipping export for workspace %s since it has more than 200 expense groups', workspace_id)
 

apps/sage_intacct/queue.py

@@ -89,7 +89,7 @@ def schedule_journal_entries_creation(
             )
             if task_log.status not in ['IN_PROGRESS', 'ENQUEUED']:
                 task_log.status = 'ENQUEUED'
-                if task_log.triggered_by != triggered_by:
+                if triggered_by and task_log.triggered_by != triggered_by:
                     task_log.triggered_by = triggered_by
                 task_log.save()
 
@@ -190,7 +190,7 @@ def schedule_expense_reports_creation(workspace_id: int, expense_group_ids: list
             )
             if task_log.status not in ['IN_PROGRESS', 'ENQUEUED']:
                 task_log.status = 'ENQUEUED'
-                if task_log.triggered_by != triggered_by:
+                if triggered_by and task_log.triggered_by != triggered_by:
                     task_log.triggered_by = triggered_by
                 task_log.save()
 
@@ -254,7 +254,7 @@ def schedule_bills_creation(workspace_id: int, expense_group_ids: list[str], is_
             )
             if task_log.status not in ['IN_PROGRESS', 'ENQUEUED']:
                 task_log.status = 'ENQUEUED'
-                if task_log.triggered_by != triggered_by:
+                if triggered_by and task_log.triggered_by != triggered_by:
                     task_log.triggered_by = triggered_by
                 task_log.save()
 
@@ -318,7 +318,7 @@ def schedule_charge_card_transaction_creation(workspace_id: int, expense_group_i
             )
             if task_log.status not in ['IN_PROGRESS', 'ENQUEUED']:
                 task_log.status = 'ENQUEUED'
-                if task_log.triggered_by != triggered_by:
+                if triggered_by and task_log.triggered_by != triggered_by:
                     task_log.triggered_by = triggered_by
                 task_log.save()
 

apps/sage_intacct/views.py

@@ -13,6 +13,7 @@
 from fyle_accounting_mappings.serializers import DestinationAttributeSerializer
 from fyle_accounting_library.common_resources.models import DimensionDetail
 from fyle_accounting_library.common_resources.enums import DimensionDetailSourceTypeEnum
+from fyle_accounting_library.fyle_platform.enums import ExpenseImportSourceEnum
 
 from sageintacctsdk.exceptions import InvalidTokenError
 
@@ -124,7 +125,7 @@ def post(self, request: Request, *args, **kwargs) -> Response:
         """
         Trigger exports
         """
-        export_to_intacct(workspace_id=self.kwargs['workspace_id'])
+        export_to_intacct(workspace_id=self.kwargs['workspace_id'], triggered_by=ExpenseImportSourceEnum.DASHBOARD_SYNC)
 
         return Response(
             status=status.HTTP_200_OK

apps/workspaces/actions.py

@@ -19,7 +19,7 @@
 logger.level = logging.INFO
 
 
-def export_to_intacct(workspace_id: int, export_mode: bool = None, expense_group_ids: list = [], triggered_by: ExpenseImportSourceEnum = None) -> None:
+def export_to_intacct(workspace_id: int, expense_group_ids: list = [], triggered_by: ExpenseImportSourceEnum = None) -> None:
     """
     Export expenses to Intacct
     :param workspace_id: Workspace ID
@@ -33,7 +33,7 @@ def export_to_intacct(workspace_id: int, export_mode: bool = None, expense_group
 
     last_exported_at = datetime.now()
     is_expenses_exported = False
-    export_mode = export_mode or 'MANUAL'
+    export_mode = 'MANUAL' if triggered_by in (ExpenseImportSourceEnum.DASHBOARD_SYNC, ExpenseImportSourceEnum.DIRECT_EXPORT, ExpenseImportSourceEnum.CONFIGURATION_UPDATE) else 'AUTO'
     expense_group_filters = {
         'exported_at__isnull': True,
         'workspace_id': workspace_id

apps/workspaces/apis/advanced_settings/serializers.py

@@ -146,7 +146,8 @@ class Meta:
             'enabled',
             'interval_hours',
             'additional_email_options',
-            'emails_selected'
+            'emails_selected',
+            'is_real_time_export_enabled'
         ]
 
 

apps/workspaces/apis/advanced_settings/triggers.py

@@ -17,7 +17,8 @@ def run_post_configurations_triggers(workspace_id: int, workspace_schedule: Work
             schedule_enabled=workspace_schedule.get('enabled'),
             hours=workspace_schedule.get('interval_hours'),
             email_added=workspace_schedule.get('additional_email_options'),
-            emails_selected=workspace_schedule.get('emails_selected')
+            emails_selected=workspace_schedule.get('emails_selected'),
+            is_real_time_export_enabled=workspace_schedule.get('is_real_time_export_enabled')
         )
 
     @staticmethod

apps/workspaces/migrations/0045_workspaceschedule_is_real_time_export_enabled.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.21 on 2025-05-21 17:04
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('workspaces', '0044_configuration_je_single_credit_line'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='workspaceschedule',
+            name='is_real_time_export_enabled',
+            field=models.BooleanField(default=False),
+        ),
+    ]

apps/workspaces/models.py

@@ -207,6 +207,7 @@ class WorkspaceSchedule(models.Model):
     error_count = models.IntegerField(null=True, help_text='Number of errors in export')
     additional_email_options = JSONField(default=list, help_text='Email and Name of person to send email', null=True)
     emails_selected = ArrayField(base_field=models.CharField(max_length=255), null=True, help_text='Emails that has to be send mail')
+    is_real_time_export_enabled = models.BooleanField(default=False)
     schedule = models.OneToOneField(Schedule, on_delete=models.PROTECT, null=True)
     created_at = models.DateTimeField(auto_now_add=True, null=True, help_text='Created at datetime')
     updated_at = models.DateTimeField(auto_now=True, null=True, help_text='Updated at datetime')