bedrock.endpoints.decorators.query_params.common

 1from urllib.parse import quote  # pragma: unit
 2from bedrock.exceptions import BadRequestException  # pragma: unit
 3
 4
 5def extract_query_parameter(event, param_name, param_type, param_kwarg,
 6                            allow_multi_value: bool = False):  # pragma: unit
 7    kwarg = param_kwarg if param_kwarg is not None else param_name
 8    if allow_multi_value:
 9        return _extract_multi_value_query_parameter(event, param_name, param_type, kwarg)
10    else:
11        return _extract_single_query_parameter(event, param_name, param_type, kwarg)
12
13
14def _extract_single_query_parameter(event, param_name, param_type, kwarg):  # pragma: unit
15    arguments = {}
16    if "queryStringParameters" in event and event["queryStringParameters"] is not None:
17        if param_name in event["queryStringParameters"]:
18            arguments[kwarg] = _argument_converter(event["queryStringParameters"][param_name], param_type)
19        elif quote(param_name) in event["queryStringParameters"]:
20            arguments[kwarg] = _argument_converter(event["queryStringParameters"][quote(param_name)], param_type)
21    return arguments
22
23
24def _extract_multi_value_query_parameter(event, param_name, param_type, kwarg):  # pragma: unit
25    arguments = {}
26    if "multiValueQueryStringParameters" in event and event["multiValueQueryStringParameters"] is not None:
27        if param_name in event["multiValueQueryStringParameters"]:
28            arguments[kwarg] = _construct_multi_value_query_list(event, param_name, param_type)
29        elif quote(param_name) in event["multiValueQueryStringParameters"]:
30            arguments[kwarg] = _construct_multi_value_query_list(event, quote(param_name), param_type)
31    return arguments
32
33
34def _construct_multi_value_query_list(event, param_name, param_type):  # pragma: unit
35    if param_type == str:
36        return [_argument_converter(value, param_type)
37                for param in event["multiValueQueryStringParameters"][param_name]
38                for value in param.split(",")]
39    else:
40        return [_argument_converter(value, param_type)
41                for value in event["multiValueQueryStringParameters"][param_name]]
42
43
44def _argument_converter(value, value_type):  # pragma: unit
45    if value == "null":
46        return None
47    try:
48        if value_type == str:
49            return f"{value}"
50        if value_type == int:
51            return int(f"{value}")
52        if value_type == float:
53            return float(f"{value}")
54        if value_type == bool:
55            return f"{value}".lower() == "true"
56    except ValueError:
57        raise BadRequestException(f"Unsupported type: {value_type}")
58    raise BadRequestException(f"Unsupported type: {value_type}")