From 7d27cd57e0d98d3b8a10767cfe39527fa31bfb4e Mon Sep 17 00:00:00 2001 From: GitHub Actions Date: Thu, 12 Sep 2024 15:22:55 +0530 Subject: [PATCH] Field mapping changes --- apps/spotlight/service.py | 87 ++++++++++++++++++++++++++++++++++++--- 1 file changed, 82 insertions(+), 5 deletions(-) diff --git a/apps/spotlight/service.py b/apps/spotlight/service.py index 4c21561..7796349 100644 --- a/apps/spotlight/service.py +++ b/apps/spotlight/service.py @@ -1,5 +1,6 @@ from dataclasses import dataclass from typing import Callable, Dict +import json import requests @@ -10,9 +11,6 @@ from . import llm -code_action_map = { - "trigger_export": 'http://localhost:8000/api/workspaces/2/trigger_export/' -} @dataclass @@ -66,7 +64,11 @@ class ActionService: @classmethod def _get_action_function_from_code(cls, *, code: str) -> Callable: code_to_function_map = { - "trigger_export": cls.trigger_export + "trigger_export": cls.trigger_export, + "set_customer_field_mapping_to_project": cls.set_customer_field_mapping_to_project, + "set_customer_field_mapping_to_cost_center": cls.set_customer_field_mapping_to_cost_center, + "set_class_field_mapping_to_project": cls.set_class_field_mapping_to_project, + "set_class_field_mapping_to_cost_center": cls.set_class_field_mapping_to_cost_center } return code_to_function_map[code] @@ -97,6 +99,81 @@ def trigger_export(cls, *, workspace_id: int): return ActionResponse(message="Export triggered failed", is_success=False) @classmethod - def action(cls, *, code: str, workspace_id: str) -> ActionResponse: + def set_customer_field_mapping_to_project(cls, *, workspace_id: int): + access_token = cls.get_access_token(workspace_id=workspace_id) + headers = cls.get_headers(access_token=access_token) + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json" + } + url = f'http://localhost:8000/api/workspaces/{workspace_id}/field_mappings/' + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json" + } + action_response = requests.get(url, headers=headers) + action_response= action_response.json() + if action_response.get('project_type') != 'PROJECT' and action_response.get('class_type') != 'COST_CENTER': + action_response['project_type'] = 'PROJECT' + post_response = requests.post(url, headers=headers, data=json.dumps(action_response)) + return ActionResponse(message="Field mapping updated successfully", is_success=True) + return ActionResponse(message="Field mapping already exists", is_success=False) + + @classmethod + def set_customer_field_mapping_to_cost_center(cls, *, workspace_id: int): + access_token = cls.get_access_token(workspace_id=workspace_id) + headers = cls.get_headers(access_token=access_token) + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json" + } + url = f'http://localhost:8000/api/workspaces/{workspace_id}/field_mappings/' + + action_response = requests.get(url, headers=headers) + action_response= action_response.json() + if action_response.get('project_type') != 'COST_CENTER' and action_response.get('class_type') != 'PROJECT': + action_response['project_type'] = 'COST_CENTER' + post_response = requests.post(url, headers=headers, data=json.dumps(action_response)) + return ActionResponse(message="Field mapping updated successfully", is_success=True) + return ActionResponse(message="Field mapping already exists", is_success=False) + + @classmethod + def set_class_field_mapping_to_project(cls, *, workspace_id: int): + access_token = cls.get_access_token(workspace_id=workspace_id) + headers = cls.get_headers(access_token=access_token) + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json" + } + url = f'http://localhost:8000/api/workspaces/{workspace_id}/field_mappings/' + + action_response = requests.get(url, headers=headers) + action_response= action_response.json() + if action_response.get('project_type') != 'PROJECT' and action_response.get('class_type') != 'COST_CENTER': + action_response['class_type'] = 'PROJECT' + post_response = requests.post(url, headers=headers, data=json.dumps(action_response)) + return ActionResponse(message="Field mapping updated successfully", is_success=True) + return ActionResponse(message="Field mapping already exists", is_success=False) + + @classmethod + def set_class_field_mapping_to_cost_center(cls, *, workspace_id: int): + access_token = cls.get_access_token(workspace_id=workspace_id) + headers = cls.get_headers(access_token=access_token) + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json" + } + url = f'http://localhost:8000/api/workspaces/{workspace_id}/field_mappings/' + + action_response = requests.get(url, headers=headers) + action_response= action_response.json() + if action_response.get('project_type') != 'COST_CENTER' and action_response.get('class_type') != 'PROJECT': + action_response['class_type'] = 'COST_CENTER' + post_response = requests.post(url, headers=headers, data=json.dumps(action_response)) + return ActionResponse(message="Field mapping updated successfully", is_success=True) + return ActionResponse(message="Field mapping already exists", is_success=False) + + @classmethod + def action(cls, *, code: str, workspace_id: str): action_function = cls._get_action_function_from_code(code=code) return action_function(workspace_id=workspace_id)