Business Central creds apis

apps/workspaces/helpers.py

@@ -0,0 +1,97 @@
+import logging
+import base64
+import requests
+import json
+
+from django.conf import settings
+from future.moves.urllib.parse import urlencode
+from ms_dynamics_business_central_sdk import InternalServerError, InvalidTokenError
+
+from apps.workspaces.models import BusinessCentralCredentials, Workspace
+from apps.business_central.utils import BusinessCentralConnector
+
+
+logger = logging.getLogger(__name__)
+
+
+def generate_token(authorization_code: str, redirect_uri: str = None) -> str:
+    api_data = {
+        "grant_type": "authorization_code",
+        "code": authorization_code,
+        "redirect_uri": settings.BUSINESS_CENTRAL_REDIRECT_URI
+        if not redirect_uri
+        else redirect_uri,
+    }
+
+    auth = "{0}:{1}".format(settings.BUSINESS_CENTRAL_ID, settings.BUSINESS_CENTRAL_SECRET)
+    auth = base64.b64encode(auth.encode("utf-8"))
+
+    request_header = {
+        "Accept": "application/json",
+        "Content-type": "application/x-www-form-urlencoded",
+        "Authorization": "Basic {0}".format(str(auth.decode())),
+    }
+
+    token_url = settings.BUSINESS_CENTRAL_TOKEN_URI
+    response = requests.post(
+        url=token_url, data=urlencode(api_data), headers=request_header
+    )
+    return response
+
+
+def generate_business_central_refresh_token(authorization_code: str, redirect_uri: str = None) -> str:
+    """
+    Generate Business Central refresh token from authorization code
+    """
+    response = generate_token(authorization_code, redirect_uri)
+
+    if response.status_code == 200:
+        successful_response = json.loads(response.text)
+        return successful_response["refresh_token"]
+
+    elif response.status_code == 401:
+        raise InvalidTokenError(
+            "Wrong client secret or/and refresh token", response.text
+        )
+
+    elif response.status_code == 500:
+        raise InternalServerError("Internal server error", response.text)
+
+
+def connect_business_central(authorization_code, redirect_uri, workspace_id):
+    if redirect_uri:
+        refresh_token = generate_business_central_refresh_token(authorization_code, redirect_uri)
+    else:
+        refresh_token = generate_business_central_refresh_token(authorization_code)
+    business_central_credentials = BusinessCentralCredentials.objects.filter(workspace_id=workspace_id).first()
+
+    workspace = Workspace.objects.get(pk=workspace_id)
+
+    if not business_central_credentials:
+        business_central_credentials = BusinessCentralCredentials.objects.create(
+            refresh_token=refresh_token, workspace_id=workspace_id
+        )
+    else:
+        business_central_credentials.refresh_token = refresh_token
+        business_central_credentials.is_expired = False
+        business_central_credentials.save()
+
+    if workspace and not workspace.business_central_company_id:
+        business_central_connector = BusinessCentralConnector(business_central_credentials, workspace_id=workspace_id)
+        connections = business_central_connector.connection.connections.get_all()
+        connection = list(
+            filter(
+                lambda connection: connection["id"] == workspace.business_central_company_id,
+                connections,
+            )
+        )
+
+        if connection:
+            workspace.business_central_company_id = connection[0]["id"]
+            workspace.save()
+
+    if workspace.onboarding_state == "CONNECTION":
+        workspace.onboarding_state = "EXPORT_SETTINGS"
+        workspace.save()
+
+    return business_central_credentials

apps/workspaces/serializers.py

@@ -11,6 +11,7 @@
 from apps.workspaces.models import (
     Workspace,
     FyleCredential,
+    BusinessCentralCredentials,
     ExportSetting,
     ImportSetting,
     AdvancedSetting
@@ -70,6 +71,16 @@ def create(self, validated_data):
         return workspace
 
 
+class BusinessCentralCredentialSerializer(serializers.ModelSerializer):
+    """
+    Business Central credential serializer
+    """
+
+    class Meta:
+        model = BusinessCentralCredentials
+        fields = "__all__"
+
+
 class ExportSettingsSerializer(serializers.ModelSerializer):
     """
     Export Settings serializer

apps/workspaces/urls.py

@@ -6,13 +6,16 @@
     ExportSettingView,
     ImportSettingView,
     AdvancedSettingView,
-    WorkspaceAdminsView
+    WorkspaceAdminsView,
+    ConnectBusinessCentralView
 )
 
 
 workspace_app_paths = [
     path('', WorkspaceView.as_view(), name='workspaces'),
     path('ready/', ReadyView.as_view(), name='ready'),
+    path("<int:workspace_id>/connect_business_central/authorization_code/", ConnectBusinessCentralView.as_view()),
+    path("<int:workspace_id>/credentials/business_central/", ConnectBusinessCentralView.as_view()),
     path('<int:workspace_id>/export_settings/', ExportSettingView.as_view(), name='export-settings'),
     path('<int:workspace_id>/import_settings/', ImportSettingView.as_view(), name='import-settings'),
     path('<int:workspace_id>/advanced_settings/', AdvancedSettingView.as_view(), name='advanced-settings'),

apps/workspaces/views.py

@@ -8,14 +8,17 @@
 
 
 from ms_business_central_api.utils import assert_valid
+from apps.workspaces.helpers import connect_business_central
 from apps.workspaces.models import (
     Workspace,
     ExportSetting,
     ImportSetting,
-    AdvancedSetting
+    AdvancedSetting,
+    BusinessCentralCredentials
 )
 from apps.workspaces.serializers import (
     WorkspaceSerializer,
+    BusinessCentralCredentialSerializer,
     ExportSettingsSerializer,
     ImportSettingsSerializer,
     AdvancedSettingSerializer,
@@ -77,6 +80,47 @@ def get(self, request, *args, **kwargs):
         )
 
 
+class ConnectBusinessCentralView(generics.CreateAPIView, generics.RetrieveAPIView):
+    """
+    Business Central Connect Oauth View
+    """
+
+    def post(self, request, **kwargs):
+        """
+        Post of Business Central Credentials
+        """
+
+        authorization_code = request.data.get("code")
+        redirect_uri = request.data.get("redirect_uri")
+
+        business_central_credentials = connect_business_central(
+            authorization_code=authorization_code,
+            redirect_uri=redirect_uri,
+            workspace_id=kwargs["workspace_id"],
+        )
+
+        return Response(
+            data=BusinessCentralCredentialSerializer(business_central_credentials).data,
+            status=status.HTTP_200_OK,
+        )
+
+    def get(self, request, **kwargs):
+        """
+        Get Business Central Credentials in Workspace
+        """
+
+        business_central_credentials = BusinessCentralCredentials.objects.get(
+            workspace_id=kwargs["workspace_id"],
+            is_expired=False,
+            refresh_token__isnull=False,
+        )
+
+        return Response(
+            data=BusinessCentralCredentialSerializer(business_central_credentials).data,
+            status=status.HTTP_200_OK,
+        )
+
+
 class ExportSettingView(generics.CreateAPIView, generics.RetrieveAPIView):
     """
     Retrieve or Create Export Settings