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
|
import json
from moto.core.responses import BaseResponse
from .models import ResourceGroupsTaggingAPIBackend, resourcegroupstaggingapi_backends
class ResourceGroupsTaggingAPIResponse(BaseResponse):
def __init__(self) -> None:
super().__init__(service_name="resourcegroupstaggingapi")
@property
def backend(self) -> ResourceGroupsTaggingAPIBackend:
return resourcegroupstaggingapi_backends[self.current_account][self.region]
def get_resources(self) -> str:
pagination_token = self._get_param("PaginationToken")
tag_filters = self._get_param("TagFilters", [])
resources_per_page = self._get_int_param("ResourcesPerPage", 50)
tags_per_page = self._get_int_param("TagsPerPage", 100)
resource_type_filters = self._get_param("ResourceTypeFilters", [])
pagination_token, resource_tag_mapping_list = self.backend.get_resources(
pagination_token=pagination_token,
tag_filters=tag_filters,
resources_per_page=resources_per_page,
tags_per_page=tags_per_page,
resource_type_filters=resource_type_filters,
)
# Format tag response
response = {
"ResourceTagMappingList": resource_tag_mapping_list,
"PaginationToken": pagination_token,
}
return json.dumps(response)
def get_tag_keys(self) -> str:
pagination_token = self._get_param("PaginationToken")
pagination_token, tag_keys = self.backend.get_tag_keys(
pagination_token=pagination_token
)
response = {"TagKeys": tag_keys, "PaginationToken": pagination_token}
return json.dumps(response)
def get_tag_values(self) -> str:
pagination_token = self._get_param("PaginationToken")
key = self._get_param("Key")
pagination_token, tag_values = self.backend.get_tag_values(
pagination_token=pagination_token, key=key
)
response = {"TagValues": tag_values, "PaginationToken": pagination_token}
return json.dumps(response)
def tag_resources(self) -> str:
resource_arns = self._get_param("ResourceARNList")
tags = self._get_param("Tags")
failed_resources = self.backend.tag_resources(
resource_arns=resource_arns, tags=tags
)
return json.dumps({"FailedResourcesMap": failed_resources})
def untag_resources(self) -> str:
resource_arn_list = self._get_param("ResourceARNList")
tag_keys = self._get_param("TagKeys")
failed_resources = self.backend.untag_resources(
resource_arn_list=resource_arn_list,
tag_keys=tag_keys,
)
return json.dumps({"FailedResourcesMap": failed_resources})
|
