1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
import json
import string
from typing import Any, Union
import yaml
from moto.moto_api._internal import mock_random as random
from moto.utilities.id_generator import ResourceIdentifier, Tags, generate_str_id
class ApigwIdentifier(ResourceIdentifier):
service = "apigateway"
def __init__(self, account_id: str, region: str, name: str):
super().__init__(account_id, region, name)
def generate(
self, existing_ids: Union[list[str], None] = None, tags: Tags = None
) -> str:
return generate_str_id(
resource_identifier=self,
existing_ids=existing_ids,
tags=tags,
length=10,
include_digits=True,
lower_case=True,
)
class ApigwApiKeyIdentifier(ApigwIdentifier):
resource = "api_key"
def __init__(self, account_id: str, region: str, value: str):
super().__init__(account_id, region, value)
class ApigwAuthorizerIdentifier(ApigwIdentifier):
resource = "authorizer"
class ApigwDeploymentIdentifier(ApigwIdentifier):
resource = "deployment"
def __init__(self, account_id: str, region: str, stage_name: str):
super().__init__(account_id, region, stage_name)
class ApigwModelIdentifier(ApigwIdentifier):
resource = "model"
class ApigwRequestValidatorIdentifier(ApigwIdentifier):
resource = "request_validator"
class ApigwResourceIdentifier(ApigwIdentifier):
resource = "resource"
def __init__(
self, account_id: str, region: str, parent_id: str = "", path_name: str = "/"
):
super().__init__(
account_id,
region,
".".join((parent_id, path_name)),
)
class ApigwRestApiIdentifier(ApigwIdentifier):
resource = "rest_api"
class ApigwUsagePlanIdentifier(ApigwIdentifier):
resource = "usage_plan"
class ApigwVpcLinkIdentifier(ApigwIdentifier):
resource = "vpc_link"
def create_id() -> str:
size = 10
chars = list(range(10)) + list(string.ascii_lowercase)
return "".join(str(random.choice(chars)) for x in range(size))
def deserialize_body(body: str) -> dict[str, Any]:
try:
api_doc = json.loads(body)
except json.JSONDecodeError:
api_doc = yaml.safe_load(body)
if "openapi" in api_doc or "swagger" in api_doc:
return api_doc
return {}
def to_path(prop: str) -> str:
return "/" + prop
|
