Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Techdebt: MyPy Quicksight (#6194)

Browse Source
This commit is contained in:
Bert Blommers 2023-04-09 14:21:53 +00:00 committed by GitHub
parent 201e57585b
commit c45e79a566
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 108 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
moto/quicksight/exceptions.py
View File

@ -3,5 +3,5 @@ from moto.core.exceptions import JsonRESTError
class ResourceNotFoundException(JsonRESTError):
def __init__(self, msg):
def __init__(self, msg: str):
super().__init__("ResourceNotFoundException", msg)

116
moto/quicksight/models.py
View File

@ -1,22 +1,21 @@
"""QuickSightBackend class with methods for supported APIs."""
from typing import Any, Dict, Iterable
from moto.core import BaseBackend, BackendDict, BaseModel
from .exceptions import ResourceNotFoundException
def _create_id(aws_account_id, namespace, _id):
def _create_id(aws_account_id: str, namespace: str, _id: str) -> str:
return f"{aws_account_id}:{namespace}:{_id}"
class QuicksightDataSet(BaseModel):
def __init__(self, account_id, region, _id, name):
def __init__(self, account_id: str, region: str, _id: str, name: str):
self.arn = f"arn:aws:quicksight:{region}:{account_id}:data-set/{_id}"
self._id = _id
self.name = name
self.region = region
self.account_id = account_id
def to_json(self):
def to_json(self) -> Dict[str, Any]:
return {
"Arn": self.arn,
"DataSetId": self._id,
@ -25,11 +24,13 @@ class QuicksightDataSet(BaseModel):
class QuicksightIngestion(BaseModel):
def __init__(self, account_id, region, data_set_id, ingestion_id):
def __init__(
self, account_id: str, region: str, data_set_id: str, ingestion_id: str
):
self.arn = f"arn:aws:quicksight:{region}:{account_id}:data-set/{data_set_id}/ingestions/{ingestion_id}"
self.ingestion_id = ingestion_id
def to_json(self):
def to_json(self) -> Dict[str, Any]:
return {
"Arn": self.arn,
"IngestionId": self.ingestion_id,
@ -38,19 +39,26 @@ class QuicksightIngestion(BaseModel):
class QuicksightMembership(BaseModel):
def __init__(self, account_id, region, group, user):
def __init__(self, account_id: str, region: str, group: str, user: str):
self.group = group
self.user = user
self.arn = (
f"arn:aws:quicksight:{region}:{account_id}:group/default/{group}/{user}"
)
def to_json(self):
def to_json(self) -> Dict[str, str]:
return {"Arn": self.arn, "MemberName": self.user}
class QuicksightGroup(BaseModel):
def __init__(self, region, group_name, description, aws_account_id, namespace):
def __init__(
self,
region: str,
group_name: str,
description: str,
aws_account_id: str,
namespace: str,
):
self.arn = (
f"arn:aws:quicksight:{region}:{aws_account_id}:group/default/{group_name}"
)
@ -60,25 +68,25 @@ class QuicksightGroup(BaseModel):
self.namespace = namespace
self.region = region
self.members = dict()
self.members: Dict[str, QuicksightMembership] = dict()
def add_member(self, user_name):
def add_member(self, user_name: str) -> QuicksightMembership:
membership = QuicksightMembership(
self.aws_account_id, self.region, self.group_name, user_name
)
self.members[user_name] = membership
return membership
def delete_member(self, user_name):
def delete_member(self, user_name: str) -> None:
self.members.pop(user_name, None)
def get_member(self, user_name):
def get_member(self, user_name: str) -> QuicksightMembership:
return self.members[user_name]
def list_members(self):
def list_members(self) -> Iterable[QuicksightMembership]:
return self.members.values()
def to_json(self):
def to_json(self) -> Dict[str, Any]:
return {
"Arn": self.arn,
"GroupName": self.group_name,
@ -89,7 +97,15 @@ class QuicksightGroup(BaseModel):
class QuicksightUser(BaseModel):
def __init__(self, account_id, region, email, identity_type, username, user_role):
def __init__(
self,
account_id: str,
region: str,
email: str,
identity_type: str,
username: str,
user_role: str,
):
self.arn = f"arn:aws:quicksight:{region}:{account_id}:user/default/{username}"
self.email = email
self.identity_type = identity_type
@ -97,7 +113,7 @@ class QuicksightUser(BaseModel):
self.user_role = user_role
self.active = False
def to_json(self):
def to_json(self) -> Dict[str, Any]:
return {
"Arn": self.arn,
"Email": self.email,
@ -111,17 +127,19 @@ class QuicksightUser(BaseModel):
class QuickSightBackend(BaseBackend):
"""Implementation of QuickSight APIs."""
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.groups = dict()
self.users = dict()
self.groups: Dict[str, QuicksightGroup] = dict()
self.users: Dict[str, QuicksightUser] = dict()
def create_data_set(self, data_set_id, name):
def create_data_set(self, data_set_id: str, name: str) -> QuicksightDataSet:
return QuicksightDataSet(
self.account_id, self.region_name, data_set_id, name=name
)
def create_group(self, group_name, description, aws_account_id, namespace):
def create_group(
self, group_name: str, description: str, aws_account_id: str, namespace: str
) -> QuicksightGroup:
group = QuicksightGroup(
region=self.region_name,
group_name=group_name,
@ -133,20 +151,26 @@ class QuickSightBackend(BaseBackend):
self.groups[_id] = group
return group
def create_group_membership(self, aws_account_id, namespace, group_name, user_name):
def create_group_membership(
self, aws_account_id: str, namespace: str, group_name: str, user_name: str
) -> QuicksightMembership:
group = self.describe_group(aws_account_id, namespace, group_name)
return group.add_member(user_name)
def create_ingestion(self, data_set_id, ingestion_id):
def create_ingestion(
self, data_set_id: str, ingestion_id: str
) -> QuicksightIngestion:
return QuicksightIngestion(
self.account_id, self.region_name, data_set_id, ingestion_id
)
def delete_group(self, aws_account_id, namespace, group_name):
def delete_group(
self, aws_account_id: str, namespace: str, group_name: str
) -> None:
_id = _create_id(aws_account_id, namespace, group_name)
self.groups.pop(_id, None)
def delete_user(self, aws_account_id, namespace, user_name):
def delete_user(self, aws_account_id: str, namespace: str, user_name: str) -> None:
# Delete users from all groups
for group in self.groups.values():
group.delete_member(user_name)
@ -154,25 +178,31 @@ class QuickSightBackend(BaseBackend):
_id = _create_id(aws_account_id, namespace, user_name)
self.users.pop(_id, None)
def describe_group(self, aws_account_id, namespace, group_name):
def describe_group(
self, aws_account_id: str, namespace: str, group_name: str
) -> QuicksightGroup:
_id = _create_id(aws_account_id, namespace, group_name)
if _id not in self.groups:
raise ResourceNotFoundException(f"Group {group_name} not found")
return self.groups[_id]
def describe_group_membership(
self, aws_account_id, namespace, group_name, user_name
):
self, aws_account_id: str, namespace: str, group_name: str, user_name: str
) -> QuicksightMembership:
group = self.describe_group(aws_account_id, namespace, group_name)
return group.get_member(user_name)
def describe_user(self, aws_account_id, namespace, user_name):
def describe_user(
self, aws_account_id: str, namespace: str, user_name: str
) -> QuicksightUser:
_id = _create_id(aws_account_id, namespace, user_name)
if _id not in self.users:
raise ResourceNotFoundException(f"User {user_name} not found")
return self.users[_id]
def list_groups(self, aws_account_id, namespace):
def list_groups(
self, aws_account_id: str, namespace: str
) -> Iterable[QuicksightGroup]:
"""
The NextToken and MaxResults parameters are not yet implemented
"""
@ -181,14 +211,18 @@ class QuickSightBackend(BaseBackend):
group for _id, group in self.groups.items() if _id.startswith(id_for_ns)
]
def list_group_memberships(self, aws_account_id, namespace, group_name):
def list_group_memberships(
self, aws_account_id: str, namespace: str, group_name: str
) -> Iterable[QuicksightMembership]:
"""
The NextToken and MaxResults parameters are not yet implemented
"""
group = self.describe_group(aws_account_id, namespace, group_name)
return group.list_members()
def list_users(self, aws_account_id, namespace):
def list_users(
self, aws_account_id: str, namespace: str
) -> Iterable[QuicksightUser]:
"""
The NextToken and MaxResults parameters are not yet implemented
"""
@ -196,8 +230,14 @@ class QuickSightBackend(BaseBackend):
return [user for _id, user in self.users.items() if _id.startswith(id_for_ns)]
def register_user(
self, identity_type, email, user_role, aws_account_id, namespace, user_name
):
self,
identity_type: str,
email: str,
user_role: str,
aws_account_id: str,
namespace: str,
user_name: str,
) -> QuicksightUser:
"""
The following parameters are not yet implemented:
IamArn, SessionName, CustomsPermissionsName, ExternalLoginFederationProviderType, CustomFederationProviderUrl, ExternalLoginId
@ -214,7 +254,9 @@ class QuickSightBackend(BaseBackend):
self.users[_id] = user
return user
def update_group(self, aws_account_id, namespace, group_name, description):
def update_group(
self, aws_account_id: str, namespace: str, group_name: str, description: str
) -> QuicksightGroup:
group = self.describe_group(aws_account_id, namespace, group_name)
group.description = description
return group

54
moto/quicksight/responses.py
View File

@ -1,34 +1,34 @@
"""Handles incoming quicksight requests, invokes methods, returns responses."""
import json
from typing import Any
from moto.core.common_types import TYPE_RESPONSE
from moto.core.responses import BaseResponse
from .models import quicksight_backends
from .models import quicksight_backends, QuickSightBackend
class QuickSightResponse(BaseResponse):
"""Handler for QuickSight requests and responses."""
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="quicksight")
@property
def quicksight_backend(self):
def quicksight_backend(self) -> QuickSightBackend:
"""Return backend instance specific for this region."""
return quicksight_backends[self.current_account][self.region]
def dataset(self, request, full_url, headers):
def dataset(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "POST":
return self.create_data_set()
def groups(self, request, full_url, headers):
def groups(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "POST":
return self.create_group()
if request.method == "GET":
return self.list_groups()
def group(self, request, full_url, headers):
def group(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "GET":
return self.describe_group()
@ -37,45 +37,45 @@ class QuickSightResponse(BaseResponse):
if request.method == "PUT":
return self.update_group()
def group_member(self, request, full_url, headers):
def group_member(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "PUT":
return self.create_group_membership()
if request.method == "GET":
return self.describe_group_membership()
def group_members(self, request, full_url, headers):
def group_members(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "GET":
return self.list_group_memberships()
def ingestion(self, request, full_url, headers):
def ingestion(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "PUT":
return self.create_ingestion()
def users(self, request, full_url, headers):
def users(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "POST":
return self.register_user()
if request.method == "GET":
return self.list_users()
def user(self, request, full_url, headers):
def user(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "GET":
return self.describe_user()
if request.method == "DELETE":
return self.delete_user()
def create_data_set(self):
def create_data_set(self) -> TYPE_RESPONSE:
params = json.loads(self.body)
data_set_id = params.get("DataSetId")
name = params.get("Name")
data_set = self.quicksight_backend.create_data_set(data_set_id, name)
return 200, {}, json.dumps(data_set.to_json())
def create_group(self):
def create_group(self) -> TYPE_RESPONSE:
params = json.loads(self.body)
group_name = params.get("GroupName")
description = params.get("Description")
@ -89,7 +89,7 @@ class QuickSightResponse(BaseResponse):
)
return 200, {}, json.dumps(dict(Group=group.to_json()))
def create_group_membership(self):
def create_group_membership(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-7]
namespace = self.path.split("/")[-5]
group_name = self.path.split("/")[-3]
@ -99,13 +99,13 @@ class QuickSightResponse(BaseResponse):
)
return 200, {}, json.dumps({"GroupMember": member.to_json()})
def create_ingestion(self):
def create_ingestion(self) -> TYPE_RESPONSE:
data_set_id = self.path.split("/")[-3]
ingestion_id = self.path.split("/")[-1]
ingestion = self.quicksight_backend.create_ingestion(data_set_id, ingestion_id)
return 200, {}, json.dumps(ingestion.to_json())
def describe_group_membership(self):
def describe_group_membership(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-7]
namespace = self.path.split("/")[-5]
group_name = self.path.split("/")[-3]
@ -115,13 +115,13 @@ class QuickSightResponse(BaseResponse):
)
return 200, {}, json.dumps({"GroupMember": member.to_json()})
def list_groups(self):
def list_groups(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-4]
namespace = self.path.split("/")[-2]
groups = self.quicksight_backend.list_groups(aws_account_id, namespace)
return 200, {}, json.dumps(dict(GroupList=[g.to_json() for g in groups]))
def list_group_memberships(self):
def list_group_memberships(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-6]
namespace = self.path.split("/")[-4]
group_name = self.path.split("/")[-2]
@ -130,13 +130,13 @@ class QuickSightResponse(BaseResponse):
)
return 200, {}, json.dumps({"GroupMemberList": [m.to_json() for m in members]})
def list_users(self):
def list_users(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-4]
namespace = self.path.split("/")[-2]
users = self.quicksight_backend.list_users(aws_account_id, namespace)
return 200, {}, json.dumps(dict(UserList=[u.to_json() for u in users]))
def register_user(self):
def register_user(self) -> TYPE_RESPONSE:
params = json.loads(self.body)
identity_type = params.get("IdentityType")
email = params.get("Email")
@ -154,7 +154,7 @@ class QuickSightResponse(BaseResponse):
)
return 200, {}, json.dumps(dict(User=user.to_json(), UserInvitationUrl="TBD"))
def describe_group(self):
def describe_group(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-5]
namespace = self.path.split("/")[-3]
group_name = self.path.split("/")[-1]
@ -164,7 +164,7 @@ class QuickSightResponse(BaseResponse):
)
return 200, {}, json.dumps(dict(Group=group.to_json()))
def describe_user(self):
def describe_user(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-5]
namespace = self.path.split("/")[-3]
user_name = self.path.split("/")[-1]
@ -174,7 +174,7 @@ class QuickSightResponse(BaseResponse):
)
return 200, {}, json.dumps(dict(User=user.to_json()))
def delete_group(self):
def delete_group(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-5]
namespace = self.path.split("/")[-3]
group_name = self.path.split("/")[-1]
@ -182,7 +182,7 @@ class QuickSightResponse(BaseResponse):
self.quicksight_backend.delete_group(aws_account_id, namespace, group_name)
return 204, {}, json.dumps({"Status": 204})
def delete_user(self):
def delete_user(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-5]
namespace = self.path.split("/")[-3]
user_name = self.path.split("/")[-1]
@ -190,7 +190,7 @@ class QuickSightResponse(BaseResponse):
self.quicksight_backend.delete_user(aws_account_id, namespace, user_name)
return 204, {}, json.dumps({"Status": 204})
def update_group(self):
def update_group(self) -> TYPE_RESPONSE:
aws_account_id = self.path.split("/")[-5]
namespace = self.path.split("/")[-3]
group_name = self.path.split("/")[-1]

2
setup.cfg
View File

@ -235,7 +235,7 @@ disable = W,C,R,E
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import
[mypy]
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/rdsdata
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/q*,moto/rdsdata
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract