from typing import List, Dict, Any, Optional from sigma_api_client import SigmaAPIClient from enum import Enum class GranteeType(Enum): USER = "user" TEAM = "team" class PermissionType(Enum): VIEW = "view" EXPLORE = "explore" EDIT = "edit" class UserManager: """ Manages Sigma users and their permissions """ def __init__(self, api_client: SigmaAPIClient): self.api = api_client def list_members(self, limit: int = 50) -> List[Dict[str, Any]]: """ List all members in the organization """ params = {'limit': limit} response = self.api.get('/v2/members', params=params) return response.get('entries', []) def get_member_by_email(self, email: str) -> Optional[Dict[str, Any]]: """ Find a member by email address """ members = self.list_members() for member in members: if member.get('email', '').lower() == email.lower(): return member return None def create_member(self, email: str, first_name: str, last_name: str, account_type: str = "viewer") -> Dict[str, Any]: """ Create a new member Args: email: User's email address first_name: User's first name last_name: User's last name account_type: Account type (viewer, explorer, creator, admin) """ data = { 'email': email, 'firstName': first_name, 'lastName': last_name, 'accountType': account_type } return self.api.post('/v2/members', data) def update_member_account_type(self, member_id: str, account_type: str) -> Dict[str, Any]: """ Update a member's account type Args: member_id: Member's ID account_type: New account type (viewer, explorer, creator, admin) """ data = {'accountType': account_type} return self.api.patch(f'/v2/members/{member_id}', data) def deactivate_member(self, member_id: str) -> bool: """ Deactivate a member """ return self.api.delete(f'/v2/members/{member_id}') class PermissionManager: """ Manages permissions and grants for Sigma objects """ def __init__(self, api_client: SigmaAPIClient): self.api = api_client def list_grants(self, limit: int = 50) -> List[Dict[str, Any]]: """ List all grants in the organization """ params = {'limit': limit} response = self.api.get('/v2/grants', params=params) return response.get('entries', []) def grant_workbook_access(self, workbook_id: str, grantee_id: str, grantee_type: GranteeType, permission: PermissionType) -> Dict[str, Any]: """ Grant access to a workbook Args: workbook_id: ID of the workbook grantee_id: ID of the user or team grantee_type: Type of grantee (user or team) permission: Permission level (view, explore, edit) """ data = { 'resource': workbook_id, 'grantee': grantee_id, 'granteeType': grantee_type.value, 'permission': permission.value } return self.api.post('/v2/grants', data) def grant_workspace_access(self, workspace_id: str, grantee_id: str, grantee_type: GranteeType, permission: PermissionType) -> Dict[str, Any]: """ Grant access to a workspace Args: workspace_id: ID of the workspace grantee_id: ID of the user or team grantee_type: Type of grantee (user or team) permission: Permission level (view, explore, edit) """ data = { 'resource': workspace_id, 'grantee': grantee_id, 'granteeType': grantee_type.value, 'permission': permission.value } return self.api.post('/v2/grants', data) def revoke_access(self, grant_id: str) -> bool: """ Revoke access by deleting a grant Args: grant_id: ID of the grant to delete """ return self.api.delete(f'/v2/grants/{grant_id}') def find_grants_for_user(self, user_id: str) -> List[Dict[str, Any]]: """ Find all grants for a specific user """ grants = self.list_grants() return [grant for grant in grants if grant.get('grantee') == user_id] def find_grants_for_resource(self, resource_id: str) -> List[Dict[str, Any]]: """ Find all grants for a specific resource (workbook/workspace) """ grants = self.list_grants() return [grant for grant in grants if grant.get('resource') == resource_id] class UserAttributeManager: """ Manages user attributes for row-level security """ def __init__(self, api_client: SigmaAPIClient): self.api = api_client def list_user_attributes(self) -> List[Dict[str, Any]]: """ List all user attributes """ response = self.api.get('/v2/user-attributes') return response.get('entries', []) def create_user_attribute(self, name: str, description: str = "") -> Dict[str, Any]: """ Create a new user attribute Args: name: Name of the user attribute description: Optional description """ data = { 'name': name, 'description': description } return self.api.post('/v2/user-attributes', data) def assign_user_attribute(self, attribute_id: str, user_id: str, value: str) -> Dict[str, Any]: """ Assign a user attribute value to a user Args: attribute_id: ID of the user attribute user_id: ID of the user value: Value to assign """ data = { 'userId': user_id, 'value': value } return self.api.post(f'/v2/user-attributes/{attribute_id}/assignments', data)