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
|
from collections import OrderedDict
from datetime import date
from typing import Any, Optional
from moto.core.base_backend import BackendDict, BaseBackend
from moto.core.common_models import BaseModel
from moto.utilities.utils import get_partition
from .exceptions import (
ContainerNotFoundException,
PolicyNotFoundException,
ResourceNotFoundException,
)
class Container(BaseModel):
def __init__(self, **kwargs: Any):
self.arn = kwargs.get("arn")
self.name = kwargs.get("name")
self.endpoint = kwargs.get("endpoint")
self.status = kwargs.get("status")
self.creation_time = kwargs.get("creation_time")
self.lifecycle_policy: Optional[str] = None
self.policy: Optional[str] = None
self.metric_policy: Optional[str] = None
self.tags = kwargs.get("tags")
def to_dict(self, exclude: Optional[list[str]] = None) -> dict[str, Any]:
data = {
"ARN": self.arn,
"Name": self.name,
"Endpoint": self.endpoint,
"Status": self.status,
"CreationTime": self.creation_time,
"Tags": self.tags,
}
if exclude:
for key in exclude:
del data[key]
return data
class MediaStoreBackend(BaseBackend):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self._containers: dict[str, Container] = OrderedDict()
def create_container(self, name: str, tags: dict[str, str]) -> Container:
arn = f"arn:{get_partition(self.region_name)}:mediastore:container:{name}"
container = Container(
arn=arn,
name=name,
endpoint=f"/{name}",
status="CREATING",
creation_time=date.today().strftime("%m/%d/%Y, %H:%M:%S"),
tags=tags,
)
self._containers[name] = container
return container
def delete_container(self, name: str) -> None:
if name not in self._containers:
raise ContainerNotFoundException()
del self._containers[name]
def describe_container(self, name: str) -> Container:
if name not in self._containers:
raise ResourceNotFoundException()
container = self._containers[name]
container.status = "ACTIVE"
return container
def list_containers(self) -> list[dict[str, Any]]:
"""
Pagination is not yet implemented
"""
return [c.to_dict() for c in self._containers.values()]
def list_tags_for_resource(self, name: str) -> Optional[dict[str, str]]:
if name not in self._containers:
raise ContainerNotFoundException()
tags = self._containers[name].tags
return tags
def put_lifecycle_policy(self, container_name: str, lifecycle_policy: str) -> None:
if container_name not in self._containers:
raise ResourceNotFoundException()
self._containers[container_name].lifecycle_policy = lifecycle_policy
def get_lifecycle_policy(self, container_name: str) -> str:
if container_name not in self._containers:
raise ResourceNotFoundException()
lifecycle_policy = self._containers[container_name].lifecycle_policy
if not lifecycle_policy:
raise PolicyNotFoundException()
return lifecycle_policy
def put_container_policy(self, container_name: str, policy: str) -> None:
if container_name not in self._containers:
raise ResourceNotFoundException()
self._containers[container_name].policy = policy
def get_container_policy(self, container_name: str) -> str:
if container_name not in self._containers:
raise ResourceNotFoundException()
policy = self._containers[container_name].policy
if not policy:
raise PolicyNotFoundException()
return policy
def put_metric_policy(self, container_name: str, metric_policy: str) -> None:
if container_name not in self._containers:
raise ResourceNotFoundException()
self._containers[container_name].metric_policy = metric_policy
def get_metric_policy(self, container_name: str) -> str:
if container_name not in self._containers:
raise ResourceNotFoundException()
metric_policy = self._containers[container_name].metric_policy
if not metric_policy:
raise PolicyNotFoundException()
return metric_policy
mediastore_backends = BackendDict(MediaStoreBackend, "mediastore")
|
