From f5cda6f1214c196ecf2671c0568fbc1fc34f8fea Mon Sep 17 00:00:00 2001 From: Aastha Gupta Date: Mon, 6 May 2024 13:14:25 +0530 Subject: [PATCH] fixed review comments --- wandb-scim/custom_roles.py | 54 +++++++++---------- wandb-scim/examples/custom_roles_example.py | 60 +++++++++------------ wandb-scim/examples/teams_example.py | 44 +++++++-------- wandb-scim/examples/user_example.py | 51 ++++++++---------- wandb-scim/teams.py | 35 ++++++------ wandb-scim/users.py | 40 +++++++------- 6 files changed, 134 insertions(+), 150 deletions(-) diff --git a/wandb-scim/custom_roles.py b/wandb-scim/custom_roles.py index 09b8389d..3f0a6c5d 100644 --- a/wandb-scim/custom_roles.py +++ b/wandb-scim/custom_roles.py @@ -1,29 +1,31 @@ -import requests import base64 +import json +import requests class CustomRole(object): - def __init__(self, username, api_key): + def __init__(self, base_url, username, api_key): """ Initializes the CustomRole object with username and API key. Args: + base_url (str): Host url. username (str): The username for authentication. api_key (str): The API key for authentication. """ # Encode the username and API key into a base64-encoded string for Basic Authentication auth_str = f"{username}:{api_key}" auth_bytes = auth_str.encode('ascii') + base_url = base_url self.auth_token = base64.b64encode(auth_bytes).decode('ascii') # Create the authorization header for API requests self.authorization_header = f"Basic {self.auth_token}" - def _create_custom_role(self, url, request_payload): + def create(self, request_payload): """ Creates a new custom role. Args: - url (str): The URL for the custom role creation endpoint. request_payload (dict): The payload containing custom role data. It should contain the following keys: - 'permissionJson': The permissions JSON for the custom role. @@ -44,18 +46,19 @@ def _create_custom_role(self, url, request_payload): "Authorization": self.authorization_header, "Content-Type": "application/json" } + url = f"{self.base_url}/scim/Roles" response = requests.post(url, json=data, headers=headers) if response.status_code == 201: return "Custom role has been created!" return f"Custom role creation failed. Status code: {response.status_code}" - def _get_custom_role(self, url): + def get(self, role_id): """ - Retrieves custom role details. + Retrieves custom role details for role_id. Args: - url (str): The URL for the custom role retrieval endpoint. + role_id (str): role_id from the custom role. Returns: str: A message containing custom role details or indicating failure. @@ -66,20 +69,17 @@ def _get_custom_role(self, url): "Authorization": self.authorization_header, "Content-Type": "application/json" } - + url = f"{self.base_url}/scim/Roles/{role_id}" response = requests.get(url, headers=headers) if response.status_code == 200: - return f"Custom role details: {response.text}" + return json.dumps(response.json, indent=4) return f"Get custom role failed. Status code: {response.status_code}" - def _get_all_custom_role(self, url): + def get_all(self): """ Retrieves details of all custom roles in the organization. - Args: - url (str): The URL for the endpoint to get all custom roles. - Returns: str: A message containing details of all custom roles or indicating failure. """ @@ -89,19 +89,19 @@ def _get_all_custom_role(self, url): "Authorization": self.authorization_header, "Content-Type": "application/json" } - + url = f"{self.base_url}/scim/Roles" response = requests.get(url, headers=headers) if response.status_code == 200: - return f"All the custom roles details: {response.text}" + return json.dumps(response.json, indent=4) return f"Get all custom roles failed. Status code: {response.status_code}" - def _add_permission(self, url, request_payload): + def add_permissions(self, role_id, request_payload): """ Adds permission to a custom role. Args: - url (str): The URL for the endpoint to add permission to the custom role. + role_id (str): role_id from the custom role. request_payload (dict): The payload containing permission information. It should contain the following key: - 'permissionJson': The permissions JSON to be added to the custom role. @@ -120,7 +120,7 @@ def _add_permission(self, url, request_payload): } ] } - + url = f"{self.base_url}/scim/Roles/{role_id}" headers = { "Authorization": self.authorization_header, "Content-Type": "application/json" @@ -138,12 +138,12 @@ def _add_permission(self, url, request_payload): else: return f"Failed to update custom role. Status code: {response.status_code}" - def _remove_permission(self, url, request_payload): + def remove_permission(self, role_id, request_payload): """ Removes permission from a custom role. Args: - url (str): The URL for the endpoint to remove permission from the custom role. + role_id (str): role_id from the custom role. request_payload (dict): The payload containing permission information. It should contain the following key: - 'permissionJson': The permissions JSON to be removed from the custom role. @@ -166,7 +166,7 @@ def _remove_permission(self, url, request_payload): "Authorization": self.authorization_header, "Content-Type": "application/json" } - + url = f"{self.base_url}/scim/Roles/{role_id}" response = requests.patch(url, json=data, headers=headers) if response.status_code == 200: @@ -179,12 +179,12 @@ def _remove_permission(self, url, request_payload): else: return f"Failed to update custom role. Status code: {response.status_code}" - def _update_custom_role(self, url, request_payload): + def update_custom_role(self, role_id, request_payload): """ Updates name and description of a custom role. Args: - url (str): The URL for the endpoint to update the custom role. + role_id (str): role_id from the custom role. request_payload (dict): The payload containing role information. It should contain the following keys: - 'roleName': The name of the custom role. @@ -205,7 +205,7 @@ def _update_custom_role(self, url, request_payload): "Authorization": self.authorization_header, "Content-Type": "application/json" } - + url = f"{self.base_url}/scim/Roles/{role_id}" response = requests.put(url, json=data, headers=headers) if response.status_code == 200: @@ -218,12 +218,12 @@ def _update_custom_role(self, url, request_payload): else: return f"Failed to update custom role. Status code: {response.status_code}" - def _delete_custom_role(self, url): + def delete_custom_role(self, role_id): """ Deletes a custom role. Args: - url (str): The URL for the endpoint to delete the custom role. + role_id (str): role_id from the custom role. Returns: str: A message indicating whether the custom role deletion was successful or failed. @@ -234,7 +234,7 @@ def _delete_custom_role(self, url): "Authorization": self.authorization_header, "Content-Type": "application/json" } - + url = f"{self.base_url}/scim/Roles/{role_id}" response = requests.delete(url, headers=headers) if response.status_code == 204: diff --git a/wandb-scim/examples/custom_roles_example.py b/wandb-scim/examples/custom_roles_example.py index 284ff561..cdd6f1c6 100644 --- a/wandb-scim/examples/custom_roles_example.py +++ b/wandb-scim/examples/custom_roles_example.py @@ -5,20 +5,18 @@ sys.path.append('../') from custom_roles import CustomRole # Assuming CustomRole class is defined in custom_role.py -def create_custom_role(base_url, custom_role, permission_json, inherited_from): +def create_custom_role(custom_role, permission_json, inherited_from): """ Creates a new custom role. Args: - base_url (str): The base URL of the API. custom_role (CustomRole): An instance of the CustomRole class. permission_json (str): JSON string representing permissions for the custom role. inherited_from (str): The source from which the custom role inherits permissions. """ try: # Create a new custom role - create_role_response = custom_role._create_custom_role( - url=f"{base_url}/Roles", + create_role_response = custom_role.create( request_payload={ "permissionJson": permission_json, "inheritedFrom": inherited_from @@ -28,51 +26,48 @@ def create_custom_role(base_url, custom_role, permission_json, inherited_from): except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def get_custom_role(base_url, custom_role, role_id): +def get_custom_role(custom_role, role_id): """ Retrieves details of a specific custom role. Args: - base_url (str): The base URL of the API. custom_role (CustomRole): An instance of the CustomRole class. role_id (str): The ID of the custom role to retrieve. """ try: # Get details of a custom role - get_role_response = custom_role._get_custom_role(f"{base_url}/Roles/{role_id}") + get_role_response = custom_role.get(role_id) print(get_role_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def get_all_roles(base_url, custom_role): +def get_all_roles(custom_role): """ Retrieves details of all custom roles. Args: - base_url (str): The base URL of the API. custom_role (CustomRole): An instance of the CustomRole class. """ try: # Get all custom roles - get_all_roles_response = custom_role._get_all_custom_role(f"{base_url}/Roles") + get_all_roles_response = custom_role.get_all() print(get_all_roles_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def add_permission(base_url, custom_role, role_id, permission_json): +def add_permission(custom_role, role_id, permission_json): """ Adds permission to a custom role. Args: - base_url (str): The base URL of the API. custom_role (CustomRole): An instance of the CustomRole class. role_id (str): The ID of the custom role to update. permission_json (str): JSON string representing the permission to add. """ try: # Add permission to a custom role - update_role_response = custom_role._add_permission( - url=f"{base_url}/Roles/{role_id}", + update_role_response = custom_role.add_permissions( + role_id, request_payload={ "permissionJson": permission_json } @@ -81,20 +76,19 @@ def add_permission(base_url, custom_role, role_id, permission_json): except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def remove_permission(base_url, custom_role, role_id, permission_json): +def remove_permission(custom_role, role_id, permission_json): """ Removes permission from a custom role. Args: - base_url (str): The base URL of the API. custom_role (CustomRole): An instance of the CustomRole class. role_id (str): The ID of the custom role to update. permission_json (str): JSON string representing the permission to remove. """ try: # Remove permission from a custom role - remove_role_response = custom_role._remove_permission( - url=f"{base_url}/Roles/{role_id}", + remove_role_response = custom_role.remove_permission( + role_id, request_payload={ "permissionJson": permission_json } @@ -103,12 +97,11 @@ def remove_permission(base_url, custom_role, role_id, permission_json): except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def update_custom_role(base_url, custom_role, role_id, role_name, role_description, inherited_from): +def update_custom_role(custom_role, role_id, role_name, role_description, inherited_from): """ Updates a custom role. Args: - base_url (str): The base URL of the API. custom_role (CustomRole): An instance of the CustomRole class. role_id (str): The ID of the custom role to update. role_name (str): The updated name of the custom role. @@ -117,8 +110,8 @@ def update_custom_role(base_url, custom_role, role_id, role_name, role_descripti """ try: # Update a custom role - update_custom_role_response = custom_role._update_custom_role( - url=f"{base_url}/Roles/{role_id}", + update_custom_role_response = custom_role.update_custom_role( + role_id, request_payload={ "roleName": role_name, "roleDescription": role_description, @@ -129,20 +122,17 @@ def update_custom_role(base_url, custom_role, role_id, role_name, role_descripti except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def delete_custom_role(base_url, custom_role, role_id): +def delete_custom_role(custom_role, role_id): """ Deletes a custom role. Args: - base_url (str): The base URL of the API. custom_role (CustomRole): An instance of the CustomRole class. role_id (str): The ID of the custom role to delete. """ try: # Delete the custom role - delete_role_response = custom_role._delete_custom_role( - url=f"{base_url}/Roles/{role_id}" - ) + delete_role_response = custom_role.delete_custom_role(role_id) print(delete_role_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") @@ -153,15 +143,15 @@ def delete_custom_role(base_url, custom_role, role_id): base_url = "https://example.com/api" # Base URL of the API # Instantiate the CustomRole class with your credentials - custom_role = CustomRole(username, api_key) + custom_role = CustomRole(base_url, username, api_key) role_id = "abc" # Replace with actual role ID permission_json = {"name": "project:update"} # Replace with required permissions # Test Functions - get_all_roles(base_url, custom_role) - # create_custom_role(base_url, custom_role, permission_json, "member") - # get_custom_role(base_url, custom_role, role_id) - # add_permission(base_url, custom_role, role_id, permission_json) - # remove_permission(base_url, custom_role, role_id, permission_json) - # update_custom_role(base_url, custom_role, role_id, "test-role", "sample test role description", "member") - # delete_custom_role(base_url,custom_role,role_id) \ No newline at end of file + get_all_roles(custom_role) + # create_custom_role(custom_role, permission_json, "member") + # get_custom_role(custom_role, role_id) + # add_permission(custom_role, role_id, permission_json) + # remove_permission(custom_role, role_id, permission_json) + # update_custom_role(custom_role, role_id, "test-role", "sample test role description", "member") + # delete_custom_role(custom_role, role_id) \ No newline at end of file diff --git a/wandb-scim/examples/teams_example.py b/wandb-scim/examples/teams_example.py index 00ba3363..59d1fa8c 100644 --- a/wandb-scim/examples/teams_example.py +++ b/wandb-scim/examples/teams_example.py @@ -5,20 +5,18 @@ sys.path.append('../') from teams import Teams # Assuming the Teams class is defined in teams.py -def create_team(base_url, teams, display_name, member_id): +def create_team(teams, display_name, member_id): """ Creates a new team. Args: - base_url (str): The base URL of the API. teams (Teams): An instance of the Teams class. display_name (str): The display name of the new team. member_id (str): The ID of the member to be added to the team. """ try: # Create a new team - create_team_response = teams._create_team( - url=f"{base_url}/Groups", + create_team_response = teams.create( request_payload={ "displayName": display_name, "member": member_id @@ -28,71 +26,67 @@ def create_team(base_url, teams, display_name, member_id): except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def get_team(base_url, teams, team_id): +def get_team(teams, team_id): """ Retrieves details of a specific team. Args: - base_url (str): The base URL of the API. teams (Teams): An instance of the Teams class. team_id (str): The ID of the team to retrieve. """ try: # Get team details - get_team_response = teams._get_team(f"{base_url}/Groups/{team_id}") + get_team_response = teams.get(team_id) print(get_team_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def get_all_teams(base_url, teams): +def get_all_teams(teams): """ Retrieves details of all teams in the organization. Args: - base_url (str): The base URL of the API. teams (Teams): An instance of the Teams class. """ try: # Get all teams in the organization - get_all_teams_response = teams._get_all_teams(f"{base_url}/Groups") + get_all_teams_response = teams.get_all() print(get_all_teams_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def update_team_add_member(base_url, teams, team_id, member_id): +def update_team_add_member(teams, team_id, member_id): """ Updates a team by adding a member. Args: - base_url (str): The base URL of the API. teams (Teams): An instance of the Teams class. team_id (str): The ID of the team to update. member_id (str): The ID of the member to add to the team. """ try: # Update team by adding a member - update_team_response = teams._add_team( - url=f"{base_url}/Groups/{team_id}", + update_team_response = teams.update( + team_id, request_payload={"value": member_id} ) print(update_team_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def update_team_remove_member(base_url, teams, team_id, member_id): +def update_team_remove_member(teams, team_id, member_id): """ Updates a team by removing a member. Args: - base_url (str): The base URL of the API. teams (Teams): An instance of the Teams class. team_id (str): The ID of the team to update. member_id (str): The ID of the member to remove from the team. """ try: # Update team by removing a member - update_team_response = teams._remove_team( - url=f"{base_url}/Groups/{team_id}", + update_team_response = teams.remove( + team_id, request_payload={"value": member_id} ) print(update_team_response) @@ -102,14 +96,14 @@ def update_team_remove_member(base_url, teams, team_id, member_id): if __name__ == "__main__": username = "your_username" api_key = "your_api_key" - base_url = "https://localhost/api" # Base URL of the API + base_url = "https://example.com/api" # Base URL of the API # Instantiate the Teams class with your credentials - teams = Teams(username, api_key) + teams = Teams(base_url, username, api_key) # Test Functions - get_all_teams(base_url, teams) - # create_team(base_url, teams, "test-team", "member_id") - # get_team(base_url, teams, "team_id") - # update_team_add_member(base_url, teams, "team_id", "member_id") - # update_team_remove_member(base_url, teams, "team_id", "member_id") \ No newline at end of file + get_all_teams(teams) + # create_team(teams, "test-team", "member_id") + # get_team(teams, "team_id") + # update_team_add_member(teams, "team_id", "member_id") + # update_team_remove_member(teams, "team_id", "member_id") \ No newline at end of file diff --git a/wandb-scim/examples/user_example.py b/wandb-scim/examples/user_example.py index 001cc207..0f2bd950 100644 --- a/wandb-scim/examples/user_example.py +++ b/wandb-scim/examples/user_example.py @@ -5,87 +5,81 @@ sys.path.append('../') from users import User # Assuming the User class is defined in user_module.py -def create_user(base_url, user, email, name): +def create_user(user, email, name): """ Creates a new user. Args: - base_url (str): The base URL of the API. user (User): An instance of the User class. email (str): The email address of the new user. name (str): The name of the new user. """ try: # Create a new user - create_user_response = user._create_user( - url=f"{base_url}/Users", + create_user_response = user.create( request_payload={"email": email, "name": name} ) print(create_user_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def get_user(base_url, user, user_id): +def get_user(user, user_id): """ Retrieves details of a specific user. Args: - base_url (str): The base URL of the API. user (User): An instance of the User class. user_id (str): The ID of the user to retrieve. """ try: # Get user details - get_user_response = user._get_user(f"{base_url}/Users/{user_id}") + get_user_response = user.get(user_id) print(get_user_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def get_all_users(base_url, user): +def get_all_users(user): """ Retrieves details of all users in the organization. Args: - base_url (str): The base URL of the API. user (User): An instance of the User class. """ try: # Get all users in the organization - get_all_users_response = user._get_all_user(f"{base_url}/Users") + get_all_users_response = user.get_all() print(get_all_users_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def deactivate_user(base_url, user, user_id): +def deactivate_user(user, user_id): """ Deactivates a user. Args: - base_url (str): The base URL of the API. user (User): An instance of the User class. user_id (str): The ID of the user to deactivate. """ try: # Deactivate a user - deactivate_user_response = user._deactivate_user(f"{base_url}/Users/{user_id}") + deactivate_user_response = user.deactivate(user_id) print(deactivate_user_response) except requests.exceptions.RequestException as e: print(f"Error occurred during API request: {str(e)}") -def assign_role_user(base_url, user, user_id, role_name): +def assign_role_user(user, user_id, role_name): """ Assigns a role to a user. Args: - base_url (str): The base URL of the API. user (User): An instance of the User class. user_id (str): The ID of the user to assign the role to. role_name (str): The name of the role to assign. """ try: # Assign a role to the user - assign_role_response = user._assign_role_user( - url=f"{base_url}/Users/{user_id}", + assign_role_response = user.assign_org_role( + user_id, request_payload={"roleName": role_name} ) print(assign_role_response) @@ -93,12 +87,11 @@ def assign_role_user(base_url, user, user_id, role_name): print(f"Error occurred during API request: {str(e)}") -def assign_team_user(base_url, user, user_id, team_name, role_name): +def assign_team_user(user, user_id, team_name, role_name): """ Assigns a team role to a user. Args: - base_url (str): The base URL of the API. user (User): An instance of the User class. user_id (str): The ID of the user to assign the team role to. team_name (str): The name of the team to assign the role from. @@ -106,8 +99,8 @@ def assign_team_user(base_url, user, user_id, team_name, role_name): """ try: # Assign team role to a user - assign_team_role_response = user._assign_role_team( - url=f"{base_url}/Users/{user_id}", + assign_team_role_response = user.assign_team_role( + user_id, request_payload={"roleName": role_name, "teamName": team_name} ) print(assign_team_role_response) @@ -118,15 +111,15 @@ def assign_team_user(base_url, user, user_id, team_name, role_name): if __name__ == "__main__": username = "your_username" api_key = "your_api_key" - base_url = "https://localhost/scim" + base_url = "https://example.com/api" # Instantiate the User class with your credentials - user = User(username, api_key) + user = User(base_url, username, api_key) # Test Functions - get_all_users(base_url, user) - # create_user(base_url, user, "test@example.com", "Test User") - # get_user(base_url, user, "user_id") - # deactivate_user(base_url, user, "user_id") - # assign_role_user(base_url, user, "user_id", "role_name") - # assign_team_user(base_url, user, "user_id", "team_name", "role_name") \ No newline at end of file + get_all_users(user) + # create_user(user, "test@example.com", "Test User") + # get_user(user, "user_id") + # deactivate_user(user, "user_id") + # assign_role_user(user, "user_id", "role_name") + # assign_team_user(user, "user_id", "team_name", "role_name") \ No newline at end of file diff --git a/wandb-scim/teams.py b/wandb-scim/teams.py index 0f2eae46..28722823 100644 --- a/wandb-scim/teams.py +++ b/wandb-scim/teams.py @@ -1,12 +1,14 @@ -import requests import base64 +import json +import requests class Teams(object): - def __init__(self, username, api_key): + def __init__(self, base_url, username, api_key): """ Initializes the Teams object with username and API key. Args: + base_url (str): Host url. username (str): The username for authentication. api_key (str): The API key for authentication. """ @@ -18,12 +20,11 @@ def __init__(self, username, api_key): # Create the authorization header for API requests self.authorization_header = f"Basic {self.auth_token}" - def _create_team(self, url, request_payload): + def create(self, request_payload): """ Creates a new team. Args: - url (str): The URL for the team creation endpoint. request_payload (dict): The payload containing team data. It should contain the following keys: - 'displayName': The display name of the team. @@ -47,18 +48,19 @@ def _create_team(self, url, request_payload): "Content-Type": "application/json" } # Send a POST request to create the team + url = f"{self.base_url}/scim/Groups" response = requests.post(url, json=data, headers=headers) if response.status_code == 201: return "Team has been created!" return f"Team creation failed. Status code: {response.status_code}" - def _get_team(self, url): + def get(self, team_id): """ Retrieves team details. Args: - url (str): The URL for the team retrieval endpoint. + team_id (str): team_id of the team_id. Returns: str: A message containing team details or indicating failure. @@ -69,19 +71,17 @@ def _get_team(self, url): "Content-Type": "application/json" } # Send a GET request to retrieve the team + url = f"{self.base_url}/scim/Groups/{team_id}" response = requests.get(url, headers=headers) if response.status_code == 200: - return f"Team details: {response.text}" + return json.dumps(response.json, indent=4) return f"Get team failed. Status code: {response.status_code}" - def _get_all_teams(self, url): + def get_all(self): """ Retrieves details of all teams in the organization. - Args: - url (str): The URL for the endpoint to get all teams. - Returns: str: A message containing details of all teams or indicating failure. """ @@ -91,18 +91,19 @@ def _get_all_teams(self, url): "Content-Type": "application/json" } # Send a GET request to retrieve all teams + url = f"{self.base_url}/scim/Groups" response = requests.get(url, headers=headers) if response.status_code == 200: - return f"Teams details: {response.text}" + return json.dumps(response.json, indent=4) return f"Get teams failed. Status code: {response.status_code}" - def _add_team(self, url, request_payload): + def update(self, team_id, request_payload): """ Adds a member to the team. Args: - url (str): The URL for the endpoint to add a member to the team. + team_id (str): team_id of the team_id. request_payload (dict): The payload containing member information. It should contain the following key: - 'value': The value of the member to be added to the team. @@ -130,6 +131,7 @@ def _add_team(self, url, request_payload): "Content-Type": "application/json" } # Send a PATCH request to add the member to the team + url = f"{self.base_url}/scim/Groups/{team_id}" response = requests.patch(url, json=data, headers=headers) if response.status_code == 200: @@ -142,12 +144,12 @@ def _add_team(self, url, request_payload): else: return f"Failed to update team. Status code: {response.status_code}" - def _remove_team(self, url, request_payload): + def remove(self, team_id, request_payload): """ Removes a member from the team. Args: - url (str): The URL for the endpoint to remove a member from the team. + team_id (str): team_id of the team_id. request_payload (dict): The payload containing member information. It should contain the following key: - 'value': The value of the member to be removed from the team. @@ -175,6 +177,7 @@ def _remove_team(self, url, request_payload): "Content-Type": "application/json" } # Send a PATCH request to remove the member from the team + url = f"{self.base_url}/scim/Groups/{team_id}" response = requests.patch(url, json=data, headers=headers) if response.status_code == 200: diff --git a/wandb-scim/users.py b/wandb-scim/users.py index 698df6e8..ace161a9 100644 --- a/wandb-scim/users.py +++ b/wandb-scim/users.py @@ -1,12 +1,14 @@ -import requests import base64 +import json +import requests class User(object): - def __init__(self, username, api_key): + def __init__(self, base_url, username, api_key): """ Initialize User object with username and API key. Args: + base_url (str): Host url. username (str): The username for authentication. api_key (str): The API key for authentication. """ @@ -18,12 +20,11 @@ def __init__(self, username, api_key): # Create the authorization header for API requests self.authorization_header = f"Basic {self.auth_token}" - def _create_user(self, url, request_payload): + def create(self, request_payload): """ Creates a new user. Args: - url (str): The URL for the user creation endpoint. request_payload (dict): The payload containing user data. Returns: @@ -47,17 +48,18 @@ def _create_user(self, url, request_payload): "Content-Type": "application/json" } # Send a POST request to create the user + url = f"{self.base_url}/scim/Users" response = requests.post(url, json=data, headers=headers) if response.status_code == 201: return "User has been created!" return f"User creation failed. Status code: {response.status_code}" - def _get_user(self, url): + def get(self, user_id): """ Retrieves user details. Args: - url (str): The URL for the user retrieval endpoint. + user_id (str): user_id of the user. Returns: str: A message containing user details or indicating failure. @@ -68,19 +70,17 @@ def _get_user(self, url): "Content-Type": "application/json" } # Send a GET request to retrieve the user + url = f"{self.base_url}/scim/Users/{user_id}" response = requests.get(url, headers=headers) if response.status_code == 200: - return f"user details: {response.text}" + return json.dumps(response.json, indent=4) return f"Get user failed. Status code: {response.status_code}" - def _get_all_user(self, url): + def get_all(self): """ Retrieves details of all users in the organization. - Args: - url (str): The URL for the endpoint to get all users. - Returns: str: A message containing details of all users or indicating failure. """ @@ -89,19 +89,20 @@ def _get_all_user(self, url): "Authorization": self.authorization_header, "Content-Type": "application/json" } + url = f"{self.base_url}/scim/Users" # Send a GET request to retrieve all users response = requests.get(url, headers=headers) if response.status_code == 200: - return f"users details: {response.text}" + return json.dumps(response.json, indent=4) return f"Get users failed. Status code: {response.status_code}" - def _deactivate_user(self, url): + def deactivate(self, user_id): """ Deactivates a user. Args: - url (str): The URL for the user deactivation endpoint. + user_id (str): user_id of the user. Returns: str: A message indicating whether the user deactivation was successful or failed. @@ -112,6 +113,7 @@ def _deactivate_user(self, url): "Content-Type": "application/json" } # Send a DELETE request to deactivate the user + url = f"{self.base_url}/scim/Users/{user_id}" response = requests.delete(url, headers=headers) if response.status_code == 204: @@ -121,12 +123,12 @@ def _deactivate_user(self, url): else: return f"Failed to delete user. Status code: {response.status_code}" - def _assign_role_user(self, url, request_payload): + def assign_org_role(self, user_id, request_payload): """ Assigns a role to a user. Args: - url (str): The URL for the endpoint to assign a role to the user. + user_id (str): user_id of the user. request_payload (dict): The payload containing role information. It should contain the following key: - 'roleName': The role to be assigned to the user. It can be one of 'admin', 'viewer', or 'member'. @@ -150,6 +152,7 @@ def _assign_role_user(self, url, request_payload): "Content-Type": "application/json" } # Send a PATCH request to assign the role to the user + url = f"{self.base_url}/scim/Users/{user_id}" response = requests.patch(url, json=data, headers=headers) if response.status_code == 200: updated_data = response.json() # Get the updated resource data from the response @@ -160,12 +163,12 @@ def _assign_role_user(self, url, request_payload): else: return f"Failed to update user. Status code: {response.status_code}" - def _assign_role_team(self, url, request_payload): + def assign_team_role(self, user_id, request_payload): """ Assigns a role to a user of the team. Args: - url (str): The URL for the endpoint to assign a role to the user of the team. + user_id (str): user_id of the user. request_payload (dict): The payload containing role information. Returns: @@ -192,6 +195,7 @@ def _assign_role_team(self, url, request_payload): "Content-Type": "application/json" } # Send a PATCH request to assign the role to the user of the team + url = f"{self.base_url}/scim/Users/{user_id}" response = requests.patch(url, json=data, headers=headers) if response.status_code == 200: