Techdebt: MyPy RAM (#6202)
This commit is contained in:
Bert Blommers
GitHub
parent
d6ab0ac98b
commit
dda6e573dd
@ -4,21 +4,21 @@ from moto.core.exceptions import JsonRESTError
|
||||
class InvalidParameterException(JsonRESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("InvalidParameterException", message)
|
||||
|
||||
|
||||
class MalformedArnException(JsonRESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("MalformedArnException", message)
|
||||
|
||||
|
||||
class OperationNotPermittedException(JsonRESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(
|
||||
"OperationNotPermittedException",
|
||||
"Unable to enable sharing with AWS Organizations. "
|
||||
@ -30,5 +30,5 @@ class OperationNotPermittedException(JsonRESTError):
|
||||
class UnknownResourceException(JsonRESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("UnknownResourceException", message)
|
||||
|
||||
@ -1,11 +1,12 @@
|
||||
import re
|
||||
import string
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from moto.core import BaseBackend, BackendDict, BaseModel
|
||||
from moto.core.utils import unix_time
|
||||
from moto.moto_api._internal import mock_random as random
|
||||
from moto.organizations import organizations_backends
|
||||
from moto.organizations.models import organizations_backends, OrganizationsBackend
|
||||
from moto.ram.exceptions import (
|
||||
MalformedArnException,
|
||||
InvalidParameterException,
|
||||
@ -14,7 +15,7 @@ from moto.ram.exceptions import (
|
||||
)
|
||||
|
||||
|
||||
def random_resource_id(size):
|
||||
def random_resource_id(size: int) -> str:
|
||||
return "".join(random.choice(string.digits + "abcdef") for _ in range(size))
|
||||
|
||||
|
||||
@ -37,7 +38,7 @@ class ResourceShare(BaseModel):
|
||||
"transit-gateway", # Amazon EC2 transit gateway
|
||||
]
|
||||
|
||||
def __init__(self, account_id, region, **kwargs):
|
||||
def __init__(self, account_id: str, region: str, **kwargs: Any):
|
||||
self.account_id = account_id
|
||||
self.region = region
|
||||
|
||||
@ -50,15 +51,15 @@ class ResourceShare(BaseModel):
|
||||
self.last_updated_time = datetime.utcnow()
|
||||
self.name = kwargs["name"]
|
||||
self.owning_account_id = account_id
|
||||
self.principals = []
|
||||
self.resource_arns = []
|
||||
self.principals: List[str] = []
|
||||
self.resource_arns: List[str] = []
|
||||
self.status = "ACTIVE"
|
||||
|
||||
@property
|
||||
def organizations_backend(self):
|
||||
def organizations_backend(self) -> OrganizationsBackend:
|
||||
return organizations_backends[self.account_id]["global"]
|
||||
|
||||
def add_principals(self, principals):
|
||||
def add_principals(self, principals: List[str]) -> None:
|
||||
for principal in principals:
|
||||
match = re.search(
|
||||
r"^arn:aws:organizations::\d{12}:organization/(o-\w+)$", principal
|
||||
@ -108,7 +109,7 @@ class ResourceShare(BaseModel):
|
||||
for principal in principals:
|
||||
self.principals.append(principal)
|
||||
|
||||
def add_resources(self, resource_arns):
|
||||
def add_resources(self, resource_arns: List[str]) -> None:
|
||||
for resource in resource_arns:
|
||||
match = re.search(
|
||||
r"^arn:aws:[a-z0-9-]+:[a-z0-9-]*:[0-9]{12}:([a-z-]+)[/:].*$", resource
|
||||
@ -126,11 +127,11 @@ class ResourceShare(BaseModel):
|
||||
for resource in resource_arns:
|
||||
self.resource_arns.append(resource)
|
||||
|
||||
def delete(self):
|
||||
def delete(self) -> None:
|
||||
self.last_updated_time = datetime.utcnow()
|
||||
self.status = "DELETED"
|
||||
|
||||
def describe(self):
|
||||
def describe(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"allowExternalPrincipals": self.allow_external_principals,
|
||||
"creationTime": unix_time(self.creation_time),
|
||||
@ -142,7 +143,7 @@ class ResourceShare(BaseModel):
|
||||
"status": self.status,
|
||||
}
|
||||
|
||||
def update(self, **kwargs):
|
||||
def update(self, **kwargs: Any) -> None:
|
||||
self.allow_external_principals = kwargs.get(
|
||||
"allowExternalPrincipals", self.allow_external_principals
|
||||
)
|
||||
@ -151,15 +152,15 @@ class ResourceShare(BaseModel):
|
||||
|
||||
|
||||
class ResourceAccessManagerBackend(BaseBackend):
|
||||
def __init__(self, region_name, account_id):
|
||||
def __init__(self, region_name: str, account_id: str):
|
||||
super().__init__(region_name, account_id)
|
||||
self.resource_shares = []
|
||||
self.resource_shares: List[ResourceShare] = []
|
||||
|
||||
@property
|
||||
def organizations_backend(self):
|
||||
def organizations_backend(self) -> OrganizationsBackend:
|
||||
return organizations_backends[self.account_id]["global"]
|
||||
|
||||
def create_resource_share(self, **kwargs):
|
||||
def create_resource_share(self, **kwargs: Any) -> Dict[str, Any]:
|
||||
resource = ResourceShare(self.account_id, self.region_name, **kwargs)
|
||||
resource.add_principals(kwargs.get("principals", []))
|
||||
resource.add_resources(kwargs.get("resourceArns", []))
|
||||
@ -171,7 +172,7 @@ class ResourceAccessManagerBackend(BaseBackend):
|
||||
|
||||
return dict(resourceShare=response)
|
||||
|
||||
def get_resource_shares(self, **kwargs):
|
||||
def get_resource_shares(self, **kwargs: Any) -> Dict[str, Any]:
|
||||
owner = kwargs["resourceOwner"]
|
||||
|
||||
if owner not in ["SELF", "OTHER-ACCOUNTS"]:
|
||||
@ -189,7 +190,7 @@ class ResourceAccessManagerBackend(BaseBackend):
|
||||
|
||||
return dict(resourceShares=resouces)
|
||||
|
||||
def update_resource_share(self, **kwargs):
|
||||
def update_resource_share(self, **kwargs: Any) -> Dict[str, Any]:
|
||||
arn = kwargs["resourceShareArn"]
|
||||
|
||||
resource = next(
|
||||
@ -205,7 +206,7 @@ class ResourceAccessManagerBackend(BaseBackend):
|
||||
|
||||
return dict(resourceShare=response)
|
||||
|
||||
def delete_resource_share(self, arn):
|
||||
def delete_resource_share(self, arn: str) -> Dict[str, Any]:
|
||||
resource = next(
|
||||
(resource for resource in self.resource_shares if arn == resource.arn), None
|
||||
)
|
||||
@ -217,7 +218,7 @@ class ResourceAccessManagerBackend(BaseBackend):
|
||||
|
||||
return dict(returnValue=True)
|
||||
|
||||
def enable_sharing_with_aws_organization(self):
|
||||
def enable_sharing_with_aws_organization(self) -> Dict[str, Any]:
|
||||
if not self.organizations_backend.org:
|
||||
raise OperationNotPermittedException
|
||||
|
||||
|
||||
@ -1,39 +1,37 @@
|
||||
from moto.core.responses import BaseResponse
|
||||
from .models import ram_backends
|
||||
import json
|
||||
from typing import Any, Dict
|
||||
|
||||
from moto.core.responses import BaseResponse
|
||||
from .models import ram_backends, ResourceAccessManagerBackend
|
||||
|
||||
|
||||
class ResourceAccessManagerResponse(BaseResponse):
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(service_name="ram")
|
||||
|
||||
@property
|
||||
def ram_backend(self):
|
||||
def ram_backend(self) -> ResourceAccessManagerBackend:
|
||||
return ram_backends[self.current_account][self.region]
|
||||
|
||||
@property
|
||||
def request_params(self):
|
||||
def request_params(self) -> Dict[str, Any]: # type: ignore[misc]
|
||||
try:
|
||||
if self.method == "DELETE":
|
||||
return None
|
||||
|
||||
return json.loads(self.body)
|
||||
except ValueError:
|
||||
return {}
|
||||
|
||||
def create_resource_share(self):
|
||||
def create_resource_share(self) -> str:
|
||||
return json.dumps(self.ram_backend.create_resource_share(**self.request_params))
|
||||
|
||||
def get_resource_shares(self):
|
||||
def get_resource_shares(self) -> str:
|
||||
return json.dumps(self.ram_backend.get_resource_shares(**self.request_params))
|
||||
|
||||
def update_resource_share(self):
|
||||
def update_resource_share(self) -> str:
|
||||
return json.dumps(self.ram_backend.update_resource_share(**self.request_params))
|
||||
|
||||
def delete_resource_share(self):
|
||||
return json.dumps(
|
||||
self.ram_backend.delete_resource_share(self._get_param("resourceShareArn"))
|
||||
)
|
||||
def delete_resource_share(self) -> str:
|
||||
arn = self._get_param("resourceShareArn")
|
||||
return json.dumps(self.ram_backend.delete_resource_share(arn))
|
||||
|
||||
def enable_sharing_with_aws_organization(self):
|
||||
def enable_sharing_with_aws_organization(self) -> str:
|
||||
return json.dumps(self.ram_backend.enable_sharing_with_aws_organization())
|
||||
|
||||
@ -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/q*,moto/rdsdata,moto/scheduler
|
||||
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/ram,moto/rdsdata,moto/scheduler
|
||||
show_column_numbers=True
|
||||
show_error_codes = True
|
||||
disable_error_code=abstract
|
||||
|
||||
Loading…
Reference in New Issue
Block a user