Refactoring + stronger type annotations in file sources

lib/galaxy/files/__init__.py

@@ -3,10 +3,12 @@
 from collections import defaultdict
 from typing import (
     Any,
+    Callable,
     Dict,
     List,
     NamedTuple,
     Optional,
+    Protocol,
     Set,
 )
 
@@ -121,7 +123,7 @@ def get_file_source_path(self, uri):
         path = file_source.to_relative_path(uri)
         return FileSourcePath(file_source, path)
 
-    def validate_uri_root(self, uri: str, user_context: "ProvidesUserFileSourcesUserContext"):
+    def validate_uri_root(self, uri: str, user_context: "FileSourcesUserContext"):
         # validate a URI against Galaxy's configuration, environment, and the current
         # user. Throw appropriate exception if there is a problem with the files source
         # referenced by the URI.
@@ -164,7 +166,7 @@ def looks_like_uri(self, path_or_uri):
     def plugins_to_dict(
         self,
         for_serialization: bool = False,
-        user_context: Optional["FileSourceDictifiable"] = None,
+        user_context: "OptionalUserContext" = None,
         browsable_only: Optional[bool] = False,
         include_kind: Optional[Set[PluginKind]] = None,
         exclude_kind: Optional[Set[PluginKind]] = None,
@@ -183,9 +185,7 @@ def plugins_to_dict(
             rval.append(el)
         return rval
 
-    def to_dict(
-        self, for_serialization: bool = False, user_context: Optional["FileSourceDictifiable"] = None
-    ) -> Dict[str, Any]:
+    def to_dict(self, for_serialization: bool = False, user_context: "OptionalUserContext" = None) -> Dict[str, Any]:
         return {
             "file_sources": self.plugins_to_dict(for_serialization=for_serialization, user_context=user_context),
             "config": self._file_sources_config.to_dict(),
@@ -224,30 +224,59 @@ def __init__(
         super().__init__(FileSourcePluginsConfig())
 
 
-class FileSourceDictifiable(Dictifiable):
+class DictifiableFilesSourceContext(Protocol):
+    @property
+    def role_names(self) -> Set[str]: ...
+
+    @property
+    def group_names(self) -> Set[str]: ...
+
+    @property
+    def file_sources(self) -> ConfiguredFileSources: ...
+
+    def to_dict(
+        self, view: str = "collection", value_mapper: Optional[Dict[str, Callable]] = None
+    ) -> Dict[str, Any]: ...
+
+
+class FileSourceDictifiable(Dictifiable, DictifiableFilesSourceContext):
     dict_collection_visible_keys = ("email", "username", "ftp_dir", "preferences", "is_admin")
 
-    def to_dict(self, view="collection", value_mapper=None):
+    def to_dict(self, view="collection", value_mapper: Optional[Dict[str, Callable]] = None) -> Dict[str, Any]:
         rval = super().to_dict(view=view, value_mapper=value_mapper)
         rval["role_names"] = list(self.role_names)
         rval["group_names"] = list(self.group_names)
         return rval
 
+
+class FileSourcesUserContext(DictifiableFilesSourceContext, Protocol):
+
     @property
-    def role_names(self) -> Set[str]:
-        raise NotImplementedError
+    def email(self) -> str: ...
 
     @property
-    def group_names(self) -> Set[str]:
-        raise NotImplementedError
+    def username(self) -> str: ...
 
     @property
-    def file_sources(self):
-        """Return other filesources available in the system, for chained filesource resolution"""
-        raise NotImplementedError
+    def ftp_dir(self) -> str: ...
+
+    @property
+    def preferences(self) -> Dict[str, Any]: ...
+
+    @property
+    def is_admin(self) -> bool: ...
+
+    @property
+    def user_vault(self) -> Dict[str, Any]: ...
+
+    @property
+    def app_vault(self) -> Dict[str, Any]: ...
+
+
+OptionalUserContext = Optional[FileSourcesUserContext]
 
 
-class ProvidesUserFileSourcesUserContext(FileSourceDictifiable):
+class ProvidesFileSourcesUserContext(FileSourcesUserContext, FileSourceDictifiable):
     """Implement a FileSourcesUserContext from a Galaxy ProvidesUserContext (e.g. trans)."""
 
     def __init__(self, trans, **kwargs):
@@ -306,7 +335,7 @@ def file_sources(self):
         return self.trans.app.file_sources
 
 
-class DictFileSourcesUserContext(FileSourceDictifiable):
+class DictFileSourcesUserContext(FileSourcesUserContext, FileSourceDictifiable):
     def __init__(self, **kwd):
         self._kwd = kwd
 

lib/galaxy/files/sources/__init__.py

@@ -9,8 +9,11 @@
 from typing import (
     Any,
     ClassVar,
+    List,
     Optional,
     Set,
+    TYPE_CHECKING,
+    Union,
 )
 
 from typing_extensions import (
@@ -33,6 +36,12 @@
 DEFAULT_SCHEME = "gxfiles"
 DEFAULT_WRITABLE = False
 
+if TYPE_CHECKING:
+    from galaxy.files import (
+        FileSourcesUserContext,
+        OptionalUserContext,
+    )
+
 
 class PluginKind(str, Enum):
     """Enum to distinguish between different kinds or categories of plugins."""
@@ -140,6 +149,9 @@ class RemoteFile(RemoteEntry, TFileClass):
     ctime: str
 
 
+AnyRemoteEntry = Union[RemoteDirectory, RemoteFile]
+
+
 class SingleFileSource(metaclass=abc.ABCMeta):
     """
     Represents a protocol handler for a single remote file that can be read by or written to by Galaxy.
@@ -160,12 +172,16 @@ def get_writable(self) -> bool:
         """Return a boolean indicating whether this target is writable."""
 
     @abc.abstractmethod
-    def user_has_access(self, user_context) -> bool:
+    def user_has_access(self, user_context: "OptionalUserContext") -> bool:
         """Return a boolean indicating whether the user can access the FileSource."""
 
     @abc.abstractmethod
     def realize_to(
-        self, source_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+        self,
+        source_path: str,
+        native_path: str,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
     ):
         """Realize source path (relative to uri root) to local file system path.
 
@@ -174,14 +190,18 @@ def realize_to(
         :param native_path: local path to write to. e.g. `/tmp/myfile.txt`
         :type native_path: str
         :param user_context: A user context , defaults to None
-        :type user_context: FileSourceDictifiable, optional
+        :type user_context: OptionalUserContext, optional
         :param opts: A set of options to exercise additional control over the realize_to method. Filesource specific, defaults to None
         :type opts: Optional[FilesSourceOptions], optional
         """
 
     @abc.abstractmethod
     def write_from(
-        self, target_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+        self,
+        target_path: str,
+        native_path: str,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
     ):
         """Write file at native path to target_path (relative to uri root).
 
@@ -232,7 +252,7 @@ def to_relative_path(self, url: str) -> str:
         returned unchanged."""
 
     @abc.abstractmethod
-    def to_dict(self, for_serialization=False, user_context=None) -> FilesSourceProperties:
+    def to_dict(self, for_serialization=False, user_context: "OptionalUserContext" = None) -> FilesSourceProperties:
         """Return a dictified representation of this FileSource instance.
 
         If ``user_context`` is supplied, properties should be written so user
@@ -254,7 +274,13 @@ def get_uri_root(self) -> str:
         """Return a prefix for the root (e.g. gxfiles://prefix/)."""
 
     @abc.abstractmethod
-    def list(self, path="/", recursive=False, user_context=None, opts: Optional[FilesSourceOptions] = None) -> dict:
+    def list(
+        self,
+        path="/",
+        recursive=False,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
+    ) -> List[AnyRemoteEntry]:
         """Return dictionary of 'Directory's and 'File's."""
 
 
@@ -287,7 +313,7 @@ def get_scheme(self) -> str:
     def get_writable(self) -> bool:
         return self.writable
 
-    def user_has_access(self, user_context) -> bool:
+    def user_has_access(self, user_context: "OptionalUserContext") -> bool:
         if user_context is None and self.user_context_required:
             return False
         return (
@@ -315,7 +341,7 @@ def score_url_match(self, url: str) -> int:
         root = self.get_uri_root()
         return len(root) if root in url else 0
 
-    def uri_from_path(self, path) -> str:
+    def uri_from_path(self, path: str) -> str:
         uri_root = self.get_uri_root()
         return uri_join(uri_root, path)
 
@@ -335,7 +361,7 @@ def _parse_common_config_opts(self, kwd: FilesSourceProperties):
         kwd.pop("browsable", None)
         return kwd
 
-    def to_dict(self, for_serialization=False, user_context=None) -> FilesSourceProperties:
+    def to_dict(self, for_serialization=False, user_context: "OptionalUserContext" = None) -> FilesSourceProperties:
         rval: FilesSourceProperties = {
             "id": self.id,
             "type": self.plugin_type,
@@ -361,27 +387,45 @@ def to_dict_time(self, ctime):
             return ctime.strftime("%m/%d/%Y %I:%M:%S %p")
 
     @abc.abstractmethod
-    def _serialization_props(self, user_context=None) -> FilesSourceProperties:
+    def _serialization_props(self, user_context: "OptionalUserContext" = None) -> FilesSourceProperties:
         """Serialize properties needed to recover plugin configuration.
         Used in to_dict method if for_serialization is True.
         """
 
-    def list(self, path="/", recursive=False, user_context=None, opts: Optional[FilesSourceOptions] = None):
+    def list(
+        self,
+        path="/",
+        recursive=False,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
+    ) -> List[AnyRemoteEntry]:
         self._check_user_access(user_context)
         return self._list(path, recursive, user_context, opts)
 
-    def _list(self, path="/", recursive=False, user_context=None, opts: Optional[FilesSourceOptions] = None):
+    def _list(
+        self,
+        path="/",
+        recursive=False,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
+    ):
         pass
 
     def create_entry(
-        self, entry_data: EntryData, user_context=None, opts: Optional[FilesSourceOptions] = None
+        self,
+        entry_data: EntryData,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
     ) -> Entry:
         self._ensure_writeable()
         self._check_user_access(user_context)
         return self._create_entry(entry_data, user_context, opts)
 
     def _create_entry(
-        self, entry_data: EntryData, user_context=None, opts: Optional[FilesSourceOptions] = None
+        self,
+        entry_data: EntryData,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
     ) -> Entry:
         """Create a new entry (directory) in the file source.
 
@@ -390,21 +434,45 @@ def _create_entry(
         """
         raise NotImplementedError()
 
-    def write_from(self, target_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
+    def write_from(
+        self,
+        target_path: str,
+        native_path: str,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
+    ):
         self._ensure_writeable()
         self._check_user_access(user_context)
         self._write_from(target_path, native_path, user_context=user_context, opts=opts)
 
     @abc.abstractmethod
-    def _write_from(self, target_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
+    def _write_from(
+        self,
+        target_path: str,
+        native_path: str,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
+    ):
         pass
 
-    def realize_to(self, source_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
+    def realize_to(
+        self,
+        source_path: str,
+        native_path: str,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
+    ):
         self._check_user_access(user_context)
         self._realize_to(source_path, native_path, user_context, opts=opts)
 
     @abc.abstractmethod
-    def _realize_to(self, source_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
+    def _realize_to(
+        self,
+        source_path: str,
+        native_path: str,
+        user_context: "OptionalUserContext" = None,
+        opts: Optional[FilesSourceOptions] = None,
+    ):
         pass
 
     def _ensure_writeable(self):
@@ -420,7 +488,7 @@ def _check_user_access(self, user_context):
         if user_context is not None and not self.user_has_access(user_context):
             raise ItemAccessibilityException(f"User {user_context.username} has no access to file source.")
 
-    def _evaluate_prop(self, prop_val: Any, user_context):
+    def _evaluate_prop(self, prop_val: Any, user_context: "OptionalUserContext"):
         rval = prop_val
         if isinstance(prop_val, str) and "$" in prop_val:
             template_context = dict(
@@ -436,12 +504,12 @@ def _evaluate_prop(self, prop_val: Any, user_context):
 
         return rval
 
-    def _user_has_required_roles(self, user_context) -> bool:
+    def _user_has_required_roles(self, user_context: "FileSourcesUserContext") -> bool:
         if self.requires_roles:
             return self._evaluate_security_rules(self.requires_roles, user_context.role_names)
         return True
 
-    def _user_has_required_groups(self, user_context) -> bool:
+    def _user_has_required_groups(self, user_context: "FileSourcesUserContext") -> bool:
         if self.requires_groups:
             return self._evaluate_security_rules(self.requires_groups, user_context.group_names)
         return True
@@ -464,7 +532,7 @@ def _get_error_msg_for(rule_name: str) -> str:
             raise ConfigurationError(_get_error_msg_for("requires_groups"))
 
 
-def uri_join(*args):
+def uri_join(*args: str) -> str:
     # url_join doesn't work with non-standard scheme
     if "://" in (arg0 := args[0]):
         scheme, path = arg0.split("://", 1)
@@ -474,6 +542,6 @@ def uri_join(*args):
     return rval
 
 
-def slash_join(*args):
+def slash_join(*args: str) -> str:
     # https://codereview.stackexchange.com/questions/175421/joining-strings-to-form-a-url
     return "/".join(arg.strip("/") for arg in args)