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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
"""IVSBackend class with methods for supported APIs."""
from typing import Any, Optional
from moto.core.base_backend import BackendDict, BaseBackend
from moto.ivs.exceptions import ResourceNotFoundException
from moto.moto_api._internal import mock_random
from moto.utilities.paginator import paginate
from moto.utilities.utils import get_partition
class IVSBackend(BaseBackend):
"""Implementation of IVS APIs."""
PAGINATION_MODEL = {
"list_channels": {
"input_token": "next_token",
"limit_key": "max_results",
"limit_default": 100,
"unique_attribute": "arn",
},
}
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.channels: list[dict[str, Any]] = []
def create_channel(
self,
authorized: bool,
insecure_ingest: bool,
latency_mode: str,
name: str,
preset: str,
recording_configuration_arn: str,
tags: dict[str, str],
channel_type: str,
) -> tuple[dict[str, Any], dict[str, Any]]:
channel_id = mock_random.get_random_string(12)
channel_arn = f"arn:{get_partition(self.region_name)}:ivs:{self.region_name}:{self.account_id}:channel/{channel_id}"
channel = {
"arn": channel_arn,
"authorized": authorized,
"ingestEndpoint": "ingest.example.com",
"insecureIngest": insecure_ingest,
"latencyMode": latency_mode,
"name": name,
"playbackUrl": f"https://playback.example.com/{self.region_name}.{self.account_id}.{channel_id}.m3u8",
"preset": preset,
"recordingConfigurationArn": recording_configuration_arn,
"tags": tags,
"type": channel_type,
}
self.channels.append(channel)
stream_key_id = mock_random.get_random_string(12)
stream_key_arn = f"arn:{get_partition(self.region_name)}:ivs:{self.region_name}:{self.account_id}:stream-key/{stream_key_id}"
stream_key = {
"arn": stream_key_arn,
"channelArn": channel_arn,
"tags": tags,
"value": f"sk_{self.region_name}_{mock_random.token_urlsafe(32)}",
}
return channel, stream_key
@paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc]
def list_channels( # type: ignore[misc]
self,
filter_by_name: Optional[str],
filter_by_recording_configuration_arn: Optional[str],
) -> list[dict[str, Any]]:
if filter_by_name is not None:
channels = [
channel
for channel in self.channels
if channel["name"] == filter_by_name
]
elif filter_by_recording_configuration_arn is not None:
channels = [
channel
for channel in self.channels
if channel["recordingConfigurationArn"]
== filter_by_recording_configuration_arn
]
else:
channels = self.channels
return channels
def _find_channel(self, arn: str) -> dict[str, Any]:
try:
return next(channel for channel in self.channels if channel["arn"] == arn)
except StopIteration:
raise ResourceNotFoundException(f"Resource: {arn} not found")
def get_channel(self, arn: str) -> dict[str, Any]:
return self._find_channel(arn)
def batch_get_channel(
self, arns: list[str]
) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
return [channel for channel in self.channels if channel["arn"] in arns], []
def update_channel(
self,
arn: str,
authorized: Optional[bool],
insecure_ingest: Optional[bool],
latency_mode: Optional[str],
name: Optional[str],
preset: Optional[str],
recording_configuration_arn: Optional[str],
channel_type: Optional[str],
) -> dict[str, Any]:
channel = self._find_channel(arn)
if authorized is not None:
channel["authorized"] = authorized
if insecure_ingest is not None:
channel["insecureIngest"] = insecure_ingest
if latency_mode is not None:
channel["latencyMode"] = latency_mode
if name is not None:
channel["name"] = name
if preset is not None:
channel["preset"] = preset
if recording_configuration_arn is not None:
channel["recordingConfigurationArn"] = recording_configuration_arn
if channel_type is not None:
channel["type"] = channel_type
return channel
def delete_channel(self, arn: str) -> None:
self._find_channel(arn)
self.channels = [channel for channel in self.channels if channel["arn"] != arn]
ivs_backends = BackendDict(IVSBackend, "ivs")
|
