extract_utils

 1import boto3, logging, json
 2from botocore.exceptions import ClientError
 3from decimal import Decimal
 4from typing import List, Dict, Any
 5from datetime import datetime
 6
 7
 8def get_secret(
 9    secret_name: str = "project-onyx/totesys-db-login", region_name: str = "eu-west-2"
10) -> Dict[str, Any]:
11    """
12    Retrieves a secret from AWS Secrets Manager and parses it as a dictionary.
13
14    :param secret_name: The name of the secret to retrieve.
15    :param region_name: The AWS region where the secret is stored.
16    :raises ClientError: If there is an error retrieving the secret.
17    :return: The secret as a dictionary.
18    """
19    # Create a Secrets Manager client
20    log_message(__name__, 10, "Entered get_secret")
21    try:
22        client = boto3.client(service_name="secretsmanager", region_name=region_name)
23        get_secret_value_response = client.get_secret_value(SecretId=secret_name)
24        secret = get_secret_value_response["SecretString"]
25        result_dict = json.loads(secret)
26        return result_dict
27
28    except ClientError as e:
29        # For a list of exceptions thrown, see
30        # https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html
31        log_message(__name__, "40", e.response["Error"]["Message"])
32        raise e
33
34
35def format_response(
36    columns: List[str], response: List[List[Any]]
37) -> List[Dict[str, Any]]:
38    """
39    Formats a response into a list of dictionaries with columns as keys.
40
41    Validates that each row in the response has the same number of elements as the columns.
42
43    :param columns: A list of column names.
44    :param response: A list of rows, where each row is a list of values.
45    :return: A list of dictionaries where each dictionary represents a row.
46    :raises ValueError: If any row in the response has a different number of elements than the columns.
47    """
48    log_message(__name__, 10, "Entered format_response")
49
50    formatted_response = []
51    num_columns = len(columns)
52
53    for row in response:
54        if len(row) != num_columns:
55            raise ValueError("Mismatch between number of columns and row length")
56
57        extracted_from_response = {}
58
59        for column, value in zip(columns, row):
60            if isinstance(value, datetime):
61                value = value.strftime("%Y-%m-%d %H:%M:%S")
62            elif isinstance(value, Decimal):
63                value = float(value)
64
65            extracted_from_response[column] = value
66
67        formatted_response.append(extracted_from_response)
68
69    return formatted_response
70
71
72def log_message(name: str, level: int, message: str = ""):
73    """
74    Sends a message to the logger at a specified level.
75
76    :param name: The name of the logger.
77    :param level: The logging level (one of 10, 20, 30, 40, 50).
78    :param message: The message to log.
79    """
80    logger = logging.getLogger(name)
81
82    # Define a mapping of level integers to logging methods
83    level_map = {
84        10: logger.debug,
85        20: logger.info,
86        30: logger.warning,
87        40: logger.error,
88        50: logger.critical,
89    }
90
91    # Get the logging method from the map
92    log_method = level_map.get(level)
93
94    if log_method:
95        log_method(message)
96    else:
97        logger.error("Invalid log level: %d", level)