From 14d3c5a76a5b34b030d7a5d3137c00dba79d6c81 Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Mon, 27 Mar 2023 18:00:24 +0100 Subject: [PATCH] Techdebt: MyPy ManagedBlockchain (#6134) --- moto/managedblockchain/exceptions.py | 30 +- moto/managedblockchain/models.py | 422 ++++++++++++++------------- moto/managedblockchain/responses.py | 127 +++++--- moto/managedblockchain/utils.py | 84 +++--- setup.cfg | 2 +- 5 files changed, 360 insertions(+), 305 deletions(-) diff --git a/moto/managedblockchain/exceptions.py b/moto/managedblockchain/exceptions.py index cc78adaf3..8b1beb848 100644 --- a/moto/managedblockchain/exceptions.py +++ b/moto/managedblockchain/exceptions.py @@ -1,38 +1,46 @@ import json +from moto.core.common_types import TYPE_RESPONSE from moto.core.exceptions import JsonRESTError from functools import wraps +from typing import Any, Callable, List, Tuple -def exception_handler(f): +def exception_handler( + f: Callable[[Any, Any, Any, Any], TYPE_RESPONSE] +) -> Callable[[Any], TYPE_RESPONSE]: @wraps(f) - def _wrapper(*args, **kwargs): + def _wrapper(*args: Any, **kwargs: Any) -> TYPE_RESPONSE: # type: ignore[misc] try: return f(*args, **kwargs) except ManagedBlockchainClientError as err: - return err.code, err.get_headers(), err.description + return err.code, err.get_headers(), err.description # type: ignore return _wrapper class ManagedBlockchainClientError(JsonRESTError): - def __init__(self, error_type, message): + def __init__(self, error_type: str, message: str): super().__init__(error_type=error_type, message=message) self.error_type = error_type self.message = message self.description = json.dumps({"message": self.message}) - def get_headers(self, *args, **kwargs): # pylint: disable=unused-argument + def get_headers( + self, *args: Any, **kwargs: Any + ) -> List[Tuple[str, str]]: # pylint: disable=unused-argument return [ ("Content-Type", "application/json"), ("x-amzn-ErrorType", self.error_type), ] - def get_body(self, *args, **kwargs): # pylint: disable=unused-argument + def get_body( + self, *args: Any, **kwargs: Any + ) -> str: # pylint: disable=unused-argument return self.description class BadRequestException(ManagedBlockchainClientError): - def __init__(self, pretty_called_method, operation_error): + def __init__(self, pretty_called_method: str, operation_error: str): super().__init__( "BadRequestException", f"An error occurred (BadRequestException) when calling the {pretty_called_method} operation: {operation_error}", @@ -40,7 +48,7 @@ class BadRequestException(ManagedBlockchainClientError): class InvalidRequestException(ManagedBlockchainClientError): - def __init__(self, pretty_called_method, operation_error): + def __init__(self, pretty_called_method: str, operation_error: str): super().__init__( "InvalidRequestException", f"An error occurred (InvalidRequestException) when calling the {pretty_called_method} operation: {operation_error}", @@ -48,7 +56,7 @@ class InvalidRequestException(ManagedBlockchainClientError): class ResourceNotFoundException(ManagedBlockchainClientError): - def __init__(self, pretty_called_method, operation_error): + def __init__(self, pretty_called_method: str, operation_error: str): self.code = 404 super().__init__( "ResourceNotFoundException", @@ -57,7 +65,7 @@ class ResourceNotFoundException(ManagedBlockchainClientError): class ResourceAlreadyExistsException(ManagedBlockchainClientError): - def __init__(self, pretty_called_method, operation_error): + def __init__(self, pretty_called_method: str, operation_error: str): self.code = 409 super().__init__( "ResourceAlreadyExistsException", @@ -66,7 +74,7 @@ class ResourceAlreadyExistsException(ManagedBlockchainClientError): class ResourceLimitExceededException(ManagedBlockchainClientError): - def __init__(self, pretty_called_method, operation_error): + def __init__(self, pretty_called_method: str, operation_error: str): self.code = 429 super().__init__( "ResourceLimitExceededException", diff --git a/moto/managedblockchain/models.py b/moto/managedblockchain/models.py index 4456f95a0..7d5c85dba 100644 --- a/moto/managedblockchain/models.py +++ b/moto/managedblockchain/models.py @@ -1,5 +1,6 @@ import datetime import re +from typing import Any, Dict, List, Optional from moto.core import BaseBackend, BackendDict, BaseModel @@ -32,7 +33,7 @@ FRAMEWORKVERSIONS = [ "1.2", ] -EDITIONS = { +EDITIONS: Dict[str, Any] = { "STARTER": { "MaxMembers": 5, "MaxNodesPerMember": 2, @@ -51,15 +52,15 @@ VOTEVALUES = ["YES", "NO"] class ManagedBlockchainNetwork(BaseModel): def __init__( self, - network_id, - name, - framework, - frameworkversion, - frameworkconfiguration, - voting_policy, - member_configuration, - region, - description=None, + network_id: str, + name: str, + framework: str, + frameworkversion: str, + frameworkconfiguration: Dict[str, Any], + voting_policy: Dict[str, Any], + member_configuration: Dict[str, Any], + region: str, + description: Optional[str] = None, ): self.creationdate = datetime.datetime.utcnow() self.id = network_id @@ -73,42 +74,42 @@ class ManagedBlockchainNetwork(BaseModel): self.region = region @property - def network_name(self): + def network_name(self) -> str: return self.name @property - def network_framework(self): + def network_framework(self) -> str: return self.framework @property - def network_framework_version(self): + def network_framework_version(self) -> str: return self.frameworkversion @property - def network_creationdate(self): + def network_creationdate(self) -> str: return self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z") @property - def network_description(self): + def network_description(self) -> Optional[str]: return self.description @property - def network_edition(self): + def network_edition(self) -> str: return self.frameworkconfiguration["Fabric"]["Edition"] @property - def vote_pol_proposal_duration(self): + def vote_pol_proposal_duration(self) -> float: return self.voting_policy["ApprovalThresholdPolicy"]["ProposalDurationInHours"] @property - def vote_pol_threshold_percentage(self): + def vote_pol_threshold_percentage(self) -> float: return self.voting_policy["ApprovalThresholdPolicy"]["ThresholdPercentage"] @property - def vote_pol_threshold_comparator(self): + def vote_pol_threshold_comparator(self) -> str: return self.voting_policy["ApprovalThresholdPolicy"]["ThresholdComparator"] - def to_dict(self): + def to_dict(self) -> Dict[str, Any]: # Format for list_networks d = { "Id": self.id, @@ -122,7 +123,7 @@ class ManagedBlockchainNetwork(BaseModel): d["Description"] = self.description return d - def get_format(self): + def get_format(self) -> Dict[str, Any]: # Format for get_network frameworkattributes = { "Fabric": { @@ -154,16 +155,16 @@ class ManagedBlockchainNetwork(BaseModel): class ManagedBlockchainProposal(BaseModel): def __init__( self, - proposal_id, - networkid, - memberid, - membername, - numofmembers, - actions, - network_expirtation, - network_threshold, - network_threshold_comp, - description=None, + proposal_id: str, + networkid: str, + memberid: str, + membername: str, + numofmembers: int, + actions: Dict[str, Any], + network_expiration: float, + network_threshold: float, + network_threshold_comp: str, + description: Optional[str] = None, ): # In general, passing all values instead of creating # an apparatus to look them up @@ -173,60 +174,58 @@ class ManagedBlockchainProposal(BaseModel): self.membername = membername self.numofmembers = numofmembers self.actions = actions - self.network_expirtation = network_expirtation + self.network_expiration = network_expiration self.network_threshold = network_threshold self.network_threshold_comp = network_threshold_comp self.description = description self.creationdate = datetime.datetime.utcnow() - self.expirtationdate = self.creationdate + datetime.timedelta( - hours=network_expirtation + self.expirationdate = self.creationdate + datetime.timedelta( + hours=network_expiration ) self.yes_vote_count = 0 self.no_vote_count = 0 self.outstanding_vote_count = self.numofmembers self.status = "IN_PROGRESS" - self.votes = {} + self.votes: Dict[str, Dict[str, str]] = {} @property - def network_id(self): + def network_id(self) -> str: return self.networkid @property - def proposal_status(self): + def proposal_status(self) -> str: return self.status @property - def proposal_votes(self): + def proposal_votes(self) -> Dict[str, Any]: # type: ignore[misc] return self.votes - def proposal_actions(self, action_type): - default_return = [] + def proposal_actions(self, action_type: str) -> List[Dict[str, Any]]: if action_type.lower() == "invitations": if "Invitations" in self.actions: return self.actions["Invitations"] elif action_type.lower() == "removals": if "Removals" in self.actions: return self.actions["Removals"] - return default_return + return [] - def check_to_expire_proposal(self): - if datetime.datetime.utcnow() > self.expirtationdate: + def check_to_expire_proposal(self) -> None: + if datetime.datetime.utcnow() > self.expirationdate: self.status = "EXPIRED" - def to_dict(self): + def to_dict(self) -> Dict[str, Any]: # Format for list_proposals - d = { + return { "ProposalId": self.id, "ProposedByMemberId": self.memberid, "ProposedByMemberName": self.membername, "Status": self.status, "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), - "ExpirationDate": self.expirtationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), + "ExpirationDate": self.expirationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), } - return d - def get_format(self): + def get_format(self) -> Dict[str, Any]: # Format for get_proposal d = { "ProposalId": self.id, @@ -236,7 +235,7 @@ class ManagedBlockchainProposal(BaseModel): "ProposedByMemberName": self.membername, "Status": self.status, "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), - "ExpirationDate": self.expirtationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), + "ExpirationDate": self.expirationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "YesVoteCount": self.yes_vote_count, "NoVoteCount": self.no_vote_count, "OutstandingVoteCount": self.outstanding_vote_count, @@ -245,7 +244,7 @@ class ManagedBlockchainProposal(BaseModel): d["Description"] = self.description return d - def set_vote(self, votermemberid, votermembername, vote): + def set_vote(self, votermemberid: str, votermembername: str, vote: str) -> None: if vote.upper() == "YES": self.yes_vote_count += 1 else: @@ -284,14 +283,14 @@ class ManagedBlockchainProposal(BaseModel): class ManagedBlockchainInvitation(BaseModel): def __init__( self, - invitation_id, - networkid, - networkname, - networkframework, - networkframeworkversion, - networkcreationdate, - region, - networkdescription=None, + invitation_id: str, + networkid: str, + networkname: str, + networkframework: str, + networkframeworkversion: str, + networkcreationdate: str, + region: str, + networkdescription: Optional[str] = None, ): self.id = invitation_id self.networkid = networkid @@ -305,21 +304,21 @@ class ManagedBlockchainInvitation(BaseModel): self.region = region self.creationdate = datetime.datetime.utcnow() - self.expirtationdate = self.creationdate + datetime.timedelta(days=7) + self.expirationdate = self.creationdate + datetime.timedelta(days=7) @property - def invitation_status(self): + def invitation_status(self) -> str: return self.status @property - def invitation_networkid(self): + def invitation_networkid(self) -> str: return self.networkid - def to_dict(self): - d = { + def to_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = { "InvitationId": self.id, "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), - "ExpirationDate": self.expirtationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), + "ExpirationDate": self.expirationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "Status": self.status, "NetworkSummary": { "Id": self.networkid, @@ -334,18 +333,24 @@ class ManagedBlockchainInvitation(BaseModel): d["NetworkSummary"]["Description"] = self.networkdescription return d - def accept_invitation(self): + def accept_invitation(self) -> None: self.status = "ACCEPTED" - def reject_invitation(self): + def reject_invitation(self) -> None: self.status = "REJECTED" - def set_network_status(self, network_status): + def set_network_status(self, network_status: str) -> None: self.networkstatus = network_status class ManagedBlockchainMember(BaseModel): - def __init__(self, member_id, networkid, member_configuration, region): + def __init__( + self, + member_id: str, + networkid: str, + member_configuration: Dict[str, Any], + region: str, + ): self.creationdate = datetime.datetime.utcnow() self.id = member_id self.networkid = networkid @@ -355,18 +360,18 @@ class ManagedBlockchainMember(BaseModel): self.description = None @property - def network_id(self): + def network_id(self) -> str: return self.networkid @property - def name(self): + def name(self) -> str: return self.member_configuration["Name"] @property - def member_status(self): + def member_status(self) -> str: return self.status - def to_dict(self): + def to_dict(self) -> Dict[str, Any]: # Format for list_members d = { "Id": self.id, @@ -379,7 +384,7 @@ class ManagedBlockchainMember(BaseModel): self.description = self.member_configuration["Description"] return d - def get_format(self): + def get_format(self) -> Dict[str, Any]: # Format for get_member frameworkattributes = { "Fabric": { @@ -405,10 +410,10 @@ class ManagedBlockchainMember(BaseModel): d["Description"] = self.description return d - def delete(self): + def delete(self) -> None: self.status = "DELETED" - def update(self, logpublishingconfiguration): + def update(self, logpublishingconfiguration: Dict[str, Any]) -> None: self.member_configuration[ "LogPublishingConfiguration" ] = logpublishingconfiguration @@ -417,13 +422,13 @@ class ManagedBlockchainMember(BaseModel): class ManagedBlockchainNode(BaseModel): def __init__( self, - node_id, - networkid, - memberid, - availabilityzone, - instancetype, - logpublishingconfiguration, - region, + node_id: str, + networkid: str, + memberid: str, + availabilityzone: str, + instancetype: str, + logpublishingconfiguration: Dict[str, Any], + region: str, ): self.creationdate = datetime.datetime.utcnow() self.id = node_id @@ -436,25 +441,24 @@ class ManagedBlockchainNode(BaseModel): self.availabilityzone = availabilityzone @property - def member_id(self): + def member_id(self) -> str: return self.memberid @property - def node_status(self): + def node_status(self) -> str: return self.status - def to_dict(self): + def to_dict(self) -> Dict[str, Any]: # Format for list_nodes - d = { + return { "Id": self.id, "Status": self.status, "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "AvailabilityZone": self.availabilityzone, "InstanceType": self.instancetype, } - return d - def get_format(self): + def get_format(self) -> Dict[str, Any]: # Format for get_node frameworkattributes = { "Fabric": { @@ -463,7 +467,7 @@ class ManagedBlockchainNode(BaseModel): } } - d = { + return { "NetworkId": self.networkid, "MemberId": self.memberid, "Id": self.id, @@ -474,34 +478,33 @@ class ManagedBlockchainNode(BaseModel): "Status": self.status, "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), } - return d - def delete(self): + def delete(self) -> None: self.status = "DELETED" - def update(self, logpublishingconfiguration): + def update(self, logpublishingconfiguration: Dict[str, Any]) -> None: self.logpublishingconfiguration = logpublishingconfiguration class ManagedBlockchainBackend(BaseBackend): - def __init__(self, region_name, account_id): + def __init__(self, region_name: str, account_id: str): super().__init__(region_name, account_id) - self.networks = {} - self.members = {} - self.proposals = {} - self.invitations = {} - self.nodes = {} + self.networks: Dict[str, ManagedBlockchainNetwork] = {} + self.members: Dict[str, ManagedBlockchainMember] = {} + self.proposals: Dict[str, ManagedBlockchainProposal] = {} + self.invitations: Dict[str, ManagedBlockchainInvitation] = {} + self.nodes: Dict[str, ManagedBlockchainNode] = {} def create_network( self, - name, - framework, - frameworkversion, - frameworkconfiguration, - voting_policy, - member_configuration, - description=None, - ): + name: str, + framework: str, + frameworkversion: str, + frameworkconfiguration: Dict[str, Any], + voting_policy: Dict[str, Any], + member_configuration: Dict[str, Any], + description: Optional[str] = None, + ) -> Dict[str, str]: # Check framework if framework not in FRAMEWORKS: raise BadRequestException("CreateNetwork", "Invalid request body") @@ -542,20 +545,25 @@ class ManagedBlockchainBackend(BaseBackend): ) # Return the network and member ID - d = {"NetworkId": network_id, "MemberId": member_id} - return d + return {"NetworkId": network_id, "MemberId": member_id} - def list_networks(self): - return self.networks.values() + def list_networks(self) -> List[ManagedBlockchainNetwork]: + return list(self.networks.values()) - def get_network(self, network_id): + def get_network(self, network_id: str) -> ManagedBlockchainNetwork: if network_id not in self.networks: raise ResourceNotFoundException( "GetNetwork", f"Network {network_id} not found." ) - return self.networks.get(network_id) + return self.networks[network_id] - def create_proposal(self, networkid, memberid, actions, description=None): + def create_proposal( + self, + networkid: str, + memberid: str, + actions: Dict[str, Any], + description: Optional[str] = None, + ) -> Dict[str, str]: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -593,24 +601,21 @@ class ManagedBlockchainBackend(BaseBackend): proposal_id=proposal_id, networkid=networkid, memberid=memberid, - membername=self.members.get(memberid).name, + membername=self.members[memberid].name, numofmembers=number_of_members_in_network(self.members, networkid), actions=actions, - network_expirtation=self.networks.get(networkid).vote_pol_proposal_duration, - network_threshold=self.networks.get( + network_expiration=self.networks[networkid].vote_pol_proposal_duration, + network_threshold=self.networks[networkid].vote_pol_threshold_percentage, + network_threshold_comp=self.networks[ networkid - ).vote_pol_threshold_percentage, - network_threshold_comp=self.networks.get( - networkid - ).vote_pol_threshold_comparator, + ].vote_pol_threshold_comparator, description=description, ) # Return the proposal ID - d = {"ProposalId": proposal_id} - return d + return {"ProposalId": proposal_id} - def list_proposals(self, networkid): + def list_proposals(self, networkid: str) -> List[ManagedBlockchainProposal]: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -619,13 +624,15 @@ class ManagedBlockchainBackend(BaseBackend): proposalsfornetwork = [] for proposal_id in self.proposals: - if self.proposals.get(proposal_id).network_id == networkid: + if self.proposals[proposal_id].network_id == networkid: # See if any are expired - self.proposals.get(proposal_id).check_to_expire_proposal() + self.proposals[proposal_id].check_to_expire_proposal() proposalsfornetwork.append(self.proposals[proposal_id]) return proposalsfornetwork - def get_proposal(self, networkid, proposalid): + def get_proposal( + self, networkid: str, proposalid: str + ) -> ManagedBlockchainProposal: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -638,10 +645,12 @@ class ManagedBlockchainBackend(BaseBackend): ) # See if it needs to be set to expipred - self.proposals.get(proposalid).check_to_expire_proposal() - return self.proposals.get(proposalid) + self.proposals[proposalid].check_to_expire_proposal() + return self.proposals[proposalid] - def vote_on_proposal(self, networkid, proposalid, votermemberid, vote): + def vote_on_proposal( + self, networkid: str, proposalid: str, votermemberid: str, vote: str + ) -> None: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -662,60 +671,56 @@ class ManagedBlockchainBackend(BaseBackend): raise BadRequestException("VoteOnProposal", "Invalid request body") # See if it needs to be set to expipred - self.proposals.get(proposalid).check_to_expire_proposal() + self.proposals[proposalid].check_to_expire_proposal() # Exception if EXPIRED - if self.proposals.get(proposalid).proposal_status == "EXPIRED": + if self.proposals[proposalid].proposal_status == "EXPIRED": raise InvalidRequestException( "VoteOnProposal", f"Proposal {proposalid} is expired and you cannot vote on it.", ) # Check if IN_PROGRESS - if self.proposals.get(proposalid).proposal_status != "IN_PROGRESS": + if self.proposals[proposalid].proposal_status != "IN_PROGRESS": raise InvalidRequestException( "VoteOnProposal", - f"Proposal {proposalid} has status {self.proposals.get(proposalid).proposal_status} and you cannot vote on it.", + f"Proposal {proposalid} has status {self.proposals[proposalid].proposal_status} and you cannot vote on it.", ) # Check to see if this member already voted - if votermemberid in self.proposals.get(proposalid).proposal_votes: + if votermemberid in self.proposals[proposalid].proposal_votes: raise ResourceAlreadyExistsException( "VoteOnProposal", f"Member {votermemberid} has already voted on proposal {proposalid}.", ) # Cast vote - self.proposals.get(proposalid).set_vote( - votermemberid, self.members.get(votermemberid).name, vote.upper() + self.proposals[proposalid].set_vote( + votermemberid, self.members[votermemberid].name, vote.upper() ) - if self.proposals.get(proposalid).proposal_status == "APPROVED": + if self.proposals[proposalid].proposal_status == "APPROVED": # Generate invitations - for _ in self.proposals.get(proposalid).proposal_actions("Invitations"): + for _ in self.proposals[proposalid].proposal_actions("Invitations"): invitation_id = get_invitation_id() self.invitations[invitation_id] = ManagedBlockchainInvitation( invitation_id=invitation_id, networkid=networkid, - networkname=self.networks.get(networkid).network_name, - networkframework=self.networks.get(networkid).network_framework, - networkframeworkversion=self.networks.get( + networkname=self.networks[networkid].network_name, + networkframework=self.networks[networkid].network_framework, + networkframeworkversion=self.networks[ networkid - ).network_framework_version, - networkcreationdate=self.networks.get( - networkid - ).network_creationdate, + ].network_framework_version, + networkcreationdate=self.networks[networkid].network_creationdate, region=self.region_name, - networkdescription=self.networks.get(networkid).network_description, + networkdescription=self.networks[networkid].network_description, ) # Delete members - for propmember in self.proposals.get(proposalid).proposal_actions( - "Removals" - ): + for propmember in self.proposals[proposalid].proposal_actions("Removals"): self.delete_member(networkid, propmember["MemberId"]) - def list_proposal_votes(self, networkid, proposalid): + def list_proposal_votes(self, networkid: str, proposalid: str) -> List[str]: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -729,25 +734,25 @@ class ManagedBlockchainBackend(BaseBackend): # Output the vote summaries proposalvotesfornetwork = [] - for proposal_id in self.proposals: - if self.proposals.get(proposal_id).network_id == networkid: - for pvmemberid in self.proposals.get(proposal_id).proposal_votes: - proposalvotesfornetwork.append( - self.proposals.get(proposal_id).proposal_votes[pvmemberid] - ) + for proposal in self.proposals.values(): + if proposal.network_id == networkid: + for proposal_vote in proposal.proposal_votes.values(): + proposalvotesfornetwork.append(proposal_vote) return proposalvotesfornetwork - def list_invitations(self): - return self.invitations.values() + def list_invitations(self) -> List[ManagedBlockchainInvitation]: + return list(self.invitations.values()) - def reject_invitation(self, invitationid): + def reject_invitation(self, invitationid: str) -> None: if invitationid not in self.invitations: raise ResourceNotFoundException( "RejectInvitation", f"InvitationId {invitationid} not found." ) - self.invitations.get(invitationid).reject_invitation() + self.invitations[invitationid].reject_invitation() - def create_member(self, invitationid, networkid, member_configuration): + def create_member( + self, invitationid: str, networkid: str, member_configuration: Dict[str, Any] + ) -> Dict[str, str]: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -759,7 +764,7 @@ class ManagedBlockchainBackend(BaseBackend): "CreateMember", f"Invitation {invitationid} not valid" ) - if self.invitations.get(invitationid).invitation_status != "PENDING": + if self.invitations[invitationid].invitation_status != "PENDING": raise InvalidRequestException( "CreateMember", f"Invitation {invitationid} not valid" ) @@ -772,7 +777,7 @@ class ManagedBlockchainBackend(BaseBackend): f"Member name {member_configuration['Name']} already exists in network {networkid}.", ) - networkedition = self.networks.get(networkid).network_edition + networkedition = self.networks[networkid].network_edition if ( number_of_members_in_network(self.members, networkid) >= EDITIONS[networkedition]["MaxMembers"] @@ -797,13 +802,12 @@ class ManagedBlockchainBackend(BaseBackend): ) # Accept the invitaiton - self.invitations.get(invitationid).accept_invitation() + self.invitations[invitationid].accept_invitation() # Return the member ID - d = {"MemberId": member_id} - return d + return {"MemberId": member_id} - def list_members(self, networkid): + def list_members(self, networkid: str) -> List[ManagedBlockchainMember]: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -811,12 +815,12 @@ class ManagedBlockchainBackend(BaseBackend): ) membersfornetwork = [] - for member_id in self.members: - if self.members.get(member_id).network_id == networkid: - membersfornetwork.append(self.members[member_id]) + for member in self.members.values(): + if member.network_id == networkid: + membersfornetwork.append(member) return membersfornetwork - def get_member(self, networkid, memberid): + def get_member(self, networkid: str, memberid: str) -> ManagedBlockchainMember: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -829,14 +833,14 @@ class ManagedBlockchainBackend(BaseBackend): ) # Cannot get a member than has been deleted (it does show up in the list) - if self.members.get(memberid).member_status == "DELETED": + if self.members[memberid].member_status == "DELETED": raise ResourceNotFoundException( "GetMember", f"Member {memberid} not found." ) - return self.members.get(memberid) + return self.members[memberid] - def delete_member(self, networkid, memberid): + def delete_member(self, networkid: str, memberid: str) -> None: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -848,19 +852,16 @@ class ManagedBlockchainBackend(BaseBackend): "DeleteMember", f"Member {memberid} not found." ) - self.members.get(memberid).delete() + self.members[memberid].delete() # Is this the last member in the network? (all set to DELETED) if number_of_members_in_network( self.members, networkid, member_status="DELETED" ) == len(self.members): # Set network status to DELETED for all invitations - for invitation_id in self.invitations: - if ( - self.invitations.get(invitation_id).invitation_networkid - == networkid - ): - self.invitations.get(invitation_id).set_network_status("DELETED") + for invitation in self.invitations.values(): + if invitation.invitation_networkid == networkid: + invitation.set_network_status("DELETED") # Remove network del self.networks[networkid] @@ -869,7 +870,9 @@ class ManagedBlockchainBackend(BaseBackend): for nodeid in nodes_in_member(self.nodes, memberid): del self.nodes[nodeid] - def update_member(self, networkid, memberid, logpublishingconfiguration): + def update_member( + self, networkid: str, memberid: str, logpublishingconfiguration: Dict[str, Any] + ) -> None: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -881,16 +884,16 @@ class ManagedBlockchainBackend(BaseBackend): "UpdateMember", f"Member {memberid} not found." ) - self.members.get(memberid).update(logpublishingconfiguration) + self.members[memberid].update(logpublishingconfiguration) def create_node( self, - networkid, - memberid, - availabilityzone, - instancetype, - logpublishingconfiguration, - ): + networkid: str, + memberid: str, + availabilityzone: str, + instancetype: str, + logpublishingconfiguration: Dict[str, Any], + ) -> Dict[str, str]: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -902,7 +905,7 @@ class ManagedBlockchainBackend(BaseBackend): "CreateNode", f"Member {memberid} not found." ) - networkedition = self.networks.get(networkid).network_edition + networkedition = self.networks[networkid].network_edition if ( number_of_nodes_in_member(self.nodes, memberid) >= EDITIONS[networkedition]["MaxNodesPerMember"] @@ -953,10 +956,11 @@ class ManagedBlockchainBackend(BaseBackend): ) # Return the node ID - d = {"NodeId": node_id} - return d + return {"NodeId": node_id} - def list_nodes(self, networkid, memberid, status=None): + def list_nodes( + self, networkid: str, memberid: str, status: Optional[str] = None + ) -> List[ManagedBlockchainNode]: if networkid not in self.networks: raise ResourceNotFoundException( "ListNodes", f"Network {networkid} not found." @@ -968,20 +972,22 @@ class ManagedBlockchainBackend(BaseBackend): ) # If member is deleted, cannot list nodes - if self.members.get(memberid).member_status == "DELETED": + if self.members[memberid].member_status == "DELETED": raise ResourceNotFoundException( "ListNodes", f"Member {memberid} not found." ) nodesformember = [] - for node_id in self.nodes: - if self.nodes.get(node_id).member_id == memberid and ( - status is None or self.nodes.get(node_id).node_status == status + for node in self.nodes.values(): + if node.member_id == memberid and ( + status is None or node.node_status == status ): - nodesformember.append(self.nodes[node_id]) + nodesformember.append(node) return nodesformember - def get_node(self, networkid, memberid, nodeid): + def get_node( + self, networkid: str, memberid: str, nodeid: str + ) -> ManagedBlockchainNode: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -995,12 +1001,12 @@ class ManagedBlockchainBackend(BaseBackend): raise ResourceNotFoundException("GetNode", f"Node {nodeid} not found.") # Cannot get a node than has been deleted (it does show up in the list) - if self.nodes.get(nodeid).node_status == "DELETED": + if self.nodes[nodeid].node_status == "DELETED": raise ResourceNotFoundException("GetNode", f"Node {nodeid} not found.") - return self.nodes.get(nodeid) + return self.nodes[nodeid] - def delete_node(self, networkid, memberid, nodeid): + def delete_node(self, networkid: str, memberid: str, nodeid: str) -> None: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -1015,9 +1021,15 @@ class ManagedBlockchainBackend(BaseBackend): if nodeid not in self.nodes: raise ResourceNotFoundException("DeleteNode", f"Node {nodeid} not found.") - self.nodes.get(nodeid).delete() + self.nodes[nodeid].delete() - def update_node(self, networkid, memberid, nodeid, logpublishingconfiguration): + def update_node( + self, + networkid: str, + memberid: str, + nodeid: str, + logpublishingconfiguration: Dict[str, Any], + ) -> None: # Check if network exists if networkid not in self.networks: raise ResourceNotFoundException( @@ -1032,7 +1044,7 @@ class ManagedBlockchainBackend(BaseBackend): if nodeid not in self.nodes: raise ResourceNotFoundException("UpdateNode", f"Node {nodeid} not found.") - self.nodes.get(nodeid).update(logpublishingconfiguration) + self.nodes[nodeid].update(logpublishingconfiguration) managedblockchain_backends = BackendDict(ManagedBlockchainBackend, "managedblockchain") diff --git a/moto/managedblockchain/responses.py b/moto/managedblockchain/responses.py index e1ed7f724..46e6e6ecd 100644 --- a/moto/managedblockchain/responses.py +++ b/moto/managedblockchain/responses.py @@ -1,9 +1,11 @@ import json +from typing import Any, Dict, Optional from urllib.parse import urlparse, parse_qs +from moto.core.common_types import TYPE_RESPONSE from moto.core.responses import BaseResponse from .exceptions import exception_handler -from .models import managedblockchain_backends +from .models import managedblockchain_backends, ManagedBlockchainBackend from .utils import ( networkid_from_managedblockchain_url, proposalid_from_managedblockchain_url, @@ -14,19 +16,19 @@ from .utils import ( class ManagedBlockchainResponse(BaseResponse): - def __init__(self): + def __init__(self) -> None: super().__init__(service_name="managedblockchain") @property - def backend(self): + def backend(self) -> ManagedBlockchainBackend: return managedblockchain_backends[self.current_account][self.region] @exception_handler - def network_response(self, request, full_url, headers): + def network_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._network_response(request, headers) - def _network_response(self, request, headers): + def _network_response(self, request: Any, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method if method == "GET": return self._all_networks_response(headers) @@ -34,7 +36,7 @@ class ManagedBlockchainResponse(BaseResponse): json_body = json.loads(self.body) return self._network_response_post(json_body, headers) - def _all_networks_response(self, headers): + def _all_networks_response(self, headers: Any) -> TYPE_RESPONSE: mbcnetworks = self.backend.list_networks() response = json.dumps( {"Networks": [mbcnetwork.to_dict() for mbcnetwork in mbcnetworks]} @@ -42,7 +44,9 @@ class ManagedBlockchainResponse(BaseResponse): headers["content-type"] = "application/json" return 200, headers, response - def _network_response_post(self, json_body, headers): + def _network_response_post( + self, json_body: Dict[str, Any], headers: Any + ) -> TYPE_RESPONSE: name = json_body["Name"] framework = json_body["Framework"] frameworkversion = json_body["FrameworkVersion"] @@ -65,29 +69,29 @@ class ManagedBlockchainResponse(BaseResponse): return 200, headers, json.dumps(response) @exception_handler - def networkid_response(self, request, full_url, headers): + def networkid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._networkid_response(request, full_url, headers) - def _networkid_response(self, request, full_url, headers): + def _networkid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method if method == "GET": network_id = networkid_from_managedblockchain_url(full_url) return self._networkid_response_get(network_id, headers) - def _networkid_response_get(self, network_id, headers): + def _networkid_response_get(self, network_id: str, headers: Any) -> TYPE_RESPONSE: mbcnetwork = self.backend.get_network(network_id) response = json.dumps({"Network": mbcnetwork.get_format()}) headers["content-type"] = "application/json" return 200, headers, response @exception_handler - def proposal_response(self, request, full_url, headers): + def proposal_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._proposal_response(request, full_url, headers) - def _proposal_response(self, request, full_url, headers): + def _proposal_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method network_id = networkid_from_managedblockchain_url(full_url) if method == "GET": @@ -96,7 +100,7 @@ class ManagedBlockchainResponse(BaseResponse): json_body = json.loads(self.body) return self._proposal_response_post(network_id, json_body, headers) - def _all_proposals_response(self, network_id, headers): + def _all_proposals_response(self, network_id: str, headers: Any) -> TYPE_RESPONSE: proposals = self.backend.list_proposals(network_id) response = json.dumps( {"Proposals": [proposal.to_dict() for proposal in proposals]} @@ -104,7 +108,9 @@ class ManagedBlockchainResponse(BaseResponse): headers["content-type"] = "application/json" return 200, headers, response - def _proposal_response_post(self, network_id, json_body, headers): + def _proposal_response_post( + self, network_id: str, json_body: Dict[str, Any], headers: Any + ) -> TYPE_RESPONSE: memberid = json_body["MemberId"] actions = json_body["Actions"] @@ -117,29 +123,31 @@ class ManagedBlockchainResponse(BaseResponse): return 200, headers, json.dumps(response) @exception_handler - def proposalid_response(self, request, full_url, headers): + def proposalid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._proposalid_response(request, full_url, headers) - def _proposalid_response(self, request, full_url, headers): + def _proposalid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method network_id = networkid_from_managedblockchain_url(full_url) if method == "GET": proposal_id = proposalid_from_managedblockchain_url(full_url) return self._proposalid_response_get(network_id, proposal_id, headers) - def _proposalid_response_get(self, network_id, proposal_id, headers): + def _proposalid_response_get( + self, network_id: str, proposal_id: str, headers: Any + ) -> TYPE_RESPONSE: proposal = self.backend.get_proposal(network_id, proposal_id) response = json.dumps({"Proposal": proposal.get_format()}) headers["content-type"] = "application/json" return 200, headers, response @exception_handler - def proposal_votes_response(self, request, full_url, headers): + def proposal_votes_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._proposal_votes_response(request, full_url, headers) - def _proposal_votes_response(self, request, full_url, headers): + def _proposal_votes_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method network_id = networkid_from_managedblockchain_url(full_url) proposal_id = proposalid_from_managedblockchain_url(full_url) @@ -151,15 +159,17 @@ class ManagedBlockchainResponse(BaseResponse): network_id, proposal_id, json_body, headers ) - def _all_proposal_votes_response(self, network_id, proposal_id, headers): + def _all_proposal_votes_response( + self, network_id: str, proposal_id: str, headers: Any + ) -> TYPE_RESPONSE: proposalvotes = self.backend.list_proposal_votes(network_id, proposal_id) response = json.dumps({"ProposalVotes": proposalvotes}) headers["content-type"] = "application/json" return 200, headers, response def _proposal_votes_response_post( - self, network_id, proposal_id, json_body, headers - ): + self, network_id: str, proposal_id: str, json_body: Dict[str, Any], headers: Any + ) -> TYPE_RESPONSE: votermemberid = json_body["VoterMemberId"] vote = json_body["Vote"] @@ -167,16 +177,16 @@ class ManagedBlockchainResponse(BaseResponse): return 200, headers, "" @exception_handler - def invitation_response(self, request, full_url, headers): + def invitation_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._invitation_response(request, headers) - def _invitation_response(self, request, headers): + def _invitation_response(self, request: Any, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method if method == "GET": return self._all_invitation_response(headers) - def _all_invitation_response(self, headers): + def _all_invitation_response(self, headers: Any) -> TYPE_RESPONSE: invitations = self.backend.list_invitations() response = json.dumps( {"Invitations": [invitation.to_dict() for invitation in invitations]} @@ -185,27 +195,29 @@ class ManagedBlockchainResponse(BaseResponse): return 200, headers, response @exception_handler - def invitationid_response(self, request, full_url, headers): + def invitationid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._invitationid_response(request, full_url, headers) - def _invitationid_response(self, request, full_url, headers): + def _invitationid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method if method == "DELETE": invitation_id = invitationid_from_managedblockchain_url(full_url) return self._invitationid_response_delete(invitation_id, headers) - def _invitationid_response_delete(self, invitation_id, headers): + def _invitationid_response_delete( + self, invitation_id: str, headers: Any + ) -> TYPE_RESPONSE: self.backend.reject_invitation(invitation_id) headers["content-type"] = "application/json" return 200, headers, "" @exception_handler - def member_response(self, request, full_url, headers): + def member_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._member_response(request, full_url, headers) - def _member_response(self, request, full_url, headers): + def _member_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method network_id = networkid_from_managedblockchain_url(full_url) if method == "GET": @@ -214,13 +226,15 @@ class ManagedBlockchainResponse(BaseResponse): json_body = json.loads(self.body) return self._member_response_post(network_id, json_body, headers) - def _all_members_response(self, network_id, headers): + def _all_members_response(self, network_id: str, headers: Any) -> TYPE_RESPONSE: members = self.backend.list_members(network_id) response = json.dumps({"Members": [member.to_dict() for member in members]}) headers["content-type"] = "application/json" return 200, headers, response - def _member_response_post(self, network_id, json_body, headers): + def _member_response_post( + self, network_id: str, json_body: Dict[str, Any], headers: Any + ) -> TYPE_RESPONSE: invitationid = json_body["InvitationId"] member_configuration = json_body["MemberConfiguration"] @@ -230,11 +244,11 @@ class ManagedBlockchainResponse(BaseResponse): return 200, headers, json.dumps(response) @exception_handler - def memberid_response(self, request, full_url, headers): + def memberid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._memberid_response(request, full_url, headers) - def _memberid_response(self, request, full_url, headers): + def _memberid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method network_id = networkid_from_managedblockchain_url(full_url) member_id = memberid_from_managedblockchain_request(full_url, self.body) @@ -248,28 +262,34 @@ class ManagedBlockchainResponse(BaseResponse): elif method == "DELETE": return self._memberid_response_delete(network_id, member_id, headers) - def _memberid_response_get(self, network_id, member_id, headers): + def _memberid_response_get( + self, network_id: str, member_id: str, headers: Dict[str, Any] + ) -> TYPE_RESPONSE: member = self.backend.get_member(network_id, member_id) response = json.dumps({"Member": member.get_format()}) headers["content-type"] = "application/json" return 200, headers, response - def _memberid_response_patch(self, network_id, member_id, json_body, headers): + def _memberid_response_patch( + self, network_id: str, member_id: str, json_body: Dict[str, Any], headers: Any + ) -> TYPE_RESPONSE: logpublishingconfiguration = json_body["LogPublishingConfiguration"] self.backend.update_member(network_id, member_id, logpublishingconfiguration) return 200, headers, "" - def _memberid_response_delete(self, network_id, member_id, headers): + def _memberid_response_delete( + self, network_id: str, member_id: str, headers: Any + ) -> TYPE_RESPONSE: self.backend.delete_member(network_id, member_id) headers["content-type"] = "application/json" return 200, headers, "" @exception_handler - def node_response(self, request, full_url, headers): + def node_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._node_response(request, full_url, headers) - def _node_response(self, request, full_url, headers): + def _node_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method parsed_url = urlparse(full_url) querystring = parse_qs(parsed_url.query, keep_blank_values=True) @@ -284,13 +304,17 @@ class ManagedBlockchainResponse(BaseResponse): json_body = json.loads(self.body) return self._node_response_post(network_id, member_id, json_body, headers) - def _all_nodes_response(self, network_id, member_id, status, headers): + def _all_nodes_response( + self, network_id: str, member_id: str, status: Optional[str], headers: Any + ) -> TYPE_RESPONSE: nodes = self.backend.list_nodes(network_id, member_id, status) response = json.dumps({"Nodes": [node.to_dict() for node in nodes]}) headers["content-type"] = "application/json" return 200, headers, response - def _node_response_post(self, network_id, member_id, json_body, headers): + def _node_response_post( + self, network_id: str, member_id: str, json_body: Dict[str, Any], headers: Any + ) -> TYPE_RESPONSE: instancetype = json_body["NodeConfiguration"]["InstanceType"] availabilityzone = json_body["NodeConfiguration"]["AvailabilityZone"] logpublishingconfiguration = json_body["NodeConfiguration"][ @@ -307,11 +331,11 @@ class ManagedBlockchainResponse(BaseResponse): return 200, headers, json.dumps(response) @exception_handler - def nodeid_response(self, request, full_url, headers): + def nodeid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] self.setup_class(request, full_url, headers) return self._nodeid_response(request, full_url, headers) - def _nodeid_response(self, request, full_url, headers): + def _nodeid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] method = request.method network_id = networkid_from_managedblockchain_url(full_url) member_id = memberid_from_managedblockchain_request(full_url, self.body) @@ -326,22 +350,31 @@ class ManagedBlockchainResponse(BaseResponse): elif method == "DELETE": return self._nodeid_response_delete(network_id, member_id, node_id, headers) - def _nodeid_response_get(self, network_id, member_id, node_id, headers): + def _nodeid_response_get( + self, network_id: str, member_id: str, node_id: str, headers: Any + ) -> TYPE_RESPONSE: node = self.backend.get_node(network_id, member_id, node_id) response = json.dumps({"Node": node.get_format()}) headers["content-type"] = "application/json" return 200, headers, response def _nodeid_response_patch( - self, network_id, member_id, node_id, json_body, headers - ): + self, + network_id: str, + member_id: str, + node_id: str, + json_body: Dict[str, Any], + headers: Any, + ) -> TYPE_RESPONSE: logpublishingconfiguration = json_body self.backend.update_node( network_id, member_id, node_id, logpublishingconfiguration ) return 200, headers, "" - def _nodeid_response_delete(self, network_id, member_id, node_id, headers): + def _nodeid_response_delete( + self, network_id: str, member_id: str, node_id: str, headers: Any + ) -> TYPE_RESPONSE: self.backend.delete_node(network_id, member_id, node_id) headers["content-type"] = "application/json" return 200, headers, "" diff --git a/moto/managedblockchain/utils.py b/moto/managedblockchain/utils.py index 710939c31..9db9378e5 100644 --- a/moto/managedblockchain/utils.py +++ b/moto/managedblockchain/utils.py @@ -2,24 +2,25 @@ import json import re import string from moto.moto_api._internal import mock_random as random +from typing import Any, Dict, List, Optional from urllib.parse import parse_qs, urlparse -def networkid_from_managedblockchain_url(full_url): +def networkid_from_managedblockchain_url(full_url: str) -> str: id_search = re.search(r"\/n-[A-Z0-9]{26}", full_url, re.IGNORECASE) return_id = None if id_search: return_id = id_search.group(0).replace("/", "") - return return_id + return return_id # type: ignore[return-value] -def get_network_id(): +def get_network_id() -> str: return "n-" + "".join( random.choice(string.ascii_uppercase + string.digits) for _ in range(26) ) -def memberid_from_managedblockchain_request(full_url, body): +def memberid_from_managedblockchain_request(full_url: str, body: Dict[str, Any]) -> str: id_search = re.search(r"\/m-[A-Z0-9]{26}", full_url, re.IGNORECASE) return_id = None if id_search: @@ -29,72 +30,71 @@ def memberid_from_managedblockchain_request(full_url, body): parsed_url = urlparse(full_url) qs = parse_qs(parsed_url.query) if "memberId" in qs: - return_id = qs.get("memberId")[0] + return_id = qs.get("memberId")[0] # type: ignore elif body: - body = json.loads(body) + body = json.loads(body) # type: ignore return_id = body["MemberId"] - return return_id + return return_id # type: ignore[return-value] -def get_member_id(): +def get_member_id() -> str: return "m-" + "".join( random.choice(string.ascii_uppercase + string.digits) for _ in range(26) ) -def proposalid_from_managedblockchain_url(full_url): +def proposalid_from_managedblockchain_url(full_url: str) -> str: id_search = re.search(r"\/p-[A-Z0-9]{26}", full_url, re.IGNORECASE) return_id = None if id_search: return_id = id_search.group(0).replace("/", "") - return return_id + return return_id # type: ignore[return-value] -def get_proposal_id(): +def get_proposal_id() -> str: return "p-" + "".join( random.choice(string.ascii_uppercase + string.digits) for _ in range(26) ) -def invitationid_from_managedblockchain_url(full_url): +def invitationid_from_managedblockchain_url(full_url: str) -> str: id_search = re.search(r"\/in-[A-Z0-9]{26}", full_url, re.IGNORECASE) return_id = None if id_search: return_id = id_search.group(0).replace("/", "") - return return_id + return return_id # type: ignore[return-value] -def get_invitation_id(): +def get_invitation_id() -> str: return "in-" + "".join( random.choice(string.ascii_uppercase + string.digits) for _ in range(26) ) -def member_name_exist_in_network(members, networkid, membername): - membernamexists = False - for member_id in members: - if members.get(member_id).network_id == networkid: - if members.get(member_id).name == membername: - membernamexists = True - break - return membernamexists +def member_name_exist_in_network( + members: Dict[str, Any], networkid: str, membername: str +) -> bool: + for member in members.values(): + if member.network_id == networkid: + if member.name == membername: + return True + return False -def number_of_members_in_network(members, networkid, member_status=None): +def number_of_members_in_network( + members: Dict[str, Any], networkid: str, member_status: Optional[str] = None +) -> int: return len( [ - membid - for membid in members - if members.get(membid).network_id == networkid - and ( - member_status is None - or members.get(membid).member_status == member_status - ) + member + for member in members.values() + if member.network_id == networkid + and (member_status is None or member.member_status == member_status) ] ) -def admin_password_ok(password): +def admin_password_ok(password: str) -> bool: if not re.search("[a-z]", password): return False elif not re.search("[A-Z]", password): @@ -107,30 +107,32 @@ def admin_password_ok(password): return True -def nodeid_from_managedblockchain_url(full_url): +def nodeid_from_managedblockchain_url(full_url: str) -> str: id_search = re.search(r"\/nd-[A-Z0-9]{26}", full_url, re.IGNORECASE) return_id = None if id_search: return_id = id_search.group(0).replace("/", "") - return return_id + return return_id # type: ignore[return-value] -def get_node_id(): +def get_node_id() -> str: return "nd-" + "".join( random.choice(string.ascii_uppercase + string.digits) for _ in range(26) ) -def number_of_nodes_in_member(nodes, memberid, node_status=None): +def number_of_nodes_in_member( + nodes: Dict[str, Any], memberid: str, node_status: Optional[str] = None +) -> int: return len( [ - nodid - for nodid in nodes - if nodes.get(nodid).member_id == memberid - and (node_status is None or nodes.get(nodid).node_status == node_status) + node + for node in nodes.values() + if node.member_id == memberid + and (node_status is None or node.node_status == node_status) ] ) -def nodes_in_member(nodes, memberid): - return [nodid for nodid in nodes if nodes.get(nodid).member_id == memberid] +def nodes_in_member(nodes: Dict[str, Any], memberid: str) -> List[str]: + return [nodid for nodid in nodes if nodes[nodid].member_id == memberid] diff --git a/setup.cfg b/setup.cfg index 7d7508858..48b11dbf4 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/moto_api,moto/neptune,moto/opensearch +files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/managedblockchain,moto/moto_api,moto/neptune,moto/opensearch show_column_numbers=True show_error_codes = True disable_error_code=abstract