Import Business Central Attributes API added (#27)

apps/business_central/helpers.py

@@ -0,0 +1,61 @@
+
+from datetime import datetime, timezone
+import logging
+
+from django.utils.module_loading import import_string
+
+from apps.workspaces.models import Workspace, BusinessCentralCredentials
+
+
+logger = logging.getLogger(__name__)
+logger.level = logging.INFO
+
+
+# Import your Workspace and BusinessCentralCredentials models here
+# Also, make sure you have 'logger' defined and imported from a logging module
+def check_interval_and_sync_dimension(workspace: Workspace, business_central_credential: BusinessCentralCredentials) -> bool:
+    """
+    Check the synchronization interval and trigger dimension synchronization if needed.
+
+    :param workspace: Workspace Instance
+    :param business_central_credential: BusinessCentralCredentials Instance
+
+    :return: True if synchronization is triggered, False if not
+    """
+
+    if workspace.destination_synced_at:
+        # Calculate the time interval since the last destination sync
+        time_interval = datetime.now(timezone.utc) - workspace.destination_synced_at
+
+    if workspace.destination_synced_at is None or time_interval.days > 0:
+        # If destination_synced_at is None or the time interval is greater than 0 days, trigger synchronization
+        sync_dimensions(business_central_credential, workspace.id)
+        return True
+
+    return False
+
+
+def sync_dimensions(business_central_credential: BusinessCentralCredentials, workspace_id: int) -> None:
+    """
+    Synchronize various dimensions with Business Central using the provided credentials.
+
+    :param business_central_credential: BusinessCentralCredentials Instance
+    :param workspace_id: ID of the workspace
+
+    This function syncs dimensions like accounts, vendors, commitments, jobs, categories, and cost codes.
+    """
+
+    # Initialize the Business Central connection using the provided credentials and workspace ID
+    business_central_connection = import_string('apps.business_central.utils.BusinessCentralConnector')(business_central_credential, workspace_id)
+
+    # List of dimensions to sync
+    dimensions = ['accounts', 'vendors', 'employees', 'locations']
+
+    for dimension in dimensions:
+        try:
+            # Dynamically call the sync method based on the dimension
+            sync = getattr(business_central_connection, 'sync_{}'.format(dimension))
+            sync()
+        except Exception as exception:
+            # Log any exceptions that occur during synchronization
+            logger.info(exception)

apps/business_central/serializers.py

@@ -0,0 +1,62 @@
+import logging
+from datetime import datetime
+from rest_framework import serializers
+from rest_framework.response import Response
+from rest_framework.views import status
+
+from apps.workspaces.models import Workspace, BusinessCentralCredentials
+from apps.business_central.helpers import sync_dimensions, check_interval_and_sync_dimension
+
+
+logger = logging.getLogger(__name__)
+logger.level = logging.INFO
+
+
+class ImportBusinessCentralAttributesSerializer(serializers.Serializer):
+    """
+    Import Business Central Attributes serializer
+    """
+
+    def create(self, validated_data):
+        try:
+            # Get the workspace ID from the URL kwargs
+            workspace_id = self.context['request'].parser_context['kwargs']['workspace_id']
+
+            # Check if the 'refresh' field is provided in the request data
+            refresh_dimension = self.context['request'].data.get('refresh', False)
+
+            # Retrieve the workspace and Business Central credentials
+            workspace = Workspace.objects.get(pk=workspace_id)
+            business_central_credentials = BusinessCentralCredentials.objects.get(
+                workspace_id=workspace.id
+            )
+
+            if refresh_dimension:
+                # If 'refresh' is true, perform a full sync of dimensions
+                sync_dimensions(business_central_credentials, workspace.id)
+            else:
+                # If 'refresh' is false, check the interval and sync dimension accordingly
+                check_interval_and_sync_dimension(workspace, business_central_credentials)
+
+            # Update the destination_synced_at field and save the workspace
+            workspace.destination_synced_at = datetime.now()
+            workspace.save(update_fields=['destination_synced_at'])
+
+            # Return a success response
+            return Response(status=status.HTTP_200_OK)
+
+        except BusinessCentralCredentials.DoesNotExist:
+            # Handle the case when business central credentials are not found or invalid
+            raise serializers.ValidationError(
+                {'message': 'Business Central credentials not found / invalid in workspace'}
+            )
+
+        except Exception as exception:
+            # Handle unexpected exceptions and log the error
+            logger.error(
+                'Something unexpected happened workspace_id: %s %s',
+                workspace_id,
+                exception,
+            )
+            # Raise a custom exception or re-raise the original exception
+            raise

apps/business_central/urls.py

@@ -0,0 +1,12 @@
+from django.urls import path
+
+from apps.business_central.views import ImportBusinessCentralAttributesView
+
+
+urlpatterns = [
+    path(
+        "import_attributes/",
+        ImportBusinessCentralAttributesView.as_view(),
+        name="import-business-central-attributes",
+    )
+]

apps/business_central/views.py

@@ -0,0 +1,16 @@
+import logging
+
+from rest_framework import generics
+
+from apps.business_central.serializers import ImportBusinessCentralAttributesSerializer
+
+
+logger = logging.getLogger(__name__)
+logger.level = logging.INFO
+
+
+class ImportBusinessCentralAttributesView(generics.CreateAPIView):
+    """
+    Import Business Central Attributes View
+    """
+    serializer_class = ImportBusinessCentralAttributesSerializer

apps/workspaces/urls.py

@@ -25,6 +25,7 @@
 other_app_paths = [
     path('<int:workspace_id>/accounting_exports/', include('apps.accounting_exports.urls')),
     path('<int:workspace_id>/fyle/', include('apps.fyle.urls')),
+    path('<int:workspace_id>/business_central/', include('apps.business_central.urls')),
 ]
 
 urlpatterns = []

docker-compose-pipeline.yml

@@ -24,7 +24,7 @@ services:
       FYLE_CLIENT_SECRET: 'sample'
       FYLE_REFRESH_TOKEN: 'sample.sample.sample'
       BUSINESS_CENTRAL_REDIRECT_URI: ${BUSINESS_CENTRAL_REDIRECT_URI}
-      BUSINESS_CENTRALTOKEN_URI: ${BUSINESS_CENTRALTOKEN_URI}
+      BUSINESS_CENTRAL_TOKEN_URI: ${BUSINESS_CENTRAL_TOKEN_URI}
       ENCRYPTION_KEY: ${ENCRYPTION_KEY}
       FYLE_TOKEN_URI: 'https://sample.fyle.tech'
       FYLE_SERVER_URL: 'https://sample.fyle.tech'

tests/conftest.py

@@ -14,6 +14,7 @@
 from apps.workspaces.models import (
     Workspace,
     FyleCredential,
+    BusinessCentralCredentials
 )
 from apps.accounting_exports.models import AccountingExport, AccountingExportSummary, Error
 from apps.fyle.models import ExpenseFilter
@@ -254,3 +255,20 @@ def add_expense_filters():
             custom_field_type='SELECT',
             workspace_id=workspace_id
         )
+
+
+@pytest.fixture()
+@pytest.mark.django_db(databases=['default'])
+def add_business_central_creds():
+    """
+    Pytest fixture to add business central credentials to a workspace
+    """
+    workspace_ids = [
+        1, 2, 3
+    ]
+    for workspace_id in workspace_ids:
+        BusinessCentralCredentials.objects.create(
+            refresh_token = 'dummy_refresh_token',
+            is_expired = False,
+            workspace_id = workspace_id
+        )

tests/test_business_central/test_views.py

@@ -0,0 +1,27 @@
+import json
+from django.urls import reverse
+
+from apps.workspaces.models import BusinessCentralCredentials
+
+
+def test_sync_dimensions(api_client, test_connection, mocker, create_temp_workspace, add_business_central_creds):
+    workspace_id = 1
+
+    access_token = test_connection.access_token
+    url = reverse('import-business-central-attributes', kwargs={'workspace_id': workspace_id})
+
+    api_client.credentials(HTTP_AUTHORIZATION='Bearer {}'.format(access_token))
+
+    mocker.patch('apps.business_central.helpers.sync_dimensions', return_value=None)
+
+    response = api_client.post(url)
+    assert response.status_code == 201
+
+    business_central_credentials = BusinessCentralCredentials.objects.get(workspace_id=workspace_id)
+    business_central_credentials.delete()
+
+    response = api_client.post(url)
+    assert response.status_code == 400
+
+    response = json.loads(response.content)
+    assert response['message'] == 'Business Central credentials not found / invalid in workspace'