feat: NetSuite Internal APIs

apps/internal/actions.py

@@ -0,0 +1,30 @@
+from typing import Dict
+
+from apps.netsuite.connector import NetSuiteConnector
+from apps.workspaces.models import Workspace, NetSuiteCredentials
+
+
+def get_netsuite_connection(query_params: Dict):
+    org_id = query_params.get('org_id')
+
+    workspace = Workspace.objects.get(fyle_org_id=org_id)
+    workspace_id = workspace.id
+    ns_credentials = NetSuiteCredentials.objects.get(workspace_id=workspace.id)
+
+    return NetSuiteConnector(netsuite_credentials=ns_credentials, workspace_id=workspace_id)
+
+
+def get_accounting_fields(query_params: Dict):
+    ns_connection = get_netsuite_connection(query_params)
+    resource_type = query_params.get('resource_type')
+    internal_id = query_params.get('internal_id')
+
+    return ns_connection.get_accounting_fields(resource_type, internal_id)
+
+
+def get_exported_entry(query_params: Dict):
+    ns_connection = get_netsuite_connection(query_params)
+    resource_type = query_params.get('resource_type')
+    internal_id = query_params.get('internal_id')
+
+    return ns_connection.get_exported_entry(resource_type, internal_id)

apps/internal/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class InternalConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'apps.internal'

apps/internal/urls.py

@@ -0,0 +1,10 @@
+import itertools
+
+from django.urls import path
+
+from .views import AccountingFieldsView, ExportedEntryView
+
+urlpatterns = [
+    path('accounting_fields/', AccountingFieldsView.as_view(), name='accounting-fields'),
+    path('exported_entry/', ExportedEntryView.as_view(), name='exported-entry'),
+]

apps/internal/views.py

@@ -0,0 +1,68 @@
+import logging
+import traceback
+from rest_framework import generics
+from rest_framework.response import Response
+from rest_framework import status
+
+from apps.workspaces.permissions import IsAuthenticatedForInternalAPI
+
+from fyle_netsuite_api.utils import assert_valid
+
+from .actions import get_accounting_fields, get_exported_entry
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+
+class AccountingFieldsView(generics.GenericAPIView):
+    authentication_classes = []
+    permission_classes = [IsAuthenticatedForInternalAPI]
+
+    def get(self, request, *args, **kwargs):
+        try:
+            params = request.query_params
+
+            assert_valid(params.get('org_id') is not None, 'Org ID is required')
+            assert_valid(params.get('resource_type') is not None, 'Resource Type is required')
+
+            if params.get('resource_type') in ('custom_segments', 'custom_lists', 'custom_record_types'):
+                assert_valid(params.get('internal_id') is not None, 'Internal ID is required')
+
+            response = get_accounting_fields(request.query_params)
+            return Response(
+                data={'data': response},
+                status=status.HTTP_200_OK
+            )
+
+        except Exception:
+            logger.info(f"Error in AccountingFieldsView: {traceback.format_exc()}")
+            return Response(
+                data={'error': traceback.format_exc()},
+                status=status.HTTP_400_BAD_REQUEST
+            )
+
+
+class ExportedEntryView(generics.GenericAPIView):
+    authentication_classes = []
+    permission_classes = [IsAuthenticatedForInternalAPI]
+
+    def get(self, request, *args, **kwargs):
+        try:
+            params = request.query_params
+            assert_valid(params.get('org_id') is not None, 'Org ID is required')
+            assert_valid(params.get('resource_type') is not None, 'Resource Type is required')
+            assert_valid(params.get('internal_id') is not None, 'Internal ID is required')
+
+            response = get_exported_entry(request.query_params)
+
+            return Response(
+                data={'data': response},
+                status=status.HTTP_200_OK
+            )
+
+        except Exception:
+            logger.info(f"Error in AccountingFieldsView: {traceback.format_exc()}")
+            return Response(
+                data={'error': traceback.format_exc()},
+                status=status.HTTP_400_BAD_REQUEST
+            )

apps/netsuite/connector.py

@@ -1162,7 +1162,42 @@ def sync_customers(self):
                 attributes, 'PROJECT', self.workspace_id, True)
 
         return []
-
+
+    def get_accounting_fields(self, resource_type: str, internal_id: str):
+        """
+        Retrieve accounting fields for a specific resource type and internal ID.
+
+        Args:
+            resource_type (str): The type of resource to fetch.
+            internal_id (str): The internal ID of the resource.
+
+        Returns:
+            list or dict: Parsed JSON representation of the resource data.
+        """
+        module = getattr(self.connection, resource_type)
+        method_map = {
+            'currencies': 'get_all',
+            'custom_segments': 'get',
+            'custom_lists': 'get',
+            'custom_record_types': 'get_all_by_id',
+        }
+        method = method_map.get(resource_type, 'get_all_generator')
+
+        if method in ('get', 'get_all_by_id'):
+            response = getattr(module, method)(internal_id)
+        else:
+            response = getattr(module, method)()
+
+        if method == 'get_all_generator':
+            response = [row for responses in response for row in responses]
+
+        return json.loads(json.dumps(response, default=str))
+
+    def get_exported_entry(self, resource_type: str, export_id: str):
+        module = getattr(self.connection, resource_type)
+        response = getattr(module, 'get')(export_id)
+        return json.loads(json.dumps(response, default=str))
+
     def handle_taxed_line_items(self, base_line, line, workspace_id, export_module, general_mapping: GeneralMapping):
         """
         Handle line items where tax is applied or modified by the user.

apps/workspaces/permissions.py

@@ -34,15 +34,15 @@ def has_permission(self, request, view):
             workspace_users = Workspace.objects.filter(pk=workspace_id).values_list('user', flat=True)
             return self.validate_and_cache(workspace_users, user, workspace_id, True)
 
-class IsAuthenticatedForTest(permissions.BasePermission):
+class IsAuthenticatedForInternalAPI(permissions.BasePermission):
     """
-    Custom auth for preparing a workspace for e2e tests
+    Custom auth for internal APIs
     """
     def has_permission(self, request, view):
         # Client sends a token in the header, which we decrypt and compare with the Client Secret
         cipher_suite = Fernet(settings.ENCRYPTION_KEY)
         try:
-            decrypted_password = cipher_suite.decrypt(request.headers['X-E2E-Tests-Client-ID'].encode('utf-8')).decode('utf-8')
+            decrypted_password = cipher_suite.decrypt(request.headers['X-Internal-API-Client-ID'].encode('utf-8')).decode('utf-8')
             if decrypted_password == settings.E2E_TESTS_CLIENT_SECRET:
                 return True
         except:

apps/workspaces/views.py

@@ -33,7 +33,7 @@
 from apps.workspaces.actions import export_to_netsuite
 from .serializers import LastExportDetailSerializer, WorkspaceSerializer, FyleCredentialSerializer, NetSuiteCredentialSerializer, \
     ConfigurationSerializer, WorkspaceScheduleSerializer
-from .permissions import IsAuthenticatedForTest
+from .permissions import IsAuthenticatedForInternalAPI
 
 
 logger = logging.getLogger(__name__)
@@ -484,7 +484,7 @@ class SetupE2ETestView(viewsets.ViewSet):
     NetSuite Workspace
     """
     authentication_classes = []
-    permission_classes = [IsAuthenticatedForTest]
+    permission_classes = [IsAuthenticatedForInternalAPI]
 
     def post(self, request, **kwargs):
         """

fyle_netsuite_api/settings.py

@@ -62,6 +62,7 @@
     'apps.netsuite',
     'django_q',
     'django_filters',
+    'apps.internal'
 ]
 
 MIDDLEWARE = [

fyle_netsuite_api/urls.py

@@ -22,4 +22,5 @@
     path('api/workspaces/', include('apps.workspaces.urls')),
     path('api/user/', include('apps.users.urls')),
     path('api/v2/workspaces/', include('apps.workspaces.apis.urls')),
+    path('api/internal/', include('apps.internal.urls')),
 ]

tests/test_internal/test_actions.py

@@ -0,0 +1,42 @@
+from apps.internal.actions import get_accounting_fields, get_exported_entry
+from tests.test_netsuite.fixtures import data
+
+def test_get_accounting_fields(db, mocker):
+    query_params = {
+        'org_id': 'or79Cob97KSh',
+        'resource_type': 'employees',
+    }
+    mocker.patch(
+        'netsuitesdk.api.employees.Employees.get_all_generator',
+        return_value=data['get_all_employees']    
+    )
+
+    mocker.patch('netsuitesdk.api.currencies.Currencies.get_all')
+
+    mocker.patch(
+        'netsuitesdk.api.custom_lists.CustomLists.get',
+        return_value=data['get_custom_list']
+    )
+
+    fields = get_accounting_fields(query_params)
+    assert fields is not None
+
+    query_params['resource_type'] = 'custom_lists'
+    query_params['internal_id'] = '1'
+    fields = get_accounting_fields(query_params)
+    assert fields is not None
+
+
+def test_get_exported_entry(db, mocker):
+    query_params = {
+        'org_id': 'or79Cob97KSh',
+        'resource_type': 'vendor_bills',
+        'internal_id': '1'
+    }
+    mocker.patch(
+        'netsuitesdk.api.vendor_bills.VendorBills.get',
+        return_value={'summa': 'hehe'}
+    )
+
+    entry = get_exported_entry(query_params)
+    assert entry is not None

tests/test_internal/test_views.py

@@ -0,0 +1,61 @@
+import pytest
+from unittest.mock import patch
+from django.urls import reverse
+
+from apps.workspaces.permissions import IsAuthenticatedForInternalAPI
+
+from tests.test_netsuite.fixtures import data
+
+
+@pytest.mark.django_db(databases=['default'])
+@patch.object(IsAuthenticatedForInternalAPI, 'has_permission', return_value=True)
+def test_netsutie_fields_view(db, api_client, mocker):
+    url = reverse('accounting-fields')
+
+    response = api_client.get(url)
+    assert response.status_code == 400
+
+    response = api_client.get(url, {'org_id': 'or79Cob97KSh'})
+    assert response.status_code == 400
+
+    response = api_client.get(url, {'org_id': 'or79Cob97KSh', 'resource_type': 'custom_segments'})
+    assert response.status_code == 400
+
+    mocker.patch(
+        'netsuitesdk.api.custom_lists.CustomLists.get',
+        return_value=data['get_custom_list']
+    )
+
+    response = api_client.get(url, {'org_id': 'or79Cob97KSh', 'resource_type': 'custom_lists', 'internal_id': '1'})
+    assert response.status_code == 200
+
+    mocker.patch(
+        'netsuitesdk.api.employees.Employees.get_all_generator',
+        return_value=data['get_all_employees']    
+    )
+
+    response = api_client.get(url, {'org_id': 'or79Cob97KSh', 'resource_type': 'employees'})
+    assert response.status_code == 200
+
+
+@pytest.mark.django_db(databases=['default'])
+@patch.object(IsAuthenticatedForInternalAPI, 'has_permission', return_value=True)
+def test_exported_entry_view(db, api_client, mocker):
+    url = reverse('exported-entry')
+
+    response = api_client.get(url)
+    assert response.status_code == 400
+
+    response = api_client.get(url, {'org_id': 'or79Cob97KSh'})
+    assert response.status_code == 400
+
+    response = api_client.get(url, {'org_id': 'or79Cob97KSh', 'resource_type': 'vendor_bills'})
+    assert response.status_code == 400
+
+    mocker.patch(
+        'netsuitesdk.api.vendor_bills.VendorBills.get',
+        return_value={'summa': 'hehe'}
+    )
+
+    response = api_client.get(url, {'org_id': 'or79Cob97KSh', 'resource_type': 'vendor_bills', 'internal_id': '1'})
+    assert response.status_code == 200