bedrock.endpoints.decorators.verify_body_key_value_pairs

 1import functools  # pragma: unit
 2import json  # pragma: unit
 3
 4from bedrock.exceptions import BadRequestException, BedrockException  # pragma: unit
 5
 6
 7def verify_body_key_value_pairs(expected: dict, fail_on_missing: bool = True):  # pragma: unit
 8    """
 9    Verify that the body of the request contains the expected key-value pairs.
10
11    :param expected: A dictionary of expected key-value pairs. If the value is a list, the verification passes when the body value is contained in the list. Values can be `{{placeholders}}` for arguments in the function's kwargs.
12    :param fail_on_missing: Whether to raise an exception if the body is missing a key. Defaults to `True`.
13
14    Example usage: for a `POST` to `/countries/` where the body must include a key `country_code` with the values of
15    either `"GB"` or `"PT"`:
16    ```python
17    class Countries(Endpoint):
18        # ...
19        @verify_body_key_value_pairs({"country_code": ["GB", "PT"]})
20        def post_global(self, event):
21            # ...
22    ```
23
24    Example usage: for a `POST` to `/countries/` where the body must include the same value as a protected entity's field:
25    ```python
26    class Countries(Endpoint):
27        # ...
28        @protected("country_code")
29        @verify_body_key_value_pairs({"country_code": "{{country_code}}"})
30        def post_global(self, event, country_code):
31            # ...
32    ```
33    """
34
35    def decorator(func):
36        @functools.wraps(func)
37        def wrapper(*args, **kwargs):
38            event = args[1]
39            body = json.loads(event["body"])
40            for key, _value in expected.items():
41                if key not in body:
42                    if not fail_on_missing:
43                        continue
44                    else:
45                        raise BadRequestException(f"Missing key {key}")
46                if isinstance(_value, str) and _value.startswith("{{"):
47                    value_placeholder_name = _value.replace("{{", "").replace("}}", "")
48                    try:
49                        value = kwargs[value_placeholder_name]
50                    except KeyError:
51                        raise BedrockException(f"Missing value for placeholder {value_placeholder_name}")
52                else:
53                    value = _value
54                if not _matches(value, body[key]):
55                    raise BadRequestException(f"Invalid value for key - {key}: {value}")
56
57            return func(*args, **kwargs)
58
59        return wrapper
60
61    return decorator
62
63
64def _matches(matches, value_to_match):  # pragma: unit
65    if isinstance(matches, list):
66        return value_to_match in matches
67    return value_to_match == matches