aboutsummaryrefslogtreecommitdiffstats
path: root/contrib/python/moto/py3/moto/mediastoredata/models.py
blob: 395faa1f9588fcb7e55e9a7aee86b9b96e64313c (plain) (blame)
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
import hashlib
from collections import OrderedDict

from moto.core import BaseBackend, BaseModel
from moto.core.utils import BackendDict
from .exceptions import ClientError


class Object(BaseModel):
    def __init__(self, path, body, etag, storage_class="TEMPORAL"):
        self.path = path
        self.body = body
        self.content_sha256 = hashlib.sha256(body.encode("utf-8")).hexdigest()
        self.etag = etag
        self.storage_class = storage_class

    def to_dict(self):
        data = {
            "ETag": self.etag,
            "Name": self.path,
            "Type": "FILE",
            "ContentLength": 123,
            "StorageClass": self.storage_class,
            "Path": self.path,
            "ContentSHA256": self.content_sha256,
        }

        return data


class MediaStoreDataBackend(BaseBackend):
    def __init__(self, region_name, account_id):
        super().__init__(region_name, account_id)
        self._objects = OrderedDict()

    def put_object(self, body, path, storage_class="TEMPORAL"):
        """
        The following parameters are not yet implemented: ContentType, CacheControl, UploadAvailability
        """
        new_object = Object(
            path=path, body=body, etag="etag", storage_class=storage_class
        )
        self._objects[path] = new_object
        return new_object

    def delete_object(self, path):
        if path not in self._objects:
            error = "ObjectNotFoundException"
            raise ClientError(error, "Object with id={} not found".format(path))
        del self._objects[path]
        return {}

    def get_object(self, path):
        """
        The Range-parameter is not yet supported.
        """
        objects_found = [item for item in self._objects.values() if item.path == path]
        if len(objects_found) == 0:
            error = "ObjectNotFoundException"
            raise ClientError(error, "Object with id={} not found".format(path))
        return objects_found[0]

    def list_items(self):
        """
        The Path- and MaxResults-parameters are not yet supported.
        """
        items = self._objects.values()
        response_items = [c.to_dict() for c in items]
        return response_items


mediastoredata_backends = BackendDict(MediaStoreDataBackend, "mediastore-data")