101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
"""SimpleDBBackend class with methods for supported APIs."""
|
|
import re
|
|
from collections import defaultdict
|
|
from threading import Lock
|
|
from typing import Any, Dict, List, Iterable, Optional
|
|
|
|
from moto.core import BaseBackend, BackendDict, 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").lower() == "true":
|
|
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] = dict()
|
|
|
|
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")
|