From 43235213b54a1736c95eba2f0aa05261ed429b8f Mon Sep 17 00:00:00 2001 From: Joe Rourke Date: Tue, 15 Apr 2025 23:31:07 +0100 Subject: [PATCH 1/2] refactor: use private helper function for api calls --- sdk/rapid/rapid.py | 443 +++++++++++++++++++++------------------------ 1 file changed, 208 insertions(+), 235 deletions(-) diff --git a/sdk/rapid/rapid.py b/sdk/rapid/rapid.py index 4fb8ddce..38eef71e 100644 --- a/sdk/rapid/rapid.py +++ b/sdk/rapid/rapid.py @@ -50,6 +50,58 @@ def generate_headers(self, is_file: bool = False) -> Dict: "Authorization": f"Bearer {self.auth.fetch_token()}", **({} if is_file else {"Content-Type": "application/json"}) } + + def __perform_api_request( + self, + endpoint: str, + method: str, + data=None, + files=None, + headers={}, + timeout=TIMEOUT_PERIOD, + success_code=200, + error_handlers={} + ): + """ + A helper method to perform API requests. This method is not intended to be used directly. + + Args: + endpoint (str): The API endpoint to call. + method (str): The HTTP method to use (GET, POST, PUT, DELETE). + data (dict, optional): The data to send with the request. Defaults to an empty dictionary. + files (dict, optional): The files to send with the request. Defaults to None. + headers (dict, optional): The headers to send with the request. Defaults to an empty dictionary. + timeout (int, optional): The timeout period for the request. Defaults to TIMEOUT_PERIOD. + success_code (int, optional): The expected success status code. Defaults to 200. + error_handlers (dict, optional): A dictionary mapping status codes to exception classes. Defaults to a set of common exceptions. + + Returns: + dict: The JSON response from the API. + """ + + if not headers: + headers = self.generate_headers() + + url = f"{self.auth.url}/{endpoint}" + response = requests.request( + method, + url, + headers=headers, + params= + data=json.dumps(data), + timeout=timeout, + ) + data = json.loads(response.content.decode("utf-8")) + if response.status_code == success_code: + return data + elif response.status_code in error_handlers: + raise error_handlers[response.status_code](data) + elif "default" in error_handlers: + raise error_handlers["default"](data) + else: + raise Exception( + f"Unexpected error occurred: {response.status_code}", data + ) def list_datasets(self): """ @@ -58,12 +110,11 @@ def list_datasets(self): Returns: A JSON response of the API's response. """ - response = requests.post( - f"{self.auth.url}/datasets", - headers=self.generate_headers(), - timeout=TIMEOUT_PERIOD, + return self.__perform_api_request( + endpoint="datasets", + method="POST", ) - return json.loads(response.content.decode("utf-8")) + def fetch_job_progress(self, _id: str): """ @@ -75,14 +126,15 @@ def fetch_job_progress(self, _id: str): Returns: A JSON response of the API's response. """ - url = f"{self.auth.url}/jobs/{_id}" - response = requests.get( - url, headers=self.generate_headers(), timeout=TIMEOUT_PERIOD + + return self.__perform_api_request( + endpoint=f"jobs/{_id}", + method="GET", + error_handlers={ + "default": lambda d: UnableToFetchJobStatusException("Could not check job status", d), + }, ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return data - raise UnableToFetchJobStatusException("Could not check job status", data) + def wait_for_job_outcome(self, _id: str, interval: int = 1): """ @@ -133,23 +185,26 @@ def download_dataframe( Returns: DataFrame: A pandas DataFrame of the data """ - url = f"{self.auth.url}/datasets/{layer}/{domain}/{dataset}/query" - if version is not None: - url = f"{url}?version={version}" - response = requests.post( - url, + endpoint = f"datasets/{layer}/{domain}/{dataset}/download" + if version: + endpoint += f"?version={version}" + + data = self.__perform_api_request( + endpoint, + method="POST", + data=query.dict(exclude_none=True), headers=self.generate_headers(), - data=json.dumps(query.dict(exclude_none=True)), - timeout=TIMEOUT_PERIOD, - ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return pd.read_json(json.dumps(data), orient="index") - - raise DatasetNotFoundException( - f"Could not find dataset, {layer}/{domain}/{dataset} to download", data + success_code=200, + error_handlers={ + "default": lambda d: DatasetNotFoundException( + f"Could not find dataset, {layer}/{domain}/{dataset} to download", d + ), + }, ) - + + return pd.read_json(json.dumps(data), orient="index") + + def upload_dataframe( self, layer: str, @@ -176,34 +231,26 @@ def upload_dataframe( Returns: If wait_to_complete is True, returns "Success" if the upload is successful. If wait_to_complete is False, returns the ID of the upload job if the upload is accepted. - """ - url = f"{self.auth.url}/datasets/{layer}/{domain}/{dataset}" - response = requests.post( - url, - headers=self.generate_headers(is_file=True), + """ + data = self.__perform_api_request( + endpoint=f"datasets/{layer}/{domain}/{dataset}", + method="POST", files=self.convert_dataframe_for_file_upload(df), - timeout=TIMEOUT_PERIOD, + headers=self.generate_headers(is_file=True), + success_code=202, + error_handlers={ + 422: lambda d: DataFrameUploadValidationException("Could not upload dataframe due to an incorrect schema definition", d), + 404: lambda d: DatasetNotFoundException("Could not find dataset", d), + "default": lambda d: DataFrameUploadFailedException("Encountered an unexpected error, could not upload dataframe", d), + }, ) - data = json.loads(response.content.decode("utf-8")) + + if wait_to_complete: + self.wait_for_job_outcome(data["details"]["job_id"]) + return "Success" + + return data["details"]["job_id"] - if response.status_code == 202: - if wait_to_complete: - self.wait_for_job_outcome(data["details"]["job_id"]) - return "Success" - return data["details"]["job_id"] - if response.status_code == 422: - raise DataFrameUploadValidationException( - "Could not upload dataframe due to an incorrect schema definition" - ) - elif response.status_code == 404: - raise DatasetNotFoundException( - "Could not find dataset: {layer}/{domain}/{dataset}", data - ) - else: - raise DataFrameUploadFailedException( - "Encountered an unexpected error, could not upload dataframe", - data["details"], - ) def fetch_dataset_info(self, layer: str, domain: str, dataset: str): """ @@ -221,24 +268,15 @@ def fetch_dataset_info(self, layer: str, domain: str, dataset: str): Returns: A dictionary containing the metadata information for the dataset. """ - url = f"{self.auth.url}/datasets/{layer}/{domain}/{dataset}/info" - response = requests.get( - url, - headers=self.generate_headers(), - timeout=TIMEOUT_PERIOD, + return self.__perform_api_request( + "datasets/{layer}/{domain}/{dataset}/info", + method="GET", + error_handlers={ + 404: lambda d: DatasetNotFoundException(f"Could not find dataset, {layer}/{domain}/{dataset} to get info", d), + "default": lambda d: DatasetInfoFailedException("Failed to gather the dataset info", d) + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return data - - if response.status_code == 404: - raise DatasetNotFoundException( - f"Could not find dataset, {layer}/{domain}/{dataset} to get info", data - ) - raise DatasetInfoFailedException( - "Failed to gather the dataset info", data["details"] - ) def convert_dataframe_for_file_upload(self, df: DataFrame): """ @@ -276,20 +314,15 @@ def generate_schema( Returns: rapid.items.schema.Schema: A Schema class type from the generated schema for the DataFrame and dataset. """ - url = ( - f"{self.auth.url}/schema/{layer}/{sensitivity}/{domain}/{dataset}/generate" - ) - - response = requests.post( - url, - headers=self.generate_headers(is_file=True), + return Schema(**self.__perform_api_request( + endpoint=f"schema/{layer}/{sensitivity}/{domain}/{dataset}/generate", + method="POST", files=self.convert_dataframe_for_file_upload(df), - timeout=TIMEOUT_PERIOD, - ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return Schema(**data) - raise SchemaGenerationFailedException("Could not generate schema", data) + headers=self.generate_headers(is_file=True), + error_handlers={ + "default": lambda d: SchemaGenerationFailedException("Could not generate schema", d), + } + )) def create_schema(self, schema: Schema): """ @@ -302,21 +335,15 @@ def create_schema(self, schema: Schema): rapid.exceptions.SchemaAlreadyExistsException: If you try to create a schema that already exists in rAPId. rapid.exceptions.SchemaCreateFailedException: If an error occurs while trying to update the schema. """ - schema_dict = schema.dict() - url = f"{self.auth.url}/schema" - response = requests.post( - url, - headers=self.generate_headers(), - data=json.dumps(schema_dict), - timeout=TIMEOUT_PERIOD, + return self.__perform_api_request( + endpoint=f"schema", + data=json.dumps(schema.dict()), + method="POST", + error_handlers={ + 404: lambda d: SchemaGenerationFailedException("Could not find schema", d), + "default": lambda d: SchemaGenerationFailedException("Failed to gather the schema", d) + } ) - if response.status_code == 201: - pass - elif response.status_code == 409: - raise SchemaAlreadyExistsException("The schema already exists") - else: - data = json.loads(response.content.decode("utf-8")) - raise SchemaCreateFailedException("Could not create schema", data) def update_schema(self, schema: Schema): """ @@ -328,18 +355,15 @@ def update_schema(self, schema: Schema): Raises: rapid.exceptions.SchemaUpdateFailedException: If an error occurs while trying to update the schema. """ - schema_dict = schema.dict() - url = f"{self.auth.url}/schema" - response = requests.put( - url, - headers=self.generate_headers(), - data=json.dumps(schema_dict), - timeout=TIMEOUT_PERIOD, + return self.__perform_api_request( + endpoint=f"schema", + data=json.dumps(schema.dict()), + method="PUT", + error_handlers={ + "default": lambda d: SchemaUpdateFailedException("Could not update schema", d) + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return data - raise SchemaUpdateFailedException("Could not update schema", data) + def create_client(self, client_name: str, client_permissions: list[str]): """ @@ -352,26 +376,20 @@ def create_client(self, client_name: str, client_permissions: list[str]): Raises: rapid.exceptions.InvalidPermissionsException: If an error occurs while trying to create the client. """ - url = f"{self.auth.url}/client" - response = requests.post( - url, - headers=self.generate_headers(), + return self.__perform_api_request( + endpoint="client", + method="POST", + success_code=201, data=json.dumps( {"client_name": client_name, "permissions": client_permissions} ), - timeout=TIMEOUT_PERIOD, - ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 201: - return data - elif response.status_code == 400: - raise SubjectAlreadyExistsException( - f"The client {client_name} already exists" - ) - raise InvalidPermissionsException( - "One or more of the provided permissions is invalid or duplicated" + error_handlers={ + 400: lambda d: SubjectAlreadyExistsException(f"The client {client_name} already exists"), + "default": lambda d: InvalidPermissionsException("One or more of the provided permissions is invalid or duplicated") + } ) + def delete_client(self, client_id: str) -> None: """ Deletes a client from the API based on their id @@ -382,18 +400,14 @@ def delete_client(self, client_id: str) -> None: Raises: rapid.exceptions.SubjectNotFoundException: If the client does not exist. """ - url = f"{self.auth.url}/client/{client_id}" - response = requests.delete( - url, - headers=self.generate_headers(), - timeout=TIMEOUT_PERIOD, + return self.__perform_api_request( + endpoint=f"client/{client_id}", + method="DELETE", + error_handlers={ + "default": lambda d: SubjectNotFoundException(f"Failed to delete client with id: {client_id}, ensure it exists.") + } ) - if response.status_code == 200: - return None - raise SubjectNotFoundException( - f"Failed to delete client with id: {client_id}, ensure it exists." - ) def update_subject_permissions(self, subject_id: str, permissions: list[str]): """ @@ -406,20 +420,15 @@ def update_subject_permissions(self, subject_id: str, permissions: list[str]): Raises: rapid.exceptions.InvalidPermissionsException: If an error occurs while trying to create the client. """ - url = f"{self.auth.url}/subject/permissions" - response = requests.put( - url, - headers=self.generate_headers(), + return self.__perform_api_request( + endpoint=f"subject/permissions", + method="PUT", data=json.dumps({"subject_id": subject_id, "permissions": permissions}), - timeout=TIMEOUT_PERIOD, + error_handlers={ + "default": lambda d: InvalidPermissionsException("One or more of the provided permissions is invalid or duplicated") + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return data - raise InvalidPermissionsException( - "One or more of the provided permissions is invalid or duplicated" - ) def create_protected_domain(self, name: str): """ @@ -433,19 +442,17 @@ def create_protected_domain(self, name: str): rapid.exceptions.DomainConflictException: If the domain already exists. """ - url = f"{self.auth.url}/protected_domains/{name}" - response = requests.post( - url, headers=self.generate_headers(), timeout=TIMEOUT_PERIOD + return self.__perform_api_request( + endpoint=f"protected_domains/{name}", + method="POST", + success_code=201, + error_handlers={ + 400: lambda d: InvalidDomainNameException(d["details"]), + 409: lambda d: DomainConflictException(d["details"]), + "default": lambda d: Exception("Failed to create protected domain") + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 201: - return - elif response.status_code == 400: - raise InvalidDomainNameException(data["details"]) - elif response.status_code == 409: - raise DomainConflictException(data["details"]) - raise Exception("Failed to create protected domain") def create_user(self, user_name: str, user_email: str, user_permissions: list[str]): """ @@ -459,10 +466,11 @@ def create_user(self, user_name: str, user_email: str, user_permissions: list[st Raises: rapid.exceptions.InvalidPermissionsException: If an error occurs while trying to create the user. """ - url = f"{self.auth.url}/user" - response = requests.post( - url, - headers=self.generate_headers(), + + return self.__perform_api_request( + endpoint="user", + method="POST", + success_code=201, data=json.dumps( { "username": user_name, @@ -470,25 +478,13 @@ def create_user(self, user_name: str, user_email: str, user_permissions: list[st "permissions": user_permissions, } ), - timeout=TIMEOUT_PERIOD, + error_handlers={ + 400: lambda d: SubjectAlreadyExistsException(d["details"]) if d["details"] != invalid_perm_msg else InvalidPermissionsException(d["details"]), + 401: lambda d: ClientDoesNotHaveUserAdminPermissionsException(d["details"]), + "default": lambda d: Exception("Failed to create user") + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 201: - return data - elif response.status_code == 400: - if ( - data["details"] - == "One or more of the provided permissions is invalid or duplicated" - ): - raise InvalidPermissionsException( - "One or more of the provided permissions is invalid or duplicated" - ) - else: - raise SubjectAlreadyExistsException(data["details"]) - elif response.status_code == 401: - raise ClientDoesNotHaveUserAdminPermissionsException(data["details"]) - - raise Exception("Failed to create user") + def delete_user(self, user_name: str, user_id: str): """ @@ -500,24 +496,17 @@ def delete_user(self, user_name: str, user_id: str): Raises: rapid.exceptions.SubjectNotFoundException: If the client does not exist. """ - url = f"{self.auth.url}/user" - response = requests.delete( - url, - headers=self.generate_headers(), + return self.__perform_api_request( + endpoint="user", + method="DELETE", data=json.dumps({"username": user_name, "user_id": user_id}), - timeout=TIMEOUT_PERIOD, + error_handlers={ + 400: lambda d: SubjectNotFoundException(f"Failed to delete user with id: {user_id}, ensure it exists."), + 401: lambda d: ClientDoesNotHaveUserAdminPermissionsException(d["details"]), + "default": lambda d: Exception("Failed to delete user") + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return data - elif response.status_code == 400: - raise SubjectNotFoundException( - f"Failed to delete user with id: {user_id}, ensure it exists." - ) - elif response.status_code == 401: - raise ClientDoesNotHaveUserAdminPermissionsException(data["details"]) - raise Exception("Failed to delete user") def list_subjects(self): """ @@ -526,18 +515,15 @@ def list_subjects(self): Returns: A JSON response of the API's response. """ - response = requests.get( - f"{self.auth.url}/subjects", - headers=self.generate_headers(), - timeout=TIMEOUT_PERIOD, + return self.__perform_api_request( + endpoint="subjects", + method="GET", + error_handlers={ + 401: lambda d: ClientDoesNotHaveUserAdminPermissionsException(d["details"]), + "default": lambda d: Exception("Failed to list subjects") + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return data - elif response.status_code == 401: - raise ClientDoesNotHaveUserAdminPermissionsException(data["details"]) - raise Exception("Failed to list subjects") def list_layers(self): """ @@ -546,16 +532,15 @@ def list_layers(self): Returns: A JSON response of the API's response. """ - response = requests.get( - f"{self.auth.url}/layers", - headers=self.generate_headers(), - timeout=TIMEOUT_PERIOD, + return self.__perform_api_request( + endpoint="layers", + method="GET", + error_handlers={ + 401: lambda d: ClientDoesNotHaveUserAdminPermissionsException(d["details"]), + "default": lambda d: Exception("Failed to list layers") + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return data - raise Exception("Failed to list layers") def list_protected_domains(self): """ @@ -564,18 +549,14 @@ def list_protected_domains(self): Returns: A JSON response of the API's response. """ - response = requests.get( - f"{self.auth.url}/protected_domains", - headers=self.generate_headers(), - timeout=TIMEOUT_PERIOD, + return self.__perform_api_request( + endpoint="protected_domains", + method="GET", + error_handlers={ + 401: lambda d: ClientDoesNotHaveUserAdminPermissionsException(d["details"]), + "default": lambda d: Exception("Failed to list protected domains") + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 200: - return data - elif response.status_code == 401: - raise ClientDoesNotHaveUserAdminPermissionsException(data["details"]) - - raise Exception("Failed to list protected domains") def delete_dataset(self, layer: str, domain: str, dataset: str): """ @@ -589,20 +570,12 @@ def delete_dataset(self, layer: str, domain: str, dataset: str): Raises: rapid.exceptions.DatasetNotFoundException: If the dataset does not exist. """ - url = f"{self.auth.url}/datasets/{layer}/{domain}/{dataset}" - response = requests.delete( - url, - headers=self.generate_headers(), - timeout=TIMEOUT_PERIOD, + return self.__perform_api_request( + endpoint=f"datasets/{layer}/{domain}/{dataset}", + method="DELETE", + error_handlers={ + 400: lambda d: DatasetNotFoundException(f"Could not find dataset, {layer}/{domain}/{dataset} to delete", d), + 401: lambda d: ClientDoesNotHaveDataAdminPermissionsException(d["details"]), + "default": lambda d: Exception("Failed to delete dataset") + } ) - data = json.loads(response.content.decode("utf-8")) - if response.status_code == 202: - return data - elif response.status_code == 400: - raise DatasetNotFoundException( - f"Could not find dataset, {layer}/{domain}/{dataset} to delete", data - ) - elif response.status_code == 401: - raise ClientDoesNotHaveDataAdminPermissionsException(data["details"]) - - raise Exception("Failed to delete dataset") From 9b7b829de4cc167b01e0bdd5a3bdb531588c9ee9 Mon Sep 17 00:00:00 2001 From: Joe Rourke Date: Tue, 15 Apr 2025 23:40:01 +0100 Subject: [PATCH 2/2] fix: missing parameters in helper request --- sdk/rapid/rapid.py | 25 ++++++++++++------------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/sdk/rapid/rapid.py b/sdk/rapid/rapid.py index 38eef71e..7456c96f 100644 --- a/sdk/rapid/rapid.py +++ b/sdk/rapid/rapid.py @@ -3,7 +3,7 @@ import requests from datetime import datetime -from typing import Dict, Optional +from typing import Callable, Dict, Optional import pandas as pd @@ -55,12 +55,12 @@ def __perform_api_request( self, endpoint: str, method: str, - data=None, - files=None, - headers={}, - timeout=TIMEOUT_PERIOD, - success_code=200, - error_handlers={} + data: str | None = None, + files: Dict[str, tuple[str, bytes]] | None = None, + headers: Dict[str, str] = {}, + timeout: int = TIMEOUT_PERIOD, + success_code: int = 200, + error_handlers: Dict[str | int, Callable[[Dict], Exception]] = {} ): """ A helper method to perform API requests. This method is not intended to be used directly. @@ -87,7 +87,7 @@ def __perform_api_request( method, url, headers=headers, - params= + files=files, data=json.dumps(data), timeout=timeout, ) @@ -99,9 +99,7 @@ def __perform_api_request( elif "default" in error_handlers: raise error_handlers["default"](data) else: - raise Exception( - f"Unexpected error occurred: {response.status_code}", data - ) + raise Exception(f"Unexpected error occurred: {response.status_code}", data) def list_datasets(self): """ @@ -339,9 +337,10 @@ def create_schema(self, schema: Schema): endpoint=f"schema", data=json.dumps(schema.dict()), method="POST", + success_code=201, error_handlers={ - 404: lambda d: SchemaGenerationFailedException("Could not find schema", d), - "default": lambda d: SchemaGenerationFailedException("Failed to gather the schema", d) + 409: lambda d: SchemaAlreadyExistsException("The schema already exists"), + "default": lambda d: SchemaCreateFailedException("Could not create schema", d) } )