#!/usr/bin/env python3 """ Script to grant users access to Snowflake objects using dataset IDs directly Useful when objects aren't discoverable through the datasets API """ import sys import argparse from sigma_api_client import SigmaAPIClient from user_management import UserManager, PermissionManager, GranteeType, PermissionType from config import ConfigManager def grant_dataset_access(user_email: str, dataset_id: str, permission: str = "view", object_name: str = None): """ Grant a user access to a dataset using its ID directly Args: user_email: Email of the user to grant access dataset_id: Direct dataset ID from Sigma permission: Permission level (view, explore, edit) object_name: Optional friendly name for the object """ try: # Load configuration config_manager = ConfigManager() sigma_config = config_manager.get_sigma_config() # Initialize API client and managers api_client = SigmaAPIClient(sigma_config) user_manager = UserManager(api_client) permission_manager = PermissionManager(api_client) print(f"Granting {permission} access to dataset") print(f"User: {user_email}") # Extract the actual inodeId from the full identifier if '-' in dataset_id: actual_inode_id = dataset_id.split('-', 1)[1] # Get part after first dash object_type = dataset_id.split('-', 1)[0] # Get part before first dash else: actual_inode_id = dataset_id object_type = "table" # Default assumption print(f"Full ID: {dataset_id}") print(f"Extracted inodeId: {actual_inode_id}") print(f"Object type: {object_type}") if object_name: print(f"Object: {object_name}") print("=" * 60) # Find the user user = user_manager.get_member_by_email(user_email) if not user: print(f"❌ User not found: {user_email}") return False user_id = user['memberId'] print(f"Found user: {user.get('firstName', '')} {user.get('lastName', '')} (ID: {user_id})") # Convert permission string to enum permission_type = PermissionType.VIEW if permission.lower() == "explore": permission_type = PermissionType.EXPLORE elif permission.lower() == "edit": permission_type = PermissionType.EDIT # Map object names to inodeTypes inode_type_map = { 'FACT_ANALYTICS': 'scope', 'ORCHARD_APP_REPORTING_V2': 'scope', # Add more mappings as needed } inode_type = inode_type_map.get(object_type, 'scope') # Default to scope # For scope type, use "usage" permission (based on the existing grants) if inode_type == 'scope': permission_value = 'usage' else: permission_value = permission_type.value # Grant access using the grants API directly grant_data = { 'inodeType': inode_type, 'inodeId': actual_inode_id, 'grantee': {'memberId': user_id}, 'permission': permission_value } print(f"Creating grant with permission: {permission_type.value}") print(f"Grant data: {grant_data}") result = api_client.post('/v2/grants', grant_data) print(f"✅ Successfully granted access!") print(f" Grant ID: {result.get('grantId', 'N/A')}") print(f" Permission: {result.get('permission', 'N/A')}") print(f" Resource: {result.get('resource', 'N/A')}") return True except Exception as e: print(f"❌ Error: {e}") # Try to get more detailed error info if hasattr(e, 'response') and hasattr(e.response, 'text'): print(f" Response: {e.response.text}") return False def list_user_grants(user_email: str): """ List all grants for a specific user """ try: config_manager = ConfigManager() sigma_config = config_manager.get_sigma_config() api_client = SigmaAPIClient(sigma_config) user_manager = UserManager(api_client) permission_manager = PermissionManager(api_client) print(f"Listing grants for user: {user_email}") print("=" * 50) # Find the user user = user_manager.get_member_by_email(user_email) if not user: print(f"❌ User not found: {user_email}") return False user_id = user['memberId'] print(f"User: {user.get('firstName', '')} {user.get('lastName', '')} (ID: {user_id})") # Get user's grants try: grants = permission_manager.find_grants_for_user(user_id) if not grants: print("No grants found for this user") return True print(f"\nFound {len(grants)} grants:") for i, grant in enumerate(grants, 1): print(f"{i}. Grant ID: {grant.get('grantId')}") print(f" Resource: {grant.get('resource')}") print(f" Permission: {grant.get('permission')}") print(f" Type: {grant.get('granteeType')}") print() except Exception as e: print(f"Error listing grants: {e}") return False return True except Exception as e: print(f"❌ Error: {e}") return False def main(): parser = argparse.ArgumentParser( description="Grant Sigma users access using dataset IDs directly", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Grant access using dataset ID python grant_by_dataset_id.py grant user@company.com abc123-def456-ghi789 --permission explore --name "FACTS.PROD.FACT_ANALYTICS" # List user's current grants python grant_by_dataset_id.py list user@company.com """ ) parser.add_argument("action", choices=["grant", "list"], help="Action to perform") parser.add_argument("user_email", help="Email of the user") parser.add_argument("dataset_id", nargs="?", help="Dataset ID (required for grant)") parser.add_argument("--permission", "-p", choices=["view", "explore", "edit"], default="view", help="Permission level (default: view)") parser.add_argument("--name", "-n", help="Friendly name for the object") args = parser.parse_args() # Validate arguments if args.action == "grant" and not args.dataset_id: print("❌ Error: Dataset ID is required for grant action") sys.exit(1) # Execute based on action success = False if args.action == "grant": success = grant_dataset_access(args.user_email, args.dataset_id, args.permission, args.name) elif args.action == "list": success = list_user_grants(args.user_email) if success: print("\n🎉 Operation completed successfully!") sys.exit(0) else: print("\n💥 Operation failed") sys.exit(1) if __name__ == "__main__": main()