diff --git a/moto/quicksight/exceptions.py b/moto/quicksight/exceptions.py index 045cf4c7e..c456fe015 100644 --- a/moto/quicksight/exceptions.py +++ b/moto/quicksight/exceptions.py @@ -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) diff --git a/moto/quicksight/models.py b/moto/quicksight/models.py index 495e67014..da8646c01 100644 --- a/moto/quicksight/models.py +++ b/moto/quicksight/models.py @@ -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 diff --git a/moto/quicksight/responses.py b/moto/quicksight/responses.py index 4ef3efb69..1ec85525a 100644 --- a/moto/quicksight/responses.py +++ b/moto/quicksight/responses.py @@ -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] diff --git a/setup.cfg b/setup.cfg index a362aea2a..2f4529ffb 100644 --- a/setup.cfg +++ b/setup.cfg @@ -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