File: ContainerProvider.py

package info (click to toggle)
uranium 5.0.0-9
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 5,328 kB
  • sloc: python: 31,765; sh: 132; makefile: 12
file content (172 lines) | stat: -rw-r--r-- 7,557 bytes parent folder | download | duplicates (3)
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
# Copyright (c) 2018 Ultimaker B.V.
# Uranium is released under the terms of the LGPLv3 or higher.

from typing import Any, cast, Dict, Iterable, Optional

from UM.Logger import Logger
from UM.PluginObject import PluginObject #We're implementing this.
from UM.PluginRegistry import PluginRegistry #To get the priority metadata to sort by.
from UM.Settings.Interfaces import ContainerInterface

class ContainerProvider(PluginObject):
    """This class serves as a database for containers.

    A plug-in can define a new source for containers by implementing the
    ``getAllIds``, ``loadMetadata`` and ``loadContainer`` methods.
    """

    def __init__(self) -> None:
        """Initialises the provider, which creates a few empty fields."""

        super().__init__()

        #The container data, dictionaries indexed by ID.
        self._metadata = {} #type: Dict[str, Dict[str, Any]] #The metadata of all containers this provider can provide.
        self._containers = {} #type: Dict[str, ContainerInterface] #The complete containers that have been loaded so far. This is filled lazily upon requesting profiles.

    def __getitem__(self, container_id: str) -> ContainerInterface:
        """Gets a container with a specified ID.

        This should be implemented lazily. An implementation should first check

        :param container_id: The ID of a container to get.
        :return: The specified container.
        """

        if container_id not in self._containers:
            try:
                self._containers[container_id] = self.loadContainer(container_id)
            except:
                Logger.logException("e", "Failed to load container %s", container_id)
                raise
        return self._containers[container_id]

    def __lt__(self, other: object) -> bool:
        """Compares container providers by their priority so that they are easy to
        sort.

        :param other: The other container provider to compare with.
        :return: A positive number if this provider has lower priority than the
        other, or a negative number if this provider has higher priority than
        the other.
        """

        if type(other) is not type(self):
            return False
        other = cast(ContainerProvider, other)

        plugin_registry = PluginRegistry.getInstance()
        my_metadata = plugin_registry.getMetaData(self.getPluginId())
        other_metadata = plugin_registry.getMetaData(other.getPluginId())
        return my_metadata["container_provider"]["priority"] < other_metadata["container_provider"]["priority"]

    def __eq__(self, other: object) -> bool:
        if type(other) is not type(self):
            return False
        other = cast(ContainerProvider, other)

        plugin_registry = PluginRegistry.getInstance()
        my_metadata = plugin_registry.getMetaData(self.getPluginId())
        other_metadata = plugin_registry.getMetaData(other.getPluginId())
        return my_metadata["container_provider"]["priority"] == other_metadata["container_provider"]["priority"]

    def addMetadata(self, metadata: Dict[str, Any]) -> None:
        """Adds an item to the list of metadata.

        This is intended to be called from the implementation of
        ``loadMetadata``.
        """

        if "id" not in metadata:
            raise ValueError("The specified metadata has no ID.")
        if metadata["id"] in self._metadata:
            raise KeyError("The specified metadata already exists.")
        self._metadata[metadata["id"]] = metadata

    def getMetadata(self, container_id: str) -> Optional[Dict[str, Any]]:
        """Gets the metadata of a specified container.

        If the metadata of the container doesn't exist yet, it is loaded from
        the container source by the implementation of the provider.

        Note that due to inheritance, this may also trigger the metadata of
        other containers to load.

        :param container_id: The container to get the metadata of.
        :return: A dictionary of metadata for this container, or ``None`` if it
        failed to load.
        """

        if container_id not in self._metadata:
            metadata = self.loadMetadata(container_id)
            if metadata is None:
                Logger.log("e", "Failed to load metadata of container %s", container_id)
                return None
        return self._metadata[container_id]

    def getContainerFilePathById(self, container_id: str) -> Optional[str]:
        """ Gets the container file path with for the container with the given ID. Returns None if the container/file
        doesn't exist.
        :param container_id:
        :return:
        """
        raise NotImplementedError(f"The container provider {self.__class__.__name__} doesn't properly implement getContainerFilePathById.")

    def getLastModifiedTime(self, container_id: str) -> float:
        raise NotImplementedError(f"The container provider {self.__class__.__name__} doesn't properly implement getLastModifiedTime.")

    def getAllIds(self) -> Iterable[str]:
        """Gets a list of IDs of all containers this provider provides.

        :return: A list of all container IDs.
        """

        raise NotImplementedError("The container provider {class_name} doesn't properly implement getAllIds.".format(class_name = self.__class__.__name__))

    def isReadOnly(self, container_id: str) -> bool:
        """Returns whether a container is considered read-only by this provider.

        Some providers don't allow modifying their containers at all. Some only
        allow some containers to be modified.
        :return: Whether the specified container is read-only.
        """

        raise NotImplementedError("The container provider {class_name} doesn't properly implement isReadOnly.".format(class_name = self.__class__.__name__))

    def loadContainer(self, container_id: str) -> "ContainerInterface":
        """Loads the container with the specified ID.

        This is called lazily, so it should only request to load each container
        once and only when it's really needed. The container must be fully
        loaded after this is completed, so it may take some time.

        :return: The fully loaded container.
        """

        raise NotImplementedError("The container provider {class_name} doesn't properly implement loadContainer.".format(class_name = self.__class__.__name__))

    def loadMetadata(self, container_id: str) -> Dict[str, Any]:
        """Loads the metadata of a specified container.

        This will be called during start-up. It should be efficient.

        :param container_id: The ID of the container to load the metadata of.
        :return: A dictionary of metadata dictionaries, indexed by their IDs.
        """

        raise NotImplementedError("The container provider {class_name} doesn't properly implement loadMetadata.".format(class_name = self.__class__.__name__))

    def metadata(self) -> Dict[str, Dict[str, Any]]:
        """Gets a dictionary of metadata of all containers, indexed by ID."""

        return self._metadata

    def removeContainer(self, container_id: str) -> None:
        """Delete a container from this provider.

        This deletes the container from the source. If it's read only, this
        should give an exception.
        :param container_id: The ID of the container to remove.
        """

        raise NotImplementedError("The container provider {class_name} doesn't properly implement removeContainer.".format(class_name = self.__class__.__name__))