Extract ElementSources classjuerg/element-sources

An ElementSources object represents the combined sources of an element.

1 files changed, 319 insertions, 0 deletions

diff --git a/src/buildstream/_elementsources.py b/src/buildstream/_elementsources.py
new file mode 100644
index 000000000..fdb404b67
--- /dev/null
+++ b/src/buildstream/_elementsources.py
@@ -0,0 +1,319 @@
+#
+#  Copyright (C) 2016-2018 Codethink Limited
+#  Copyright (C) 2017-2020 Bloomberg Finance LP
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 2 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
+
+from typing import TYPE_CHECKING, Iterator
+
+from ._context import Context
+
+from .storage._casbaseddirectory import CasBasedDirectory
+
+if TYPE_CHECKING:
+    from typing import List
+
+    from .source import Source
+
+# An ElementSources object represents the combined sources of an element.
+class ElementSources:
+    def __init__(self, context: Context):
+
+        self._context = context
+        self._sources = []  # type: List[Source]
+        self.vdir = None  # Directory with staged sources
+        self._sourcecache = context.sourcecache  # Source cache
+        self._is_resolved = False  # Whether the source is fully resolved or not
+        self._cached = None  # If the sources are known to be successfully cached in CAS
+
+        # the index of the last source in this element that requires previous
+        # sources for staging
+        self._last_source_requires_previous_idx = None
+
+    # add_source():
+    #
+    # Append source to this list of element sources.
+    #
+    # Args:
+    #   source (Source): The source to add
+    #
+    def add_source(self, source):
+        self._sources.append(source)
+
+    # sources():
+    #
+    # A generator function to enumerate the element sources
+    #
+    # Yields:
+    #   Source: The individual sources
+    #
+    def sources(self) -> Iterator["Source"]:
+        for source in self._sources:
+            yield source
+
+    # track():
+    #
+    # Calls track() on the Element sources
+    #
+    # Raises:
+    #    SourceError: If one of the element sources has an error
+    #
+    # Returns:
+    #    (list): A list of Source object ids and their new references
+    #
+    def track(self, workspace):
+        refs = []
+        for index, source in enumerate(self._sources):
+            old_ref = source.get_ref()
+            new_ref = source._track(self._sources[0:index])
+            refs.append((source._unique_id, new_ref))
+
+            # Complimentary warning that the new ref will be unused.
+            if old_ref != new_ref and workspace:
+                detail = (
+                    "This source has an open workspace.\n"
+                    + "To start using the new reference, please close the existing workspace."
+                )
+                source.warn("Updated reference will be ignored as source has open workspace", detail=detail)
+
+        return refs
+
+    # stage():
+    #
+    # Stage the element sources to a directory
+    #
+    # Returns:
+    #     (:class:`.storage.Directory`): A virtual directory object to stage sources into.
+    #
+    def stage(self):
+        # Assert sources are cached
+        assert self.cached()
+
+        self.vdir = CasBasedDirectory(self._context.get_cascache())
+
+        if self._sources:
+            # find last required source
+            last_required_previous_idx = self._last_source_requires_previous()
+
+            for source in self._sources[last_required_previous_idx:]:
+                source_dir = self._sourcecache.export(source)
+                self.vdir.import_files(source_dir)
+
+        return self.vdir
+
+    # fetch_done()
+    #
+    # Indicates that fetching the sources for this element has been done.
+    #
+    # Args:
+    #   fetched_original (bool): Whether the original sources had been asked (and fetched) or not
+    #
+    def fetch_done(self, fetched_original):
+        self._cached = True
+
+        for source in self._sources:
+            source._fetch_done(fetched_original)
+
+    # push()
+    #
+    # Push the element's sources.
+    #
+    # Returns:
+    #   (bool): True if the remote was updated, False if it already existed
+    #           and no updated was required
+    #
+    def push(self):
+        pushed = False
+
+        for source in self.sources():
+            if self._sourcecache.push(source):
+                pushed = True
+
+        return pushed
+
+    # init_workspace():
+    #
+    # Initialises a new workspace from the element sources.
+    #
+    # Args:
+    #   directory (str): Path of the workspace to init
+    #
+    def init_workspace(self, directory: str):
+        for source in self.sources():
+            source._init_workspace(directory)
+
+    # fetch():
+    #
+    # Fetch the element sources.
+    #
+    # Raises:
+    #    SourceError: If one of the element sources has an error
+    #
+    def fetch(self, fetch_original=False):
+        previous_sources = []
+        fetch_needed = False
+
+        if self._sources and not fetch_original:
+            for source in self._sources:
+                if self._sourcecache.contains(source):
+                    continue
+
+                # try and fetch from source cache
+                if not source._is_cached() and self._sourcecache.has_fetch_remotes():
+                    if self._sourcecache.pull(source):
+                        continue
+
+                fetch_needed = True
+
+        # We need to fetch original sources
+        if fetch_needed or fetch_original:
+            for source in self.sources():
+                if not source._is_cached():
+                    source._fetch(previous_sources)
+                previous_sources.append(source)
+
+            self._cache_sources()
+
+    # get_unique_key():
+    #
+    # Return something which uniquely identifies the combined sources of the
+    # element.
+    #
+    # Returns:
+    #    (str, list, dict): A string, list or dictionary as unique identifier
+    #
+    def get_unique_key(self):
+        result = []
+
+        for source in self._sources:
+            result.append({"key": source._get_unique_key(), "name": source._get_source_name()})
+
+        return result
+
+    # cached():
+    #
+    # Check if the element sources are cached in CAS, generating the source
+    # cache keys if needed.
+    #
+    # Returns:
+    #    (bool): True if the element sources are cached
+    #
+    def cached(self):
+        if self._cached is not None:
+            return self._cached
+
+        sourcecache = self._sourcecache
+
+        # Go through sources we'll cache generating keys
+        for ix, source in enumerate(self._sources):
+            if not source._key:
+                if source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
+                    source._generate_key(self._sources[:ix])
+                else:
+                    source._generate_key([])
+
+        # Check all sources are in source cache
+        for source in self._sources:
+            if not sourcecache.contains(source):
+                return False
+
+        self._cached = True
+        return True
+
+    # is_resolved():
+    #
+    # Get whether all sources of the element are resolved
+    #
+    # Returns:
+    #    (bool): True if all element sources are resolved
+    #
+    def is_resolved(self):
+        return self._is_resolved
+
+    # cached_original():
+    #
+    # Get whether all the sources of the element have their own cached
+    # copy of their sources.
+    #
+    # Returns:
+    #    (bool): True if all element sources have the original sources cached
+    #
+    def cached_original(self):
+        return all(source._is_cached() for source in self._sources)
+
+    # update_resolved_state():
+    #
+    # Updates source's resolved state
+    #
+    # An element's source state must be resolved before it may compute
+    # cache keys, because the source's ref, whether defined in yaml or
+    # from the workspace, is a component of the element's cache keys.
+    #
+    def update_resolved_state(self):
+        for source in self._sources:
+            if not source.is_resolved():
+                break
+        else:
+            self._is_resolved = True
+
+    # preflight():
+    #
+    # A internal wrapper for calling the abstract preflight() method on
+    # the element and its sources.
+    #
+    def preflight(self):
+        # Ensure that the first source does not need access to previous sources
+        if self._sources and self._sources[0]._requires_previous_sources():
+            from .element import ElementError  # pylint: disable=cyclic-import
+
+            raise ElementError(
+                "{}: {} cannot be the first source of an element "
+                "as it requires access to previous sources".format(self, self._sources[0])
+            )
+
+        # Preflight the sources
+        for source in self.sources():
+            source._preflight()
+
+    # _cache_sources():
+    #
+    # Caches the sources into the local CAS
+    #
+    def _cache_sources(self):
+        if self._sources and not self.cached():
+            last_requires_previous = 0
+            # commit all other sources by themselves
+            for idx, source in enumerate(self._sources):
+                if source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
+                    self._sourcecache.commit(source, self._sources[last_requires_previous:idx])
+                    last_requires_previous = idx
+                else:
+                    self._sourcecache.commit(source, [])
+
+    # _last_source_requires_previous
+    #
+    # This is the last source that requires previous sources to be cached.
+    # Sources listed after this will be cached separately.
+    #
+    # Returns:
+    #    (int): index of last source that requires previous sources
+    #
+    def _last_source_requires_previous(self):
+        if self._last_source_requires_previous_idx is None:
+            last_requires_previous = 0
+            for idx, source in enumerate(self._sources):
+                if source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
+                    last_requires_previous = idx
+            self._last_source_requires_previous_idx = last_requires_previous
+        return self._last_source_requires_previous_idx