Require Admin function decorators enhance

TM1py/Exceptions/Exceptions.py

@@ -39,6 +39,26 @@ def __init__(self, function: str):
     def __str__(self):
         return f"Function '{self.function}' requires admin permissions"
 
+class TM1pyNotDataAdminException(Exception):
+    def __init__(self, function: str):
+        self.function = function
+
+    def __str__(self):
+        return f"Function '{self.function}' requires DataAdmin permissions"
+
+class TM1pyNotSecurityAdminException(Exception):
+    def __init__(self, function: str):
+        self.function = function
+
+    def __str__(self):
+        return f"Function '{self.function}' requires SecurityAdmin permissions"
+
+class TM1pyNotOpsAdminException(Exception):
+    def __init__(self, function: str):
+        self.function = function
+
+    def __str__(self):
+        return f"Function '{self.function}' requires OperationsAdmin permissions"
 
 class TM1pyException(Exception):
     """ The default exception for TM1py

TM1py/Objects/User.py

@@ -76,6 +76,21 @@ def password(self) -> str:
     @property
     def is_admin(self) -> bool:
         return "ADMIN" in CaseAndSpaceInsensitiveSet(*self.groups)
+
+    @property
+    def is_data_admin(self) -> bool:
+        return any(g in CaseAndSpaceInsensitiveSet(
+            *self.groups) for g in ["Admin", "DataAdmin"])
+
+    @property
+    def is_security_admin(self) -> bool:
+        return any(g in CaseAndSpaceInsensitiveSet(
+            *self.groups) for g in ["Admin", "SecurityAdmin"])
+
+    @property
+    def is_ops_admin(self) -> bool:
+        return any(g in CaseAndSpaceInsensitiveSet(
+            *self.groups) for g in ["Admin", "OperationsAdmin"])
 
     @property
     def groups(self) -> List[str]:

TM1py/Services/CellService.py

@@ -32,7 +32,7 @@
 from TM1py.Utils.Utils import build_pandas_dataframe_from_cellset, dimension_name_from_element_unique_name, \
     CaseAndSpaceInsensitiveDict, wrap_in_curly_braces, CaseAndSpaceInsensitiveTuplesDict, \
     abbreviate_mdx, build_csv_from_cellset_dict, require_version, require_pandas, build_cellset_from_pandas_dataframe, \
-    case_and_space_insensitive_equals, get_cube, resembles_mdx, require_admin, extract_compact_json_cellset, \
+    case_and_space_insensitive_equals, get_cube, resembles_mdx, require_data_admin, require_ops_admin, extract_compact_json_cellset, \
     cell_is_updateable, build_mdx_from_cellset, build_mdx_and_values_from_cellset, \
     dimension_names_from_element_unique_names, frame_to_significant_digits, build_dataframe_from_csv, \
     drop_dimension_properties, decohints, verify_version
@@ -525,7 +525,8 @@ def clear_spread(
         return self._post_against_cellset(cellset_id=cellset_id, payload=payload, delete_cellset=True,
                                           sandbox_name=sandbox_name, **kwargs)
 
-    @require_admin
+    @require_data_admin
+    @require_ops_admin
     @require_version(version="11.7")
     def clear(self, cube: str, **kwargs):
         """
@@ -566,7 +567,8 @@ def clear(self, cube: str, **kwargs):
 
         return self.clear_with_mdx(cube=cube, mdx=mdx_builder.to_mdx(), **kwargs)
 
-    @require_admin
+    @require_data_admin
+    @require_ops_admin
     @require_version(version="11.7")
     def clear_with_mdx(self, cube: str, mdx: str, sandbox_name: str = None, **kwargs):
         """ clear a slice in a cube based on an MDX query.
@@ -956,7 +958,8 @@ def drop_non_updateable_cells(self, cells: Dict, cube_name: str, dimensions: Lis
                 updateable_cells[elements] = cells[elements]
         return updateable_cells
 
-    @require_admin
+    @require_data_admin
+    @require_ops_admin
     @manage_transaction_log
     def write_through_unbound_process(self, cube_name: str, cellset_as_dict: Dict, increment: bool = False,
                                       sandbox_name: str = None, precision: int = None,
@@ -1042,7 +1045,8 @@ def write_through_unbound_process(self, cube_name: str, cellset_as_dict: Dict, i
         if not all(successes):
             raise TM1pyWritePartialFailureException(statuses, log_files, len(successes))
 
-    @require_admin
+    @require_data_admin
+    @require_ops_admin
     @manage_transaction_log
     @require_pandas
     def write_through_blob(self, cube_name: str, cellset_as_dict: dict, increment: bool = False,
@@ -3801,7 +3805,8 @@ def _get_attributes_by_dimension(self, cube: str, **kwargs) -> Dict[str, List[st
 
         return attributes_by_dimension
 
-    @require_admin
+    @require_data_admin
+    @require_ops_admin
     def _execute_view_csv_use_blob(self, cube_name: str, view_name: str, top: int, skip: int, skip_zeros: bool,
                                    skip_consolidated_cells: bool, skip_rule_derived_cells: bool,
                                    value_separator: str, cube_dimensions: List[str] = None,
@@ -3932,7 +3937,8 @@ def _execute_view_csv_use_blob(self, cube_name: str, view_name: str, top: int, s
             with suppress(Exception):
                 file_service.delete(file_name)
 
-    @require_admin
+    @require_data_admin
+    @require_ops_admin
     def _execute_mdx_csv_use_blob(self, mdx: Union[str, MdxBuilder], top: int, skip: int, skip_zeros: bool,
                                   skip_consolidated_cells: bool, skip_rule_derived_cells: bool,
                                   value_separator: str, cube_dimensions: List[str] = None,

TM1py/Services/CubeService.py

@@ -10,7 +10,7 @@
 from TM1py.Services.ObjectService import ObjectService
 from TM1py.Services.RestService import RestService
 from TM1py.Services.ViewService import ViewService
-from TM1py.Utils import format_url, require_version, require_admin, case_and_space_insensitive_equals
+from TM1py.Utils import format_url, require_version, require_data_admin, case_and_space_insensitive_equals
 
 
 class CubeService(ObjectService):
@@ -135,7 +135,7 @@ def check_rules(self, cube_name: str, **kwargs) -> Response:
         errors = response.json()["value"]
         return errors
 
-    @require_admin
+    @require_data_admin
     def delete(self, cube_name: str, **kwargs) -> Response:
         """ Delete a cube in TM1
 
@@ -289,7 +289,7 @@ def get_storage_dimension_order(self, cube_name: str, **kwargs) -> List[str]:
         response = self._rest.GET(url, **kwargs)
         return [dimension["Name"] for dimension in response.json()["value"]]
 
-    @require_admin
+    @require_data_admin
     @require_version(version="11.4")
     def update_storage_dimension_order(self, cube_name: str, dimension_names: Iterable[str]) -> float:
         """ Update the storage dimension order of a cube
@@ -306,7 +306,7 @@ def update_storage_dimension_order(self, cube_name: str, dimension_names: Iterab
         response = self._rest.POST(url=url, data=json.dumps(payload))
         return response.json()["value"]
 
-    @require_admin
+    @require_data_admin
     @require_version(version="11.6")
     def load(self, cube_name: str, **kwargs) -> Response:
         """ Load the cube into memory on the server
@@ -317,7 +317,7 @@ def load(self, cube_name: str, **kwargs) -> Response:
         url = format_url("/Cubes('{}')/tm1.Load", cube_name)
         return self._rest.POST(url=url, **kwargs)
 
-    @require_admin
+    @require_data_admin
     @require_version(version="11.6")
     def unload(self, cube_name: str, **kwargs) -> Response:
         """ Unload the cube from memory
@@ -346,7 +346,7 @@ def unlock(self, cube_name: str, **kwargs) -> Response:
         url = format_url("/Cubes('{}')/tm1.Unlock", cube_name)
         return self._rest.POST(url=url, **kwargs)
 
-    @require_admin
+    @require_data_admin
     def cube_save_data(self, cube_name: str, **kwargs) -> Response:
         """ Serializes a cube by saving data updates
 

TM1py/Services/ElementService.py

@@ -19,7 +19,7 @@
 from TM1py.Objects import ElementAttribute, Element
 from TM1py.Services.ObjectService import ObjectService
 from TM1py.Services.RestService import RestService
-from TM1py.Utils import CaseAndSpaceInsensitiveDict, format_url, CaseAndSpaceInsensitiveSet, require_admin, \
+from TM1py.Utils import CaseAndSpaceInsensitiveDict, format_url, CaseAndSpaceInsensitiveSet, require_data_admin, \
     dimension_hierarchy_element_tuple_from_unique_name, require_pandas, require_version
 from TM1py.Utils import build_element_unique_names, CaseAndSpaceInsensitiveTuplesDict, verify_version
 from itertools import islice
@@ -1189,7 +1189,7 @@ def hierarchy_exists(self, dimension_name, hierarchy_name):
         hierarchy_service = self._get_hierarchy_service()
         return hierarchy_service.exists(dimension_name, hierarchy_name)
 
-    @require_admin
+    @require_data_admin
     def _element_is_ancestor_ti(self, dimension_name: str, hierarchy_name: str, element_name: str,
                                 ancestor_name: str) -> bool:
         process_service = self.get_process_service()

TM1py/Services/HierarchyService.py

@@ -21,7 +21,8 @@
 from TM1py.Services.RestService import RestService
 from TM1py.Services.SubsetService import SubsetService
 from TM1py.Utils.Utils import case_and_space_insensitive_equals, format_url, CaseAndSpaceInsensitiveDict, \
-    CaseAndSpaceInsensitiveSet, CaseAndSpaceInsensitiveTuplesDict, require_pandas, require_admin, verify_version
+    CaseAndSpaceInsensitiveSet, CaseAndSpaceInsensitiveTuplesDict, require_pandas, require_data_admin, \
+    require_ops_admin, verify_version
 
 
 class HierarchyService(ObjectService):
@@ -415,7 +416,8 @@ def is_balanced(self, dimension_name: str, hierarchy_name: str, **kwargs):
             raise RuntimeError(f"Unexpected return value from TM1 API request: {str(structure)}")
 
     @require_pandas
-    @require_admin
+    @require_data_admin
+    @require_ops_admin
     def update_or_create_hierarchy_from_dataframe(
             self,
             dimension_name: str,

TM1py/Services/ObjectService.py

@@ -74,3 +74,15 @@ def version(self) -> str:
     @property
     def is_admin(self) -> bool:
         return self._rest.is_admin
+
+    @property
+    def is_data_admin(self) -> bool:
+        return self._rest.is_data_admin
+
+    @property
+    def is_security_admin(self) -> bool:
+        return self._rest.is_security_admin
+
+    @property
+    def is_ops_admin(self) -> bool:
+        return self._rest.is_ops_admin

TM1py/Services/ProcessService.py

@@ -13,7 +13,7 @@
 from TM1py.Objects.ProcessDebugBreakpoint import ProcessDebugBreakpoint
 from TM1py.Services.ObjectService import ObjectService
 from TM1py.Services.RestService import RestService
-from TM1py.Utils import format_url, require_admin
+from TM1py.Utils import format_url, require_data_admin
 from TM1py.Utils.Utils import require_version, deprecated_in_version
 
 
@@ -349,7 +349,7 @@ def _execute_with_return_parse_response(self, response):
             "Filename"]
         return success, status, error_log_file
 
-    @require_admin
+    @require_data_admin
     def execute_ti_code(self, lines_prolog: Iterable[str], lines_epilog: Iterable[str] = None, **kwargs) -> Response:
         """ Execute lines of code on the TM1 Server
 
@@ -628,7 +628,7 @@ def debug_get_current_breakpoint(self, debug_id: str, **kwargs) -> ProcessDebugB
         response = self._rest.GET(url=url, **kwargs)
         return ProcessDebugBreakpoint.from_dict(response.json()["CurrentBreakpoint"])
 
-    @require_admin
+    @require_data_admin
     def evaluate_boolean_ti_expression(self, formula: str):
         prolog_procedure = f"""
         if (~{formula.strip(";")});
@@ -645,7 +645,7 @@ def evaluate_boolean_ti_expression(self, formula: str):
         else:
             raise TM1pyException(f"Unexpected TI return status: '{status}' for expression: '{formula}'")
 
-    @require_admin
+    @require_data_admin
     def evaluate_ti_expression(self, formula: str, **kwargs) -> str:
         """ This function is same functionality as hitting "Evaluate" within variable formula editor in TI
             Function creates temporary TI and then starts a debug session on that TI

TM1py/Services/RestService.py

@@ -165,6 +165,18 @@ def __init__(self, **kwargs):
         # populated later on the fly for users with the name different from 'Admin'
         self._is_admin = self._determine_is_admin(kwargs.get('user', None))
 
+        # populated on the fly
+        if kwargs.get('user'):
+            self._is_admin = True if case_and_space_insensitive_equals(kwargs.get('user'), 'ADMIN') else None
+            self._is_data_admin = True if case_and_space_insensitive_equals(kwargs.get('user'), 'ADMIN') else None
+            self._is_security_admin = True if case_and_space_insensitive_equals(kwargs.get('user'), 'ADMIN') else None
+            self._is_ops_admin = True if case_and_space_insensitive_equals(kwargs.get('user'), 'ADMIN') else None
+        else:
+            self._is_admin = None
+            self._is_data_admin = None
+            self._is_security_admin = None
+            self._is_ops_admin = None
+
         self._verify = self._determine_verify(kwargs.get('verify', None))
 
         self._base_url, self._auth_url = self._construct_service_and_auth_root()
@@ -788,6 +800,33 @@ def is_admin(self) -> bool:
 
         return self._is_admin
 
+    @property
+    def is_data_admin(self) -> bool:
+        if self._is_data_admin is None:
+            response = self.GET("ActiveUser/Groups")
+            self._is_data_admin = any(g in CaseAndSpaceInsensitiveSet(
+                *[group["Name"] for group in response.json()["value"]]) for g in ["Admin", "DataAdmin"])
+
+        return self._is_data_admin
+
+    @property
+    def is_security_admin(self) -> bool:
+        if self._is_security_admin is None:
+            response = self.GET("/ActiveUser/Groups")
+            self._is_security_admin = any(g in CaseAndSpaceInsensitiveSet(
+                *[group["Name"] for group in response.json()["value"]]) for g in ["Admin", "SecurityAdmin"])
+
+        return self._is_security_admin
+
+    @property
+    def is_ops_admin(self) -> bool:
+        if self._is_ops_admin is None:
+            response = self.GET("/ActiveUser/Groups")
+            self._is_ops_admin = any(g in CaseAndSpaceInsensitiveSet(
+                *[group["Name"] for group in response.json()["value"]]) for g in ["Admin", "OperationsAdmin"])
+
+        return self._is_ops_admin
+
     @property
     def sandboxing_disabled(self):
         if self._sandboxing_disabled is None:

TM1py/Services/SecurityService.py

@@ -8,7 +8,7 @@
 from TM1py.Objects.User import User
 from TM1py.Services.ObjectService import ObjectService
 from TM1py.Services.RestService import RestService
-from TM1py.Utils.Utils import format_url, CaseAndSpaceInsensitiveSet, require_admin
+from TM1py.Utils.Utils import format_url, CaseAndSpaceInsensitiveSet, require_security_admin, require_admin
 
 
 class SecurityService(ObjectService):
@@ -25,7 +25,7 @@ def determine_actual_user_name(self, user_name: str, **kwargs) -> str:
     def determine_actual_group_name(self, group_name: str, **kwargs) -> str:
         return self.determine_actual_object_name(object_class="Groups", object_name=group_name, **kwargs)
 
-    @require_admin
+    @require_security_admin
     def create_user(self, user: User, **kwargs) -> Response:
         """ Create a user on TM1 Server
 
@@ -35,7 +35,7 @@ def create_user(self, user: User, **kwargs) -> Response:
         url = '/Users'
         return self._rest.POST(url, user.body, **kwargs)
 
-    @require_admin
+    @require_security_admin
     def create_group(self, group_name: str, **kwargs) -> Response:
         """ Create a Security group in the TM1 Server
 
@@ -67,7 +67,7 @@ def get_current_user(self, **kwargs) -> User:
         response = self._rest.GET(url, **kwargs)
         return User.from_dict(response.json())
 
-    @require_admin
+    @require_security_admin
     def update_user(self, user: User, **kwargs) -> Response:
         """ Update user on TM1 Server
 
@@ -86,7 +86,7 @@ def update_user_password(self, user_name: str, password: str, **kwargs) -> Respo
         body = {"Password": password}
         return self._rest.PATCH(url, json.dumps(body), **kwargs)
 
-    @require_admin
+    @require_security_admin
     def delete_user(self, user_name: str, **kwargs) -> Response:
         """ Delete user on TM1 Server
 
@@ -97,7 +97,7 @@ def delete_user(self, user_name: str, **kwargs) -> Response:
         url = format_url("/Users('{}')", user_name)
         return self._rest.DELETE(url, **kwargs)
 
-    @require_admin
+    @require_security_admin
     def delete_group(self, group_name: str, **kwargs) -> Response:
         """ Delete a group in the TM1 Server
 
@@ -163,7 +163,7 @@ def get_groups(self, user_name: str, **kwargs) -> List[str]:
         response = self._rest.GET(url, **kwargs)
         return [group['Name'] for group in response.json()['value']]
 
-    @require_admin
+    @require_security_admin
     def add_user_to_groups(self, user_name: str, groups: Iterable[str], **kwargs) -> Response:
         """
 
@@ -182,7 +182,7 @@ def add_user_to_groups(self, user_name: str, groups: Iterable[str], **kwargs) ->
         }
         return self._rest.PATCH(url, json.dumps(body), **kwargs)
 
-    @require_admin
+    @require_security_admin
     def remove_user_from_group(self, group_name: str, user_name: str, **kwargs) -> Response:
         """ Remove user from group in TM1 Server