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

Techdebt: Replace string-format with f-strings (for m* dirs) (#5686)

Browse Source
This commit is contained in:
Bert Blommers 2022-11-19 23:12:31 -01:00 committed by GitHub
parent 2483111f71
commit 677457c1b9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 97 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
moto/managedblockchain/exceptions.py
View File

@ -35,9 +35,7 @@ class BadRequestException(ManagedBlockchainClientError):
def __init__(self, pretty_called_method, operation_error):
super().__init__(
"BadRequestException",
"An error occurred (BadRequestException) when calling the {0} operation: {1}".format(
pretty_called_method, operation_error
),
f"An error occurred (BadRequestException) when calling the {pretty_called_method} operation: {operation_error}",
)
@ -45,9 +43,7 @@ class InvalidRequestException(ManagedBlockchainClientError):
def __init__(self, pretty_called_method, operation_error):
super().__init__(
"InvalidRequestException",
"An error occurred (InvalidRequestException) when calling the {0} operation: {1}".format(
pretty_called_method, operation_error
),
f"An error occurred (InvalidRequestException) when calling the {pretty_called_method} operation: {operation_error}",
)
@ -56,9 +52,7 @@ class ResourceNotFoundException(ManagedBlockchainClientError):
self.code = 404
super().__init__(
"ResourceNotFoundException",
"An error occurred (ResourceNotFoundException) when calling the {0} operation: {1}".format(
pretty_called_method, operation_error
),
f"An error occurred (ResourceNotFoundException) when calling the {pretty_called_method} operation: {operation_error}",
)
@ -67,9 +61,7 @@ class ResourceAlreadyExistsException(ManagedBlockchainClientError):
self.code = 409
super().__init__(
"ResourceAlreadyExistsException",
"An error occurred (ResourceAlreadyExistsException) when calling the {0} operation: {1}".format(
pretty_called_method, operation_error
),
f"An error occurred (ResourceAlreadyExistsException) when calling the {pretty_called_method} operation: {operation_error}",
)
@ -78,7 +70,5 @@ class ResourceLimitExceededException(ManagedBlockchainClientError):
self.code = 429
super().__init__(
"ResourceLimitExceededException",
"An error occurred (ResourceLimitExceededException) when calling the {0} operation: {1}".format(
pretty_called_method, operation_error
),
f"An error occurred (ResourceLimitExceededException) when calling the {pretty_called_method} operation: {operation_error}",
)

151
moto/managedblockchain/models.py
View File

@ -128,15 +128,13 @@ class ManagedBlockchainNetwork(BaseModel):
# Format for get_network
frameworkattributes = {
"Fabric": {
"OrderingServiceEndpoint": "orderer.{0}.managedblockchain.{1}.amazonaws.com:30001".format(
self.id.lower(), self.region
),
"OrderingServiceEndpoint": f"orderer.{self.id.lower()}.managedblockchain.{self.region}.amazonaws.com:30001",
"Edition": self.frameworkconfiguration["Fabric"]["Edition"],
}
}
vpcendpointname = "com.amazonaws.{0}.managedblockchain.{1}".format(
self.region, self.id.lower()
vpcendpointname = (
f"com.amazonaws.{self.region}.managedblockchain.{self.id.lower()}"
)
d = {
@ -390,9 +388,7 @@ class ManagedBlockchainMember(BaseModel):
"AdminUsername": self.member_configuration["FrameworkConfiguration"][
"Fabric"
]["AdminUsername"],
"CaEndpoint": "ca.{0}.{1}.managedblockchain.{2}.amazonaws.com:30002".format(
self.id.lower(), self.networkid.lower(), self.region
),
"CaEndpoint": f"ca.{self.id.lower()}.{self.networkid.lower()}.managedblockchain.{self.region}.amazonaws.com:30002",
}
}
@ -464,18 +460,8 @@ class ManagedBlockchainNode(BaseModel):
# Format for get_node
frameworkattributes = {
"Fabric": {
"PeerEndpoint": "{0}.{1}.{2}.managedblockchain.{3}.amazonaws.com:30003".format(
self.id.lower(),
self.networkid.lower(),
self.memberid.lower(),
self.region,
),
"PeerEventEndpoint": "{0}.{1}.{2}.managedblockchain.{3}.amazonaws.com:30004".format(
self.id.lower(),
self.networkid.lower(),
self.memberid.lower(),
self.region,
),
"PeerEndpoint": f"{self.id.lower()}.{self.networkid.lower()}.{self.memberid.lower()}.managedblockchain.{self.region}.amazonaws.com:30003",
"PeerEventEndpoint": f"{self.id.lower()}.{self.networkid.lower()}.{self.memberid.lower()}.managedblockchain.{self.region}.amazonaws.com:30004",
}
}
@ -526,9 +512,7 @@ class ManagedBlockchainBackend(BaseBackend):
if frameworkversion not in FRAMEWORKVERSIONS:
raise BadRequestException(
"CreateNetwork",
"Invalid version {0} requested for framework HYPERLEDGER_FABRIC".format(
frameworkversion
),
f"Invalid version {frameworkversion} requested for framework HYPERLEDGER_FABRIC",
)
# Check edition
@ -569,7 +553,7 @@ class ManagedBlockchainBackend(BaseBackend):
def get_network(self, network_id):
if network_id not in self.networks:
raise ResourceNotFoundException(
"GetNetwork", "Network {0} not found.".format(network_id)
"GetNetwork", f"Network {network_id} not found."
)
return self.networks.get(network_id)
@ -577,13 +561,13 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"CreateProposal", "Network {0} not found.".format(networkid)
"CreateProposal", f"Network {networkid} not found."
)
# Check if member exists
if memberid not in self.members:
raise ResourceNotFoundException(
"CreateProposal", "Member {0} not found.".format(memberid)
"CreateProposal", f"Member {memberid} not found."
)
# CLI docs say that Invitations and Removals cannot both be passed - but it does
@ -632,7 +616,7 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"ListProposals", "Network {0} not found.".format(networkid)
"ListProposals", f"Network {networkid} not found."
)
proposalsfornetwork = []
@ -647,12 +631,12 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"GetProposal", "Network {0} not found.".format(networkid)
"GetProposal", f"Network {networkid} not found."
)
if proposalid not in self.proposals:
raise ResourceNotFoundException(
"GetProposal", "Proposal {0} not found.".format(proposalid)
"GetProposal", f"Proposal {proposalid} not found."
)
# See if it needs to be set to expipred
@ -663,17 +647,17 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"VoteOnProposal", "Network {0} not found.".format(networkid)
"VoteOnProposal", f"Network {networkid} not found."
)
if proposalid not in self.proposals:
raise ResourceNotFoundException(
"VoteOnProposal", "Proposal {0} not found.".format(proposalid)
"VoteOnProposal", f"Proposal {proposalid} not found."
)
if votermemberid not in self.members:
raise ResourceNotFoundException(
"VoteOnProposal", "Member {0} not found.".format(votermemberid)
"VoteOnProposal", f"Member {votermemberid} not found."
)
if vote.upper() not in VOTEVALUES:
@ -686,25 +670,21 @@ class ManagedBlockchainBackend(BaseBackend):
if self.proposals.get(proposalid).proposal_status == "EXPIRED":
raise InvalidRequestException(
"VoteOnProposal",
"Proposal {0} is expired and you cannot vote on it.".format(proposalid),
f"Proposal {proposalid} is expired and you cannot vote on it.",
)
# Check if IN_PROGRESS
if self.proposals.get(proposalid).proposal_status != "IN_PROGRESS":
raise InvalidRequestException(
"VoteOnProposal",
"Proposal {0} has status {1} and you cannot vote on it.".format(
proposalid, self.proposals.get(proposalid).proposal_status
),
f"Proposal {proposalid} has status {self.proposals.get(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:
raise ResourceAlreadyExistsException(
"VoteOnProposal",
"Member {0} has already voted on proposal {1}.".format(
votermemberid, proposalid
),
f"Member {votermemberid} has already voted on proposal {proposalid}.",
)
# Cast vote
@ -741,12 +721,12 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"ListProposalVotes", "Network {0} not found.".format(networkid)
"ListProposalVotes", f"Network {networkid} not found."
)
if proposalid not in self.proposals:
raise ResourceNotFoundException(
"ListProposalVotes", "Proposal {0} not found.".format(proposalid)
"ListProposalVotes", f"Proposal {proposalid} not found."
)
# Output the vote summaries
@ -765,7 +745,7 @@ class ManagedBlockchainBackend(BaseBackend):
def reject_invitation(self, invitationid):
if invitationid not in self.invitations:
raise ResourceNotFoundException(
"RejectInvitation", "InvitationId {0} not found.".format(invitationid)
"RejectInvitation", f"InvitationId {invitationid} not found."
)
self.invitations.get(invitationid).reject_invitation()
@ -773,30 +753,25 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"CreateMember", "Network {0} not found.".format(networkid)
"CreateMember", f"Network {networkid} not found."
)
if invitationid not in self.invitations:
raise InvalidRequestException(
"CreateMember", "Invitation {0} not valid".format(invitationid)
"CreateMember", f"Invitation {invitationid} not valid"
)
if self.invitations.get(invitationid).invitation_status != "PENDING":
raise InvalidRequestException(
"CreateMember", "Invitation {0} not valid".format(invitationid)
"CreateMember", f"Invitation {invitationid} not valid"
)
if (
member_name_exist_in_network(
if member_name_exist_in_network(
self.members, networkid, member_configuration["Name"]
)
is True
):
raise InvalidRequestException(
"CreateMember",
"Member name {0} already exists in network {1}.".format(
member_configuration["Name"], networkid
),
f"Member name {member_configuration['Name']} already exists in network {networkid}.",
)
networkedition = self.networks.get(networkid).network_edition
@ -806,9 +781,7 @@ class ManagedBlockchainBackend(BaseBackend):
):
raise ResourceLimitExceededException(
"CreateMember",
"You cannot create a member in network {0}.{1} is the maximum number of members allowed in a {2} Edition network.".format(
networkid, EDITIONS[networkedition]["MaxMembers"], networkedition
),
f"You cannot create a member in network {networkid}.{EDITIONS[networkedition]['MaxMembers']} is the maximum number of members allowed in a {networkedition} Edition network.",
)
memberadminpassword = member_configuration["FrameworkConfiguration"]["Fabric"][
@ -836,7 +809,7 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"ListMembers", "Network {0} not found.".format(networkid)
"ListMembers", f"Network {networkid} not found."
)
membersfornetwork = []
@ -849,18 +822,18 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"GetMember", "Network {0} not found.".format(networkid)
"GetMember", f"Network {networkid} not found."
)
if memberid not in self.members:
raise ResourceNotFoundException(
"GetMember", "Member {0} not found.".format(memberid)
"GetMember", f"Member {memberid} not found."
)
# Cannot get a member than has been deleted (it does show up in the list)
if self.members.get(memberid).member_status == "DELETED":
raise ResourceNotFoundException(
"GetMember", "Member {0} not found.".format(memberid)
"GetMember", f"Member {memberid} not found."
)
return self.members.get(memberid)
@ -869,12 +842,12 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"DeleteMember", "Network {0} not found.".format(networkid)
"DeleteMember", f"Network {networkid} not found."
)
if memberid not in self.members:
raise ResourceNotFoundException(
"DeleteMember", "Member {0} not found.".format(memberid)
"DeleteMember", f"Member {memberid} not found."
)
self.members.get(memberid).delete()
@ -902,12 +875,12 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"UpdateMember", "Network {0} not found.".format(networkid)
"UpdateMember", f"Network {networkid} not found."
)
if memberid not in self.members:
raise ResourceNotFoundException(
"UpdateMember", "Member {0} not found.".format(memberid)
"UpdateMember", f"Member {memberid} not found."
)
self.members.get(memberid).update(logpublishingconfiguration)
@ -923,12 +896,12 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"CreateNode", "Network {0} not found.".format(networkid)
"CreateNode", f"Network {networkid} not found."
)
if memberid not in self.members:
raise ResourceNotFoundException(
"CreateNode", "Member {0} not found.".format(memberid)
"CreateNode", f"Member {memberid} not found."
)
networkedition = self.networks.get(networkid).network_edition
@ -938,11 +911,7 @@ class ManagedBlockchainBackend(BaseBackend):
):
raise ResourceLimitExceededException(
"CreateNode",
"Maximum number of nodes exceeded in member {0}. The maximum number of nodes you can have in a member in a {1} Edition network is {2}".format(
memberid,
networkedition,
EDITIONS[networkedition]["MaxNodesPerMember"],
),
f"Maximum number of nodes exceeded in member {memberid}. The maximum number of nodes you can have in a member in a {networkedition} Edition network is {EDITIONS[networkedition]['MaxNodesPerMember']}",
)
# See if the instance family is correct
@ -956,7 +925,7 @@ class ManagedBlockchainBackend(BaseBackend):
if correctinstancefamily is False:
raise InvalidRequestException(
"CreateNode",
"Requested instance {0} isn't supported.".format(instancetype),
f"Requested instance {instancetype} isn't supported.",
)
# Check for specific types for starter
@ -964,9 +933,7 @@ class ManagedBlockchainBackend(BaseBackend):
if instancetype not in EDITIONS["STARTER"]["AllowedNodeInstanceTypes"]:
raise InvalidRequestException(
"CreateNode",
"Instance type {0} is not supported with STARTER Edition networks.".format(
instancetype
),
f"Instance type {instancetype} is not supported with STARTER Edition networks.",
)
# Simple availability zone check
@ -994,18 +961,18 @@ class ManagedBlockchainBackend(BaseBackend):
def list_nodes(self, networkid, memberid, status=None):
if networkid not in self.networks:
raise ResourceNotFoundException(
"ListNodes", "Network {0} not found.".format(networkid)
"ListNodes", f"Network {networkid} not found."
)
if memberid not in self.members:
raise ResourceNotFoundException(
"ListNodes", "Member {0} not found.".format(memberid)
"ListNodes", f"Member {memberid} not found."
)
# If member is deleted, cannot list nodes
if self.members.get(memberid).member_status == "DELETED":
raise ResourceNotFoundException(
"ListNodes", "Member {0} not found.".format(memberid)
"ListNodes", f"Member {memberid} not found."
)
nodesformember = []
@ -1020,24 +987,18 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"GetNode", "Network {0} not found.".format(networkid)
"GetNode", f"Network {networkid} not found."
)
if memberid not in self.members:
raise ResourceNotFoundException(
"GetNode", "Member {0} not found.".format(memberid)
)
raise ResourceNotFoundException("GetNode", f"Member {memberid} not found.")
if nodeid not in self.nodes:
raise ResourceNotFoundException(
"GetNode", "Node {0} not found.".format(nodeid)
)
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":
raise ResourceNotFoundException(
"GetNode", "Node {0} not found.".format(nodeid)
)
raise ResourceNotFoundException("GetNode", f"Node {nodeid} not found.")
return self.nodes.get(nodeid)
@ -1045,18 +1006,16 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"DeleteNode", "Network {0} not found.".format(networkid)
"DeleteNode", f"Network {networkid} not found."
)
if memberid not in self.members:
raise ResourceNotFoundException(
"DeleteNode", "Member {0} not found.".format(memberid)
"DeleteNode", f"Member {memberid} not found."
)
if nodeid not in self.nodes:
raise ResourceNotFoundException(
"DeleteNode", "Node {0} not found.".format(nodeid)
)
raise ResourceNotFoundException("DeleteNode", f"Node {nodeid} not found.")
self.nodes.get(nodeid).delete()
@ -1064,18 +1023,16 @@ class ManagedBlockchainBackend(BaseBackend):
# Check if network exists
if networkid not in self.networks:
raise ResourceNotFoundException(
"UpdateNode", "Network {0} not found.".format(networkid)
"UpdateNode", f"Network {networkid} not found."
)
if memberid not in self.members:
raise ResourceNotFoundException(
"UpdateNode", "Member {0} not found.".format(memberid)
"UpdateNode", f"Member {memberid} not found."
)
if nodeid not in self.nodes:
raise ResourceNotFoundException(
"UpdateNode", "Node {0} not found.".format(nodeid)
)
raise ResourceNotFoundException("UpdateNode", f"Node {nodeid} not found.")
self.nodes.get(nodeid).update(logpublishingconfiguration)

48
moto/mediaconnect/models.py
View File

@ -215,9 +215,7 @@ class MediaConnectBackend(BaseBackend):
flow = self._flows[flow_arn]
flow.vpc_interfaces = vpc_interfaces
else:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
return flow_arn, flow.vpc_interfaces
def add_flow_outputs(self, flow_arn, outputs):
@ -225,9 +223,7 @@ class MediaConnectBackend(BaseBackend):
flow = self._flows[flow_arn]
flow.outputs = outputs
else:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
return flow_arn, flow.outputs
def remove_flow_vpc_interface(self, flow_arn, vpc_interface_name):
@ -239,9 +235,7 @@ class MediaConnectBackend(BaseBackend):
if vpc_interface["name"] != vpc_interface_name
]
else:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
return flow_arn, vpc_interface_name
def remove_flow_output(self, flow_arn, output_name):
@ -253,9 +247,7 @@ class MediaConnectBackend(BaseBackend):
if output["name"] != output_name
]
else:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
return flow_arn, output_name
def update_flow_output(
@ -279,9 +271,7 @@ class MediaConnectBackend(BaseBackend):
vpc_interface_attachment,
):
if flow_arn not in self._flows:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
flow = self._flows[flow_arn]
for output in flow.outputs:
if output["outputArn"] == output_arn:
@ -303,15 +293,11 @@ class MediaConnectBackend(BaseBackend):
output["streamId"] = stream_id
output["vpcInterfaceAttachment"] = vpc_interface_attachment
return flow_arn, output
raise NotFoundException(
message="output with arn={} not found".format(output_arn)
)
raise NotFoundException(message=f"output with arn={output_arn} not found")
def add_flow_sources(self, flow_arn, sources):
if flow_arn not in self._flows:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
flow = self._flows[flow_arn]
for source in sources:
source_id = random.uuid4().hex
@ -342,9 +328,7 @@ class MediaConnectBackend(BaseBackend):
whitelist_cidr,
):
if flow_arn not in self._flows:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
flow = self._flows[flow_arn]
source = next(
iter(
@ -378,9 +362,7 @@ class MediaConnectBackend(BaseBackend):
entitlements,
):
if flow_arn not in self._flows:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
flow = self._flows[flow_arn]
for entitlement in entitlements:
entitlement_id = random.uuid4().hex
@ -393,16 +375,14 @@ class MediaConnectBackend(BaseBackend):
def revoke_flow_entitlement(self, flow_arn, entitlement_arn):
if flow_arn not in self._flows:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
flow = self._flows[flow_arn]
for entitlement in flow.entitlements:
if entitlement_arn == entitlement["entitlementArn"]:
flow.entitlements.remove(entitlement)
return flow_arn, entitlement_arn
raise NotFoundException(
message="entitlement with arn={} not found".format(entitlement_arn)
message=f"entitlement with arn={entitlement_arn} not found"
)
def update_flow_entitlement(
@ -416,9 +396,7 @@ class MediaConnectBackend(BaseBackend):
subscribers,
):
if flow_arn not in self._flows:
raise NotFoundException(
message="flow with arn={} not found".format(flow_arn)
)
raise NotFoundException(message=f"flow with arn={flow_arn} not found")
flow = self._flows[flow_arn]
for entitlement in flow.entitlements:
if entitlement_arn == entitlement["entitlementArn"]:
@ -429,7 +407,7 @@ class MediaConnectBackend(BaseBackend):
entitlement["subscribers"] = subscribers
return flow_arn, entitlement
raise NotFoundException(
message="entitlement with arn={} not found".format(entitlement_arn)
message=f"entitlement with arn={entitlement_arn} not found"
)
# add methods from here

4
moto/medialive/models.py
View File

@ -134,7 +134,7 @@ class MediaLiveBackend(BaseBackend):
The RequestID and Reserved parameters are not yet implemented
"""
channel_id = mock_random.uuid4().hex
arn = "arn:aws:medialive:channel:{}".format(channel_id)
arn = f"arn:aws:medialive:channel:{channel_id}"
channel = Channel(
arn=arn,
cdi_input_specification=cdi_input_specification,
@ -228,7 +228,7 @@ class MediaLiveBackend(BaseBackend):
The VPC and RequestId parameters are not yet implemented
"""
input_id = mock_random.uuid4().hex
arn = "arn:aws:medialive:input:{}".format(input_id)
arn = f"arn:aws:medialive:input:{input_id}"
a_input = Input(
arn=arn,
input_id=input_id,

24
moto/mediapackage/models.py
View File

@ -73,7 +73,7 @@ class MediaPackageBackend(BaseBackend):
self._origin_endpoints = OrderedDict()
def create_channel(self, description, channel_id, tags):
arn = "arn:aws:mediapackage:channel:{}".format(channel_id)
arn = f"arn:aws:mediapackage:channel:{channel_id}"
channel = Channel(
arn=arn,
description=description,
@ -97,7 +97,7 @@ class MediaPackageBackend(BaseBackend):
return channel.to_dict()
except KeyError:
error = "NotFoundException"
raise ClientError(error, "channel with id={} not found".format(channel_id))
raise ClientError(error, f"channel with id={channel_id} not found")
def delete_channel(self, channel_id):
try:
@ -107,7 +107,7 @@ class MediaPackageBackend(BaseBackend):
except KeyError:
error = "NotFoundException"
raise ClientError(error, "channel with id={} not found".format(channel_id))
raise ClientError(error, f"channel with id={channel_id} not found")
def create_origin_endpoint(
self,
@ -126,10 +126,8 @@ class MediaPackageBackend(BaseBackend):
time_delay_seconds,
whitelist,
):
arn = "arn:aws:mediapackage:origin_endpoint:{}".format(endpoint_id)
url = "https://origin-endpoint.mediapackage.{}.amazonaws.com/{}".format(
self.region_name, endpoint_id
)
arn = f"arn:aws:mediapackage:origin_endpoint:{endpoint_id}"
url = f"https://origin-endpoint.mediapackage.{self.region_name}.amazonaws.com/{endpoint_id}"
origin_endpoint = OriginEndpoint(
arn=arn,
authorization=authorization,
@ -157,9 +155,7 @@ class MediaPackageBackend(BaseBackend):
return origin_endpoint.to_dict()
except KeyError:
error = "NotFoundException"
raise ClientError(
error, "origin endpoint with id={} not found".format(endpoint_id)
)
raise ClientError(error, f"origin endpoint with id={endpoint_id} not found")
def list_origin_endpoints(self):
origin_endpoints = list(self._origin_endpoints.values())
@ -173,9 +169,7 @@ class MediaPackageBackend(BaseBackend):
return origin_endpoint.to_dict()
except KeyError:
error = "NotFoundException"
raise ClientError(
error, "origin endpoint with id={} not found".format(endpoint_id)
)
raise ClientError(error, f"origin endpoint with id={endpoint_id} not found")
def update_origin_endpoint(
self,
@ -209,9 +203,7 @@ class MediaPackageBackend(BaseBackend):
except KeyError:
error = "NotFoundException"
raise ClientError(
error, "origin endpoint with id={} not found".format(endpoint_id)
)
raise ClientError(error, f"origin endpoint with id={endpoint_id} not found")
mediapackage_backends = BackendDict(MediaPackageBackend, "mediapackage")

4
moto/mediastore/models.py
View File

@ -42,11 +42,11 @@ class MediaStoreBackend(BaseBackend):
self._containers = OrderedDict()
def create_container(self, name, tags):
arn = "arn:aws:mediastore:container:{}".format(name)
arn = f"arn:aws:mediastore:container:{name}"
container = Container(
arn=arn,
name=name,
endpoint="/{}".format(name),
endpoint=f"/{name}",
status="CREATING",
creation_time=date.today().strftime("%m/%d/%Y, %H:%M:%S"),
tags=tags,

4
moto/mediastoredata/models.py
View File

@ -45,7 +45,7 @@ class MediaStoreDataBackend(BaseBackend):
def delete_object(self, path):
if path not in self._objects:
error = "ObjectNotFoundException"
raise ClientError(error, "Object with id={} not found".format(path))
raise ClientError(error, f"Object with id={path} not found")
del self._objects[path]
return {}
@ -56,7 +56,7 @@ class MediaStoreDataBackend(BaseBackend):
objects_found = [item for item in self._objects.values() if item.path == path]
if len(objects_found) == 0:
error = "ObjectNotFoundException"
raise ClientError(error, "Object with id={} not found".format(path))
raise ClientError(error, f"Object with id={path} not found")
return objects_found[0]
def list_items(self):

6
moto/moto_server/utilities.py
View File

@ -23,10 +23,8 @@ class AWSTestHelper(FlaskClient):
opts = {"Action": action_name}
opts.update(kwargs)
res = self.get(
"/?{0}".format(urlencode(opts)),
headers={
"Host": "{0}.us-east-1.amazonaws.com".format(self.application.service)
},
f"/?{urlencode(opts)}",
headers={"Host": f"{self.application.service}.us-east-1.amazonaws.com"},
)
return res.data.decode("utf-8")

26
moto/moto_server/werkzeug_app.py
View File

@ -74,15 +74,13 @@ class DomainDispatcherApplication(object):
return host
for backend, pattern in self.backend_url_patterns:
if pattern.match("http://%s" % host):
if pattern.match(f"http://{host}"):
return backend
if "amazonaws.com" in host:
print( # noqa
"Unable to find appropriate backend for {}."
"Remember to add the URL to urls.py, and run scripts/update_backend_index.py to index it.".format(
host
)
f"Unable to find appropriate backend for {host}."
"Remember to add the URL to urls.py, and run scripts/update_backend_index.py to index it."
)
def infer_service_region_host(self, body, environ):
@ -129,9 +127,7 @@ class DomainDispatcherApplication(object):
elif service == "mediastore" and not target:
# All MediaStore API calls have a target header
# If no target is set, assume we're trying to reach the mediastore-data service
host = "data.{service}.{region}.amazonaws.com".format(
service=service, region=region
)
host = f"data.{service}.{region}.amazonaws.com"
elif service == "dynamodb":
if environ["HTTP_X_AMZ_TARGET"].startswith("DynamoDBStreams"):
host = "dynamodbstreams"
@ -145,21 +141,15 @@ class DomainDispatcherApplication(object):
else:
host = "dynamodb"
elif service == "sagemaker":
host = "api.{service}.{region}.amazonaws.com".format(
service=service, region=region
)
host = f"api.{service}.{region}.amazonaws.com"
elif service == "timestream":
host = "ingest.{service}.{region}.amazonaws.com".format(
service=service, region=region
)
host = f"ingest.{service}.{region}.amazonaws.com"
elif service == "s3" and (
path.startswith("/v20180820/") or "s3-control" in environ["HTTP_HOST"]
):
host = "s3control"
else:
host = "{service}.{region}.amazonaws.com".format(
service=service, region=region
)
host = f"{service}.{region}.amazonaws.com"
return host
@ -281,7 +271,7 @@ def create_backend_app(service):
for url_path, handler in backend.flask_paths.items():
view_func = convert_to_flask_response(handler)
if handler.__name__ == "dispatch":
endpoint = "{0}.dispatch".format(handler.__self__.__name__)
endpoint = f"{handler.__self__.__name__}.dispatch"
else:
endpoint = view_func.__name__