workspace.py: clean up and annotate types

1 files changed, 11 insertions, 24 deletions

diff --git a/src/buildstream/plugins/sources/workspace.py b/src/buildstream/plugins/sources/workspace.py
index 16a5a7fa4..00e7b3ba7 100644
--- a/src/buildstream/plugins/sources/workspace.py
+++ b/src/buildstream/plugins/sources/workspace.py
@@ -49,33 +49,34 @@ class WorkspaceSource(Source):
 
     BST_STAGE_VIRTUAL_DIRECTORY = True
 
-    def __init__(self, context, project, meta):
+    def __init__(self, context, project, meta) -> None:
         super().__init__(context, project, meta)
 
-        # Cached unique key to avoid multiple file system traversal if the unique key is requested multiple times.
+        # Cached unique key
         self.__unique_key = None
+        # the element source objects from the specified metasources
         self.__element_sources = []
         # the digest of the Directory following the import of the workspace
         self.__digest = None
         # the CasBasedDirectory which the fullpath is imported into
         self.__cas_dir = None
 
-    def set_element_sources(self, _element_sources):
+    def set_element_sources(self, _element_sources: [Source]) -> None:
         self.__element_sources = _element_sources
 
-    def get_element_sources(self):
+    def get_element_sources(self) -> [Source]:
         return self.__element_sources
 
     def track(self) -> SourceRef:
         return None
 
-    def configure(self, node):
+    def configure(self, node: MappingNode) -> None:
         node.validate_keys(['path', 'ref', 'kind'])
         self.path = node.get_str('path')
         self.__digest = node.get_str('ref')
         self.fullpath = os.path.join(self.get_project_directory(), self.path)
 
-    def preflight(self):
+    def preflight(self) -> None:
         for source in self.get_element_sources():
             source.preflight()
 
@@ -113,7 +114,7 @@ class WorkspaceSource(Source):
         assert not sourcecache.cas.has_open_grpc_channels()
         return (self.fullpath, self.__digest)
 
-    def init_workspace(self, directory):
+    def init_workspace(self, directory: Directory) -> None:
         # for each source held by the workspace we must call init_workspace
         # those sources may override `init_workspace` expecting str or Directory
         # and this will need to be extracted from the directory passed to this method
@@ -129,7 +130,7 @@ class WorkspaceSource(Source):
     def fetch(self) -> None:
         pass  # pragma: nocover
 
-    def stage(self, directory):
+    def stage(self, directory: Directory) -> None:
         # directory should always be a Directory object
         assert isinstance(directory, Directory)
         assert isinstance(self.__cas_dir, CasBasedDirectory)
@@ -141,24 +142,10 @@ class WorkspaceSource(Source):
                     "Failed to stage source: files clash with existing directory",
                     reason='ensure-stage-dir-fail')
 
-    def _get_local_path(self):
+    def _get_local_path(self) -> str:
         return self.fullpath
 
 
-# Create a unique key for a file
-def unique_key(filename):
-
-    # Return some hard coded things for files which
-    # have no content to calculate a key for
-    if os.path.islink(filename):
-        # For a symbolic link, use the link target as its unique identifier
-        return os.readlink(filename)
-    elif os.path.isdir(filename):
-        return "0"
-
-    return utils.sha256sum(filename)
-
-
 # Plugin entry point
-def setup():
+def setup() -> WorkspaceSource:
     return WorkspaceSource