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
|
"""SimpleDBBackend class with methods for supported APIs."""
import re
from collections import defaultdict
from collections.abc import Iterable
from threading import Lock
from typing import Any, Optional
from moto.core.base_backend import BackendDict, BaseBackend
from moto.core.common_models import BaseModel
from .exceptions import InvalidDomainName, UnknownDomainName
class FakeItem(BaseModel):
def __init__(self) -> None:
self.attributes: list[dict[str, Any]] = []
self.lock = Lock()
def get_attributes(self, names: Optional[list[str]]) -> list[dict[str, Any]]:
if not names:
return self.attributes
return [attr for attr in self.attributes if attr["Name"] in names]
def put_attributes(self, attributes: list[dict[str, Any]]) -> None:
# Replacing attributes involves quite a few loops
# Lock this, so we know noone else touches this list while we're operating on it
with self.lock:
for attr in attributes:
if attr.get("Replace", False):
self._remove_attributes(attr["Name"])
self.attributes.append(attr)
def _remove_attributes(self, name: str) -> None:
self.attributes = [attr for attr in self.attributes if attr["Name"] != name]
class FakeDomain(BaseModel):
def __init__(self, name: str):
self.name = name
self.items: dict[str, FakeItem] = defaultdict(FakeItem)
def get(self, item_name: str, attribute_names: list[str]) -> list[dict[str, Any]]:
item = self.items[item_name]
return item.get_attributes(attribute_names)
def put(self, item_name: str, attributes: list[dict[str, Any]]) -> None:
item = self.items[item_name]
item.put_attributes(attributes)
class SimpleDBBackend(BaseBackend):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.domains: dict[str, FakeDomain] = {}
def create_domain(self, domain_name: str) -> None:
self._validate_domain_name(domain_name)
self.domains[domain_name] = FakeDomain(name=domain_name)
def list_domains(self) -> Iterable[str]:
"""
The `max_number_of_domains` and `next_token` parameter have not been implemented yet - we simply return all domains.
"""
return self.domains.keys()
def delete_domain(self, domain_name: str) -> None:
self._validate_domain_name(domain_name)
# Ignore unknown domains - AWS does the same
self.domains.pop(domain_name, None)
def _validate_domain_name(self, domain_name: str) -> None:
# Domain Name needs to have at least 3 chars
# Can only contain characters: a-z, A-Z, 0-9, '_', '-', and '.'
if not re.match("^[a-zA-Z0-9-_.]{3,}$", domain_name):
raise InvalidDomainName(domain_name)
def _get_domain(self, domain_name: str) -> FakeDomain:
if domain_name not in self.domains:
raise UnknownDomainName()
return self.domains[domain_name]
def get_attributes(
self, domain_name: str, item_name: str, attribute_names: list[str]
) -> list[dict[str, Any]]:
"""
Behaviour for the consistent_read-attribute is not yet implemented
"""
self._validate_domain_name(domain_name)
domain = self._get_domain(domain_name)
return domain.get(item_name, attribute_names)
def put_attributes(
self, domain_name: str, item_name: str, attributes: list[dict[str, Any]]
) -> None:
"""
Behaviour for the expected-attribute is not yet implemented.
"""
self._validate_domain_name(domain_name)
domain = self._get_domain(domain_name)
domain.put(item_name, attributes)
sdb_backends = BackendDict(SimpleDBBackend, "sdb")
|
