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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
|
"""Handles incoming lakeformation requests, invokes methods, returns responses."""
import json
from typing import Any
from moto.core.responses import BaseResponse
from .exceptions import InvalidInput
from .models import (
LakeFormationBackend,
ListPermissionsResource,
ListPermissionsResourceDatabase,
ListPermissionsResourceDataLocation,
ListPermissionsResourceTable,
RessourceType,
lakeformation_backends,
)
class LakeFormationResponse(BaseResponse):
"""Handler for LakeFormation requests and responses."""
def __init__(self) -> None:
super().__init__(service_name="lakeformation")
@property
def lakeformation_backend(self) -> LakeFormationBackend:
"""Return backend instance specific for this region."""
return lakeformation_backends[self.current_account][self.region]
def describe_resource(self) -> str:
resource_arn = self._get_param("ResourceArn")
resource = self.lakeformation_backend.describe_resource(
resource_arn=resource_arn
)
return json.dumps({"ResourceInfo": resource.to_dict()})
def deregister_resource(self) -> str:
resource_arn = self._get_param("ResourceArn")
self.lakeformation_backend.deregister_resource(resource_arn=resource_arn)
return "{}"
def register_resource(self) -> str:
resource_arn = self._get_param("ResourceArn")
role_arn = self._get_param("RoleArn")
self.lakeformation_backend.register_resource(
resource_arn=resource_arn,
role_arn=role_arn,
)
return "{}"
def list_resources(self) -> str:
resources = self.lakeformation_backend.list_resources()
return json.dumps({"ResourceInfoList": [res.to_dict() for res in resources]})
def get_data_lake_settings(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
settings = self.lakeformation_backend.get_data_lake_settings(catalog_id)
return json.dumps({"DataLakeSettings": settings})
def put_data_lake_settings(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
settings = self._get_param("DataLakeSettings")
self.lakeformation_backend.put_data_lake_settings(catalog_id, settings)
return "{}"
def grant_permissions(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
principal = self._get_param("Principal")
resource = self._get_param("Resource")
permissions = self._get_param("Permissions")
permissions_with_grant_options = self._get_param("PermissionsWithGrantOption")
self.lakeformation_backend.grant_permissions(
catalog_id=catalog_id,
principal=principal,
resource=resource,
permissions=permissions,
permissions_with_grant_options=permissions_with_grant_options,
)
return "{}"
def revoke_permissions(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
principal = self._get_param("Principal")
resource = self._get_param("Resource")
permissions = self._get_param("Permissions")
permissions_with_grant_options = (
self._get_param("PermissionsWithGrantOption") or []
)
self.lakeformation_backend.revoke_permissions(
catalog_id=catalog_id,
principal=principal,
resource=resource,
permissions_to_revoke=permissions,
permissions_with_grant_options_to_revoke=permissions_with_grant_options,
)
return "{}"
def list_permissions(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
principal = self._get_param("Principal")
resource = self._get_param("Resource")
resource_type_param = self._get_param("ResourceType")
if principal is not None and resource is None:
# Error message is the exact string returned by the AWS-CLI
raise InvalidInput(
"An error occurred (InvalidInputException) when calling the ListPermissions operation: Resource is mandatory if Principal is set in the input."
)
if resource_type_param is None:
resource_type = None
else:
resource_type = RessourceType(resource_type_param)
if resource is None:
list_permission_resource = None
else:
database_sub_dictionary = resource.get("Database")
table_sub_dictionary = resource.get("Table")
catalog_sub_dictionary = resource.get("Catalog")
data_location_sub_dictionary = resource.get("DataLocation")
if database_sub_dictionary is None:
database = None
else:
database = ListPermissionsResourceDatabase(
name=database_sub_dictionary.get("Name"),
catalog_id=database_sub_dictionary.get("CatalogId"),
)
if table_sub_dictionary is None:
table = None
else:
table = ListPermissionsResourceTable(
database_name=table_sub_dictionary.get("DatabaseName"),
name=table_sub_dictionary.get("Name"),
catalog_id=table_sub_dictionary.get("CatalogId"),
table_wildcard=table_sub_dictionary.get("TableWildcard"),
)
if data_location_sub_dictionary is None:
data_location = None
else:
data_location = ListPermissionsResourceDataLocation(
resource_arn=data_location_sub_dictionary.get("ResourceArn"),
catalog_id=data_location_sub_dictionary.get("CatalogId"),
)
list_permission_resource = ListPermissionsResource(
catalog=catalog_sub_dictionary,
database=database,
table=table,
table_with_columns=None,
data_location=data_location,
data_cells_filter=None,
lf_tag=None,
lf_tag_policy=None,
)
permissions = self.lakeformation_backend.list_permissions(
catalog_id=catalog_id,
principal=principal,
resource=list_permission_resource,
resource_type=resource_type,
)
return json.dumps({"PrincipalResourcePermissions": permissions})
def create_lf_tag(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
key = self._get_param("TagKey")
values = self._get_param("TagValues")
self.lakeformation_backend.create_lf_tag(catalog_id, key, values)
return "{}"
def get_lf_tag(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
key = self._get_param("TagKey")
tag_values = self.lakeformation_backend.get_lf_tag(catalog_id, key)
return json.dumps(
{"CatalogId": catalog_id, "TagKey": key, "TagValues": tag_values}
)
def delete_lf_tag(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
key = self._get_param("TagKey")
self.lakeformation_backend.delete_lf_tag(catalog_id, key)
return "{}"
def list_lf_tags(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
tags = self.lakeformation_backend.list_lf_tags(catalog_id)
return json.dumps(
{
"LFTags": [
{"CatalogId": catalog_id, "TagKey": tag, "TagValues": value}
for tag, value in tags.items()
]
}
)
def update_lf_tag(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
tag_key = self._get_param("TagKey")
to_delete = self._get_param("TagValuesToDelete")
to_add = self._get_param("TagValuesToAdd")
self.lakeformation_backend.update_lf_tag(catalog_id, tag_key, to_delete, to_add)
return "{}"
def list_data_cells_filter(self) -> str:
data_cells = self.lakeformation_backend.list_data_cells_filter()
return json.dumps({"DataCellsFilters": data_cells})
def batch_grant_permissions(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
entries = self._get_param("Entries")
self.lakeformation_backend.batch_grant_permissions(catalog_id, entries)
return json.dumps({"Failures": []})
def batch_revoke_permissions(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
entries = self._get_param("Entries")
self.lakeformation_backend.batch_revoke_permissions(catalog_id, entries)
return json.dumps({"Failures": []})
def add_lf_tags_to_resource(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
resource = self._get_param("Resource")
tags = self._get_param("LFTags")
failures = self.lakeformation_backend.add_lf_tags_to_resource(
catalog_id, resource, tags
)
return json.dumps({"Failures": failures})
def get_resource_lf_tags(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
resource = self._get_param("Resource")
db, table, columns = self.lakeformation_backend.get_resource_lf_tags(
catalog_id, resource
)
resp: dict[str, Any] = {}
if db:
resp["LFTagOnDatabase"] = db
if table:
resp["LFTagsOnTable"] = table
if columns:
resp["LFTagsOnColumns"] = columns
return json.dumps(resp)
def remove_lf_tags_from_resource(self) -> str:
catalog_id = self._get_param("CatalogId") or self.current_account
resource = self._get_param("Resource")
tags = self._get_param("LFTags")
self.lakeformation_backend.remove_lf_tags_from_resource(
catalog_id, resource, tags
)
return "{}"
|
