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
|
"""Handles incoming backup requests, invokes methods, returns responses."""
import json
from urllib.parse import unquote
from moto.core.responses import BaseResponse
from .models import BackupBackend, backup_backends
class BackupResponse(BaseResponse):
"""Handler for Backup requests and responses."""
def __init__(self) -> None:
super().__init__(service_name="backup")
@property
def backup_backend(self) -> BackupBackend:
"""Return backend instance specific for this region."""
return backup_backends[self.current_account][self.region]
def create_backup_plan(self) -> str:
params = json.loads(self.body)
backup_plan = params.get("BackupPlan")
backup_plan_tags = params.get("BackupPlanTags")
creator_request_id = params.get("CreatorRequestId")
plan = self.backup_backend.create_backup_plan(
backup_plan=backup_plan,
backup_plan_tags=backup_plan_tags,
creator_request_id=creator_request_id,
)
return json.dumps(dict(plan.to_dict()))
def get_backup_plan(self) -> str:
params = self._get_params()
backup_plan_id = self.path.split("/")[-2]
version_id = params.get("versionId")
plan = self.backup_backend.get_backup_plan(
backup_plan_id=backup_plan_id, version_id=version_id
)
return json.dumps(dict(plan.to_get_dict()))
def delete_backup_plan(self) -> str:
backup_plan_id = self.path.split("/")[-1]
(
backup_plan_id,
backup_plan_arn,
deletion_date,
version_id,
) = self.backup_backend.delete_backup_plan(
backup_plan_id=backup_plan_id,
)
return json.dumps(
{
"BackupPlanId": backup_plan_id,
"BackupPlanArn": backup_plan_arn,
"DeletionDate": deletion_date,
"VersionId": version_id,
}
)
def list_backup_plans(self) -> str:
params = self._get_params()
include_deleted = params.get("includeDeleted")
backup_plans_list = self.backup_backend.list_backup_plans(
include_deleted=include_deleted
)
return json.dumps(
{"BackupPlansList": [p.to_list_dict() for p in backup_plans_list]}
)
def create_backup_vault(self) -> str:
params = json.loads(self.body)
backup_vault_name = self.path.split("/")[-1]
backup_vault_tags = params.get("BackupVaultTags")
encryption_key_arn = params.get("EncryptionKeyArn")
creator_request_id = params.get("CreatorRequestId")
backup_vault = self.backup_backend.create_backup_vault(
backup_vault_name=backup_vault_name,
backup_vault_tags=backup_vault_tags,
encryption_key_arn=encryption_key_arn,
creator_request_id=creator_request_id,
)
return json.dumps(dict(backup_vault.to_dict()))
def list_backup_vaults(self) -> str:
backup_vault_list = self.backup_backend.list_backup_vaults()
return json.dumps(
{"BackupVaultList": [v.to_list_dict() for v in backup_vault_list]}
)
def list_tags(self) -> str:
resource_arn = unquote(self.path.split("/")[-2])
tags = self.backup_backend.list_tags(
resource_arn=resource_arn,
)
return json.dumps({"Tags": tags})
def tag_resource(self) -> str:
params = json.loads(self.body)
resource_arn = unquote(self.path.split("/")[-1])
tags = params.get("Tags")
self.backup_backend.tag_resource(
resource_arn=resource_arn,
tags=tags,
)
return "{}"
def untag_resource(self) -> str:
params = json.loads(self.body)
resource_arn = unquote(self.path.split("/")[-1])
tag_key_list = params.get("TagKeyList")
self.backup_backend.untag_resource(
resource_arn=resource_arn,
tag_key_list=tag_key_list,
)
return "{}"
|
