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
|
from collections.abc import Iterable
from typing import Any
from moto.core.base_backend import BackendDict, BaseBackend
from moto.core.common_models import BaseModel
from moto.core.utils import unix_time
from moto.utilities.utils import get_partition
from .exceptions import ResourceNotFoundException
class Schema(BaseModel):
def __init__(
self,
account_id: str,
region: str,
name: str,
schema: dict[str, Any],
domain: str,
):
self.name = name
self.schema = schema
self.domain = domain
self.arn = f"arn:{get_partition(region)}:personalize:{region}:{account_id}:schema/{name}"
self.created = unix_time()
def to_dict(self, full: bool = True) -> dict[str, Any]:
d: dict[str, Any] = {
"name": self.name,
"schemaArn": self.arn,
"domain": self.domain,
"creationDateTime": self.created,
"lastUpdatedDateTime": self.created,
}
if full:
d["schema"] = self.schema
return d
class PersonalizeBackend(BaseBackend):
"""Implementation of Personalize APIs."""
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.schemas: dict[str, Schema] = {}
def create_schema(self, name: str, schema_dict: dict[str, Any], domain: str) -> str:
schema = Schema(
region=self.region_name,
account_id=self.account_id,
name=name,
schema=schema_dict,
domain=domain,
)
self.schemas[schema.arn] = schema
return schema.arn
def delete_schema(self, schema_arn: str) -> None:
if schema_arn not in self.schemas:
raise ResourceNotFoundException(schema_arn)
self.schemas.pop(schema_arn, None)
def describe_schema(self, schema_arn: str) -> Schema:
if schema_arn not in self.schemas:
raise ResourceNotFoundException(schema_arn)
return self.schemas[schema_arn]
def list_schemas(self) -> Iterable[Schema]:
"""
Pagination is not yet implemented
"""
return self.schemas.values()
personalize_backends = BackendDict(PersonalizeBackend, "personalize")
|
