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
|
"""ApiGatewayManagementApiBackend class with methods for supported APIs."""
from collections import defaultdict
from typing import Any
from moto.core.base_backend import BackendDict, BaseBackend
from moto.core.utils import unix_time
class Connection:
def __init__(self) -> None:
self.connected_at = unix_time()
self.source_ip = "192.168.0.1"
self.user_agent = "Moto Mocks"
self.data = b""
def to_dict(self) -> dict[str, Any]:
return {
"connectedAt": self.connected_at,
"lastActiveAt": unix_time(),
"identity": {
"sourceIp": self.source_ip,
"userAgent": self.user_agent,
},
}
class ApiGatewayManagementApiBackend(BaseBackend):
"""
Connecting to this API in ServerMode/Docker requires Python >= 3.8 and an up-to-date `werkzeug` version (>=2.3.x)
"""
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.connections: dict[str, Connection] = defaultdict(Connection)
def delete_connection(self, connection_id: str) -> None:
self.connections.pop(connection_id, None)
def get_connection(self, connection_id: str) -> Connection:
return self.connections[connection_id]
def post_to_connection(self, data: bytes, connection_id: str) -> None:
cnctn = self.get_connection(connection_id)
cnctn.data += data
apigatewaymanagementapi_backends = BackendDict(
ApiGatewayManagementApiBackend, "apigateway"
)
|
