1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
import json
from typing import TYPE_CHECKING, Any, TypedDict
if TYPE_CHECKING:
from typing import Union
from typing_extensions import Any, Required
# NOTE: Typing is based on the following document https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html
EventMessageType = TypedDict(
"EventMessageType",
{
"version": str,
"id": str,
"detail-type": "Required[Union[str, list[str]]]",
"source": "Required[Union[str, list[str]]]",
"account": str,
# support float type for internal use of moto.
"time": "Union[str, float]",
"replay-name": str,
"region": str,
"resources": list[str],
"detail": "Required[dict[str, Any]]",
"ingestion-time": float,
},
total=False,
)
PAGINATION_MODEL = {
"list_rules": {
"input_token": "next_token",
"limit_key": "limit",
"limit_default": 50,
"unique_attribute": "arn",
"fail_on_invalid_token": False,
},
"list_rule_names_by_target": {
"input_token": "next_token",
"limit_key": "limit",
"limit_default": 50,
"unique_attribute": "arn",
"fail_on_invalid_token": False,
},
"list_targets_by_rule": {
"input_token": "next_token",
"limit_key": "limit",
"limit_default": 50,
"unique_attribute": "Arn",
"fail_on_invalid_token": False,
},
}
_BASE_EVENT_MESSAGE: EventMessageType = {
"version": "0",
"id": "17793124-05d4-b198-2fde-7ededc63b103",
"detail-type": "",
"source": "",
"account": "",
"time": "",
"region": "",
"resources": [],
"detail": {},
}
class EventTemplateParser:
DEFAULT_EVENT_INPUT_TEMPLATE = '{"id": "<id>", "time": <time>, "version": "0", "detail-type": "<detail-type>", "source": "<source>", "region": "<region>", "resources": <resources>, "detail": <detail>}'
DEFAULT_EVENT_INPUT_PATHS_MAP = {
"account": "$.account",
"detail": "$.detail",
"detail-type": "$.detail-type",
"source": "$.source",
"id": "$.id",
"region": "$.region",
"resources": "$.resources",
"time": "$.time",
}
@staticmethod
def _stringify(result: Any) -> str: # type: ignore[misc]
if isinstance(result, dict):
result = json.dumps(result)
elif isinstance(result, list):
result = json.dumps([EventTemplateParser._stringify(x) for x in result])
elif isinstance(result, (int, float)):
result = str(result)
return result
@staticmethod
def parse( # type: ignore[misc]
input_template: str, input_paths_map: dict[str, Any], event: EventMessageType
) -> dict[str, Any]:
from jsonpath_ng.ext import parse
template_to_use = (
input_template or EventTemplateParser.DEFAULT_EVENT_INPUT_TEMPLATE
)
input_paths_map = (
input_paths_map or EventTemplateParser.DEFAULT_EVENT_INPUT_PATHS_MAP
)
for input_path in input_paths_map:
input_expr = parse(input_paths_map[input_path])
matches = input_expr.find(event)
result = (
EventTemplateParser._stringify(matches[0].value) if matches else None
)
if result:
template_to_use = template_to_use.replace(f"<{input_path}>", result)
default_inputs_map = {
"aws.events.event.json": event,
"aws.events.event.ingestion-time": event["ingestion-time"],
}
for input_path in default_inputs_map:
result = EventTemplateParser._stringify(default_inputs_map[input_path])
template_to_use = template_to_use.replace(f"<{input_path}>", result)
return json.loads(template_to_use)
|
