Merge pull request #1292 from gallardoalba/Annotate_modules_oct_13

Planemo type annotation: config, context and factory

Author: nsoranzo

planemo/config.py

@@ -4,14 +4,24 @@
 from enum import Enum
 from typing import (
     Any,
+    Callable,
     Dict,
+    List,
+    Optional,
+    Tuple,
+    TYPE_CHECKING,
+    Union,
 )
 
 import click
 import yaml
+from click.core import Option
+
+if TYPE_CHECKING:
+    from planemo.cli import PlanemoCliContext
 
 PLANEMO_CONFIG_ENV_PROP = "PLANEMO_GLOBAL_CONFIG_PATH"
-DEFAULT_CONFIG = {}  # type: Dict[str, Any]
+DEFAULT_CONFIG: Dict[str, Any] = {}
 
 VALUE_UNSET = object()
 
@@ -20,11 +30,11 @@
 
 
 def _default_callback(
-    default,
-    use_global_config=False,
-    resolve_path=False,
-    extra_global_config_vars=[],
-):
+    default: Any,
+    use_global_config: bool = False,
+    resolve_path: bool = False,
+    extra_global_config_vars: List[str] = [],
+) -> Callable:
     def callback(ctx, param, value):
         planemo_ctx = ctx.obj
         param_name = param.name
@@ -55,10 +65,12 @@ def callback(ctx, param, value):
     return callback
 
 
-def _find_default(ctx, param, use_global_config, extra_global_config_vars):
+def _find_default(
+    ctx: "PlanemoCliContext", param: Option, use_global_config: bool, extra_global_config_vars: List[str]
+) -> Union[Tuple[Any, OptionSource], Tuple[object, None]]:
     if use_global_config:
         global_config = ctx.global_config
-        global_config_keys = ["default_%s" % param.name] + extra_global_config_vars
+        global_config_keys = [f"default_{param.name}"] + extra_global_config_vars
         for global_config_key in global_config_keys:
             if global_config_key in global_config:
                 default_value = global_config[global_config_key]
@@ -67,15 +79,15 @@ def _find_default(ctx, param, use_global_config, extra_global_config_vars):
     return VALUE_UNSET, None
 
 
-def planemo_option(*args, **kwargs):
+def planemo_option(*args, **kwargs) -> Callable:
     """Extend ``click.option`` with planemo-config aware configuration.
 
     This extends click.option to use a callback when assigning default
     values, add ``use_global_config`` keyword argument to allow reading
     defaults from ~/.planemo.yml, and tracks how parameters are specified
     using the Planemo Context object.
     """
-    option_type = kwargs.get("type", None)
+    option_type = kwargs.get("type")
     use_global_config = kwargs.pop("use_global_config", False)
     use_env_var = kwargs.pop("use_env_var", False)
     extra_global_config_vars = kwargs.pop("extra_global_config_vars", [])
@@ -89,7 +101,7 @@ def planemo_option(*args, **kwargs):
         outer_callback = kwargs.pop("callback", None)
 
         def callback(ctx, param, value):
-            resolve_path = option_type and getattr(option_type, "resolve_path", False)
+            resolve_path = bool(option_type and getattr(option_type, "resolve_path", False))
             result = _default_callback(
                 default,
                 use_global_config=use_global_config,
@@ -113,20 +125,20 @@ def callback(ctx, param, value):
             if arg.startswith("--"):
                 name = arg[len("--") :]
         assert name
-        kwargs["envvar"] = "PLANEMO_%s" % name.upper()
+        kwargs["envvar"] = f"PLANEMO_{name.upper()}"
 
     option = click.option(*args, **kwargs)
     return option
 
 
-def global_config_path(config_path=None):
+def global_config_path(config_path: Optional[str] = None) -> str:
     if not config_path:
-        config_path = os.environ.get(PLANEMO_CONFIG_ENV_PROP, "~/.planemo.yml")
-        config_path = os.path.expanduser(config_path)
+        planemo_config_env_prop = os.environ.get(PLANEMO_CONFIG_ENV_PROP, "~/.planemo.yml")
+        config_path = os.path.expanduser(planemo_config_env_prop)
     return config_path
 
 
-def read_global_config(config_path: str):
+def read_global_config(config_path: Optional[str]) -> Dict[str, Any]:
     config_path = global_config_path(config_path)
     if not os.path.exists(config_path):
         return DEFAULT_CONFIG

planemo/context.py

@@ -9,10 +9,19 @@
 import shutil
 import sys
 import traceback
+from typing import (
+    Any,
+    Dict,
+    Optional,
+    TYPE_CHECKING,
+)
 from urllib.request import urlopen
 
 from planemo.config import read_global_config
 
+if TYPE_CHECKING:
+    from planemo.config import OptionSource
+
 
 class PlanemoContextInterface(metaclass=abc.ABCMeta):
     """Interface Planemo operations use to access workspace context."""
@@ -53,29 +62,29 @@ def cache_download(self, url, destination):
 class PlanemoContext(PlanemoContextInterface):
     """Implementation of ``PlanemoContextInterface``"""
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Construct a Context object using execution environment."""
         self.home = os.getcwd()
-        self._global_config = None
+        self._global_config: Optional[Dict] = None
         # Will be set by planemo CLI driver
         self.verbose = False
-        self.planemo_config = None
+        self.planemo_config: Optional[str] = None
         self.planemo_directory = None
-        self.option_source = {}
+        self.option_source: Dict[str, "OptionSource"] = {}
 
-    def set_option_source(self, param_name, option_source, force=False):
+    def set_option_source(self, param_name: str, option_source: "OptionSource", force: bool = False) -> None:
         """Specify how an option was set."""
         if not force:
-            assert param_name not in self.option_source, "No option source for [%s]" % param_name
+            assert param_name not in self.option_source, f"Option source for [{param_name}] already set"
         self.option_source[param_name] = option_source
 
-    def get_option_source(self, param_name):
+    def get_option_source(self, param_name: str) -> "OptionSource":
         """Return OptionSource value indicating how the option was set."""
-        assert param_name in self.option_source, "No option source for [%s]" % param_name
+        assert param_name in self.option_source, f"No option source for [{param_name}]"
         return self.option_source[param_name]
 
     @property
-    def global_config(self):
+    def global_config(self) -> Dict[str, Any]:
         """Read Planemo's global configuration.
 
         As defined most simply by ~/.planemo.yml.
@@ -84,21 +93,21 @@ def global_config(self):
             self._global_config = read_global_config(self.planemo_config)
         return self._global_config or {}
 
-    def log(self, msg, *args):
+    def log(self, msg: str, *args) -> None:
         """Log a message."""
         if args:
             msg %= args
         self._log_message(msg)
 
-    def vlog(self, msg, *args, **kwds):
+    def vlog(self, msg: str, *args, **kwds) -> None:
         """Log a message only if verbose is enabled."""
         if self.verbose:
             self.log(msg, *args)
             if kwds.get("exception", False):
                 traceback.print_exc(file=sys.stderr)
 
     @property
-    def workspace(self):
+    def workspace(self) -> str:
         """Create and return Planemo's workspace.
 
         By default this will be ``~/.planemo``.
@@ -109,7 +118,7 @@ def workspace(self):
         return self._ensure_directory(workspace, "workspace")
 
     @property
-    def galaxy_profiles_directory(self):
+    def galaxy_profiles_directory(self) -> str:
         """Create a return a directory for storing Galaxy profiles."""
         path = os.path.join(self.workspace, "profiles")
         return self._ensure_directory(path, "Galaxy profiles")
@@ -125,27 +134,25 @@ def cache_download(self, url, destination):
             with urlopen(url) as fh:
                 content = fh.read()
             if len(content) == 0:
-                raise Exception("Failed to download [%s]." % url)
+                raise Exception(f"Failed to download [{url}].")
             with open(cache_destination, "wb") as f:
                 f.write(content)
 
         shutil.copy(cache_destination, destination)
 
-    def _ensure_directory(self, path, name):
+    def _ensure_directory(self, path: str, name: str) -> str:
         if not os.path.exists(path):
             os.makedirs(path)
         if not os.path.isdir(path):
-            template = "Planemo %s directory [%s] unavailable."
-            message = template % (name, path)
-            raise Exception(message)
+            raise Exception(f"Planemo {name} directory [{path}] unavailable.")
         return path
 
     def _log_message(self, message):
         """Extension point for overriding I/O."""
         print(message)
 
 
-def configure_standard_planemo_logging(verbose):
+def configure_standard_planemo_logging(verbose: bool) -> None:
     """Configure Planemo's default logging rules."""
     logging_config = {
         "version": 1,

planemo/database/factory.py

@@ -1,12 +1,13 @@
 """Create a DatabaseSource from supplied planemo configuration."""
 from galaxy.util.commands import which
 
+from .interface import DatabaseSource
 from .postgres import LocalPostgresDatabaseSource
 from .postgres_docker import DockerPostgresDatabaseSource
 
 
-def create_database_source(**kwds):
-    """Return a :class:`planemo.database.DatabaseSource` for configuration."""
+def create_database_source(**kwds) -> DatabaseSource:
+    """Return a :class:`planemo.database.interface.DatabaseSource` for configuration."""
     database_type = kwds.get("database_type", "auto")
     if database_type == "auto":
         if which("psql"):