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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
from typing import Any, Union
from moto.core.common_models import BaseModel
from .data_models import QuicksightGroup
from .exceptions import ParamValidationError, SchemaException, ValidationException
PAGINATION_MODEL = {
"list_users": {
"input_token": "next_token",
"limit_key": "max_results",
"limit_default": 100, # This should be the sum of the directory limits
"unique_attribute": "arn",
},
"list_user_groups": {
"input_token": "next_token",
"limit_key": "max_results",
"limit_default": 100, # This should be the sum of the directory limits
"unique_attribute": "arn",
},
"list_groups": {
"input_token": "next_token",
"limit_key": "max_results",
"limit_default": 100, # This should be the sum of the directory limits
"unique_attribute": "arn",
},
"list_group_memberships": {
"input_token": "next_token",
"limit_key": "max_results",
"limit_default": 100, # This should be the sum of the directory limits
"unique_attribute": "arn",
},
"search_groups": {
"input_token": "next_token",
"limit_key": "max_results",
"limit_default": 100, # This should be the sum of the directory limits
"unique_attribute": "arn",
},
"list_dashboards": {
"input_token": "next_token",
"limit_key": "max_results",
"limit_default": 100, # This should be the sum of the directory limits
"unique_attribute": "arn",
},
"list_data_sources": {
"input_token": "next_token",
"limit_key": "max_results",
"limit_default": 100, # This should be the sum of the directory limits
"unique_attribute": "arn",
},
}
class QuicksightBaseSearchFilter(BaseModel):
"""Base Search Filter."""
schema: dict[str, Any] = {
"$schema": "https://json-schema.org/draft/2020-12/schema",
}
def __init__(self, operator: str, name: str, value: str):
self.operator = operator
self.name = name
self.value = value
def match(self, input: BaseModel) -> bool:
raise NotImplementedError
class QuicksightGroupSearchFilter(QuicksightBaseSearchFilter):
"""Group Search Filter."""
schema = {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "array",
"items": {
"type": "object",
"properties": {
"Operator": {"type": "string", "enum": ["StringEquals", "StartsWith"]},
"Name": {"type": "string", "enum": ["GROUP_NAME"]},
"Value": {"type": "string"},
},
"required": ["Operator", "Name", "Value"],
},
"minItems": 1,
"maxItems": 1,
}
def __init__(self, operator: str, name: str, value: str):
super().__init__(operator, name, value)
def match(self, input: BaseModel) -> bool:
if isinstance(input, QuicksightGroup):
if self.name == "GROUP_NAME" and self.operator == "StringEquals":
return input.group_name.lower() == self.value.lower()
if self.name == "GROUP_NAME" and self.operator == "StartsWith":
return input.group_name.lower().startswith(self.value.lower())
return False
@classmethod
def parse_filter(cls, filter: dict[str, str]) -> QuicksightBaseSearchFilter:
return QuicksightGroupSearchFilter(
operator=filter.get("Operator", ""),
name=filter.get("Name", ""),
value=filter.get("Value", ""),
)
class QuicksightSearchFilterList:
"""Generic QuickSight Search Filter List."""
def __init__(self, filters: list[QuicksightBaseSearchFilter]):
self.filters: list[QuicksightBaseSearchFilter] = filters
def match(self, input: BaseModel) -> bool:
return any(filter.match(input) for filter in self.filters)
class QuicksightSearchFilterFactory:
"""Creates the appropriate search filter."""
@classmethod
def validate_and_create_filter(
cls, model_type: type[BaseModel], input: Union[list[dict[str, str]], None]
) -> QuicksightSearchFilterList:
if issubclass(model_type, QuicksightGroup):
if input is None:
# Should never happen as there are other validations before but just in case.
raise ParamValidationError(
'Missing required parameter in input: "Filters"'
)
from jsonschema import validate
from jsonschema.exceptions import SchemaError, ValidationError
try:
validate(instance=input, schema=QuicksightGroupSearchFilter.schema)
return QuicksightSearchFilterList(
filters=[
QuicksightGroupSearchFilter.parse_filter(filter)
for filter in input
]
)
except ValidationError as err:
raise ValidationException(err.message)
except SchemaError as err:
# This exception can happen only, when the schema defined in moto is wrong.
# The schema is not presented by the user.
raise SchemaException(f"Schema definition wrong in moto: {err.message}")
raise NotImplementedError(
f"Filter for '{model_type.__name__}' not implemented."
)
|
