'''
Created Date: Monday June 17th 2024 +1000
Author: Peter Baker
-----
Last Modified: Monday June 17th 2024 5:00:47 pm +1000
Modified By: Peter Baker
-----
Description: General helper functions which are useful across the client library.
-----
HISTORY:
Date By Comments
---------- --- ---------------------------------------------------------
'''
from pydantic import BaseModel, ValidationError
from typing import Any, Dict, List, Mapping, Optional, Tuple, TypeVar, Type, Union, ByteString
import json
from httpx import Response
from httpx._types import RequestFiles
from provenaclient.utils.exceptions import AuthException, HTTPValidationException, ServerException, BadRequestException, ValidationException, NotFoundException
from provenaclient.utils.exceptions import BaseException
from ProvenaInterfaces.SharedTypes import StatusResponse
from ProvenaInterfaces.RegistryModels import ItemBase, ItemSubType
import os
# Type var to refer to base models
BaseModelType = TypeVar("BaseModelType", bound=BaseModel)
ItemModelType = TypeVar("ItemModelType", bound=ItemBase)
# Type alias for json data
JsonData = Union[List[Dict[str, Any]],Dict[str, Any]]
# Type alias for httpx file upload.
HttpxFileUpload = RequestFiles
ParamTypes = Union[str, int, bool]
[docs]
def convert_to_item_subtype(item_subtype_str: Optional[str]) -> ItemSubType:
"""Converts a string into ItemSubType supported enum type.
Parameters
----------
item_subtype_str : Optional[str]
Optional string containing similar enum text.
Returns
-------
ItemSubType
Enum type of ItemSubType.
Raises
------
ValueError
Item subtype field was not present.
ValueError
Item subtype cannot be converted to ENUM.
"""
if item_subtype_str is None:
raise ValueError("Item subtype field not found!")
try:
return ItemSubType[item_subtype_str.upper()]
except KeyError as e:
raise ValueError(f"Invalid item_subtype: {item_subtype_str}") from e
[docs]
def get_and_validate_file_path(file_path: Optional[str], write_to_file: bool, default_file_name: str) -> Optional[str]:
"""Determine and validate the file path for writing a file.
If file_path is not provided and write_to_file is True then will use a dynamic
default file name.
Parameters
----------
file_path : Optional[str]
The path to save the file at.
write_to_file : bool
A boolean flag indicating whether writing to the file is enabled.
default_file_name : str
The default file name to use if file_path is not provided.
Returns
-------
Optional[str]
The validated file path, or None if writing to file is not enabled.
Raises
------
ValueError
If a file path is provided but writing to the file is not enabled.
"""
if file_path and not write_to_file:
raise ValueError(f"Write to CSV must be enabled. Currently {write_to_file}")
if file_path and write_to_file:
# Validate provided file path directory.
validate_existing_path(file_path)
if not file_path and write_to_file:
# Create default file path directory.
file_path = default_file_name
return file_path
[docs]
def validate_existing_path(file_path: str) -> None :
"""Validates a provided file path, and checks if
the directory exists.
Parameters
----------
file_path : str
The file path to validate.
Raises
------
ValueError
If the directory part of the path does not exist.
IOError
If an I/O error occurs during file operations.
Exception
For any other exceptions that may occur.
"""
try:
# Validates if the provided directory part of the file path exists.
directory = os.path.dirname(file_path)
if not os.path.isdir(directory):
raise ValueError(f"The provided path {file_path} is incorrect. Please try again.")
except IOError as e:
raise IOError(f"Failed to validate {file_path} due to I/O error: {e}")
except Exception as e:
raise Exception(f"Path validation failed. Exception {e}")
[docs]
def write_file_helper(file_path: str, content: Union[str, bytes]) -> None:
"""
Writes the provided content (string or bytes) to a file at the specified file path.
Parameters
----------
file_name : str
The name of the file to write content into.
content : Union[str, bytes]
The content to be written to the file. It can be either:
- A `str`, which will be written in text mode.
- A `bytes` object, which will be written in binary mode.
Raises
------
IOError
If an I/O error occurs during file operations.
Exception
For non-I/O related exceptions that may occur during file writing.
"""
try:
if not os.path.exists(path=file_path):
# Make the path, before writing to a file.
os.makedirs(os.path.dirname(file_path), exist_ok=True)
if isinstance(content, str):
# Write to file
with open(file_path, 'w') as file:
file.write(content)
if isinstance(content, bytes):
# Write to file in byte mode
with open(file_path, 'wb') as file:
file.write(content)
except IOError as e:
raise IOError(f"Failed to file {file_path} due to I/O error: {e}")
except Exception as e:
raise Exception(f"File writing failed. Exception {e}")
[docs]
def read_file_helper(file_path: str) -> str:
"""Reads a valid file and returns its content
Parameters
----------
file_path : str
The path of an existing created file.
Returns
-------
str
A string representation of the file contents.
Raises
------
Exception
If there any error with reading the file
this general exception is raised.
"""
try:
file = open(file_path, 'r') # Open the file in read mode
file_content = file.read() # Save the contents of the file.
file.close()
return file_content
except Exception as e:
raise Exception(f"Error with file. Exception {e}")
[docs]
def build_params_exclude_none(params: Mapping[str, Optional[ParamTypes]]) -> Dict[str, ParamTypes]:
"""
Takes a raw params dict with optional args and returns filtered.
Args:
params (Dict[str, Optional[ParamTypes]]): The input raw dict
Returns:
Dict[str, ParamTypes]: The filtered param list with no None values
"""
return {id: val for id, val in params.items() if val is not None}
[docs]
def py_to_dict(model: BaseModel) -> JsonData:
""" This helper function converts a Pydantic model to a Python dictionary.
Requires a pydantic dump into serialised JSON to be safe against all object
types
Parameters
----------
model : BaseModel
The instance of the model that needs to be converted to a dict.
Returns
-------
JsonData
A python dictionary object which contains the fields and values of the
base model that are not none.
"""
return json.loads(model.json(exclude_none=True))
[docs]
def handle_model_parsing(json_data: JsonData, model: Type[BaseModelType]) -> BaseModelType:
"""This generic helper function parses a HTTP Response into a
python datatype based on a pydantic defined model.
Parameters
----------
json_data : JsonData
The response received after HTTP request.
model : type[T]
The type of the model being casted from HTTP response
into python datatype. For example:
MintResponse, RegistryFetchResponse.
Returns
-------
T
Returns a python datatype that conforms to the structure
of the provided model.
"""
try:
parsed_model = model.parse_obj(json_data)
return parsed_model
except ValidationError as e:
raise e
[docs]
def parse_json_payload(response: Response) -> JsonData:
"""
Parses a HTTPX response object into JSON handling the error if any occurs.
Args:
response (Response): The raw HTTP response.
Raises:
ValidationException: If JSON decoding fails, handles and raises error.
Returns:
JsonData: The returned dictionary object representing JSON data.
"""
try:
parsed_response = response.json()
except json.JSONDecodeError as e:
raise ValidationException("JSON parsing failed") from e
return parsed_response
[docs]
def handle_err_codes(response: Response, error_message: Optional[str]) -> None:
"""
This helper function checks the status code of the HTTP response and raises
a custom exception accordingly.
Also embeds error info from JSON or text result.
Parameters
----------
response : Response
The httpx.response object.
Raises
------
BadRequestException
Raised when the server returns a 400 status code.
AuthException
Raised when the server returns a 401 status code.
ValidationException
Raised when the server returns a 422 status code.
ServerException
Raised when the server returns a status code of 500 or above.
"""
text: Union[str, None] = None
try:
data = response.json()
text = json.dumps(data, indent=2)
except Exception:
text = response.text
if response.status_code == 400:
raise BadRequestException(
message=f"Bad Request. Details: {text}.", error_code=400, payload=error_message)
if response.status_code == 401:
raise AuthException(message=f"Authentication failed. Details: {text}.",
error_code=401, payload=error_message)
if response.status_code == 404:
raise NotFoundException(message=f"Url was not found at provided service endpoint. Details: {text}.",
error_code=404, payload=error_message)
if response.status_code == 422:
# This is a specific status code of this URL.
raise HTTPValidationException(
message=f"Validation error. Details: {text}.", error_code=422, payload=error_message)
if response.status_code != 200:
# Raise another exception here
raise ServerException(message=f"Server error occurred. Details: {text}.",
error_code=response.status_code, payload=error_message)
[docs]
def check_status_response(json_data: JsonData) -> None:
"""
Parses JSON data as StatusResponse model, then asserts success is true,
throwing exception with embedded details if not.
Args:
json_data (Dict): The JSON data to parse
Raises:
Exception: Exception if status if False
"""
# Check model parses
status_obj = handle_model_parsing(
json_data=json_data, model=StatusResponse)
# Check status is success
if not status_obj.status.success:
raise Exception(
f"Status object from API indicated failure. Details: {status_obj.status.details}.")
[docs]
def check_codes_and_parse_json(response: Response, error_message: Optional[str]) -> JsonData:
"""
Given raw response, validates codes and parses as JSON.
Args:
response (Response): The raw http client response
error_message (Optional[str]): The error message to embed if any
Returns:
JsonData: The json data post parse
"""
# Check codes
handle_err_codes(response=response, error_message=error_message)
# Handle JSON parsing
json_data = parse_json_payload(response=response)
return json_data
[docs]
def handle_response_non_status(response: Response, model: Type[BaseModelType], error_message: Optional[str]) -> BaseModelType:
"""
Given the raw response from http client, and the model, will validate
- 200 OK code (with common errors handled)
- Parsed as JSON
- Parsed as desired final model
Returns the parsed pydantic object.
Args:
response (Response): The raw response from http client
model (Type[T]): The model type (not instance) to parse against
error_message (Optional[str]): The error message to embed into exceptions.
Returns:
T: The parsed model
"""
# Handle JSON parsing and codes
json_data = check_codes_and_parse_json(
response=response, error_message=error_message)
# Check model parses
parsed_obj = handle_model_parsing(json_data=json_data, model=model)
return parsed_obj
[docs]
def handle_response_with_status(response: Response, model: Type[BaseModelType], error_message: Optional[str]) -> BaseModelType:
"""
Given the raw response from http client, and the model, will validate
- 200 OK code (with common errors handled)
- Parsed as JSON
- Parsed as StatusResponse and asserted for success=true, throwing with details if not
- Parsed as desired final model
Returns the parsed pydantic object.
Args:
response (Response): The raw response from http client
model (Type[T]): The model type (not instance) to parse against
error_message (Optional[str]): The error message to embed into exceptions.
Returns:
T: The parsed model
"""
# Handle JSON parsing and codes
json_data = check_codes_and_parse_json(
response=response, error_message=error_message)
# Check status result
check_status_response(json_data=json_data)
# Check model parses
parsed_obj = handle_model_parsing(json_data=json_data, model=model)
return parsed_obj