Repository: JonathonReinhart/scuba

Issue:
Implement typing and mypy support

Discussion:
github-actions[bot]: ## Test Results
  5 files  5 suites   3s [:stopwatch:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "duration of all tests")
  3 tests 0 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "passed tests") 0 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "skipped / disabled tests") 0 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "failed tests")   3 [:fire:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "test errors")
15 runs  0 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "passed tests") 0 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "skipped / disabled tests") 0 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "failed tests") 15 [:fire:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "test errors")

For more details on these errors, see [this check](https://github.com/JonathonReinhart/scuba/runs/11844835832).

Results for commit d63a6ca4.

[test-results]:data:application/gzip;base64,H4sIAGs/CGQC/1WMyw6DIBBFf8Ww7gIE1PZnmnFGElKVhsfK9N9Lqa24u+fc5GzM2HkK7NboS8NCsvEPlDxE69aMMmM+Yqj3PSTELPghHvZ5EgbsfBKT987vEZ/WT0/ofde9Lx+5wlWt8C9WCuiWxcZMjDoJHYIaoVc9N0AC6SoHOdKgOLbQauJCcmSvN+QlqfX9AAAA


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a59bc3e..87587d9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,7 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 ## [Unreleased]
 ### Changed
 - Introduced `pyproject.toml` and moved metadata from `setup.py` (#211)
+- Added type annotations to scuba package and mypy checking in CI (#207)
 
 ### Removed
 - Drop support for Python 3.5 - 3.6
diff --git a/pyproject.toml b/pyproject.toml
index a7808be..9bdfc6a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -43,3 +43,16 @@ requires = [
     "wheel",
     "setuptools >= 42.0.0",
 ]
+
+
+# mypy
+[tool.mypy]
+packages = ["scuba", "tests"]
+warn_unused_configs = true
+warn_return_any = true
+
+[[tool.mypy.overrides]]
+module = "scuba.*"
+disallow_untyped_calls = true
+disallow_untyped_defs = true
+disallow_incomplete_defs = true
diff --git a/requirements.txt b/requirements.txt
index 4fe6c60..5722978 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,8 +1,11 @@
 argcomplete >= 1.10.1
 build
+mypy~=1.2.0
 PyYAML
 pytest
 pytest-cov
+types-PyYAML
+types-pkg-resources
 coverage
 black~=23.0  # Must match .github/workflows/black.yml
 -r docs/requirements.txt
diff --git a/scuba/__main__.py b/scuba/__main__.py
index d4806d7..de7820a 100644
--- a/scuba/__main__.py
+++ b/scuba/__main__.py
@@ -8,16 +8,12 @@ import sys
 import shlex
 import itertools
 import argparse
+from typing import Any, Optional, Sequence
 
 try:
-    import argcomplete
+    import argcomplete  # type: ignore
 except ImportError:
-
-    class argcomplete:
-        @staticmethod
-        def autocomplete(*_, **__):
-            pass
-
+    from . import argcomplete_stub as argcomplete  # type: ignore[no-redef]
 
 from . import dockerutil
 from .config import find_config, ScubaConfig, ConfigError, ConfigNotFoundError
@@ -26,16 +22,20 @@ from .scuba import ScubaDive, ScubaError
 from .utils import format_cmdline, parse_env_var
 from .version import __version__
 
+g_verbose: bool = False
+
 
-def appmsg(msg):
+def appmsg(msg: str) -> None:
     print("scuba: " + msg, file=sys.stderr)
 
 
-def parse_scuba_args(argv):
-    def _list_images_completer(**_):
+def parse_scuba_args(argv: Optional[Sequence[str]]) -> argparse.Namespace:
+    def _list_images_completer(**kwargs: Any) -> Sequence[str]:
         return dockerutil.get_images()
 
-    def _list_aliases_completer(parsed_args, **_):
+    def _list_aliases_completer(
+        parsed_args: argparse.Namespace, **kwargs: Any
+    ) -> Sequence[str]:
         # We don't want to try to complete any aliases if one was already given
         if parsed_args.command:
             return []
@@ -44,8 +44,9 @@ def parse_scuba_args(argv):
             _, _, config = find_config()
             return sorted(config.aliases)
         except (ConfigNotFoundError, ConfigError):
-            argcomplete.warn(
-                "No or invalid config found.  Cannot auto-complete aliases."
+            print(
+                "\nNo or invalid config found. Cannot auto-complete aliases.",
+                file=sys.stderr,
             )
             return []
 
@@ -73,7 +74,7 @@ def parse_scuba_args(argv):
     ap.add_argument("--entrypoint", help="Override the default ENTRYPOINT of the image")
 
     img_arg = ap.add_argument("--image", help="Override Docker image")
-    img_arg.completer = _list_images_completer
+    img_arg.completer = _list_images_completer  # type: ignore[attr-defined]
 
     ap.add_argument("--shell", help="Override shell used in Docker container")
     ap.add_argument(
@@ -96,7 +97,7 @@ def parse_scuba_args(argv):
         nargs=argparse.REMAINDER,
         help="Command (and arguments) to run in the container",
     )
-    cmd_arg.completer = _list_aliases_completer
+    cmd_arg.completer = _list_aliases_completer  # type: ignore[attr-defined]
 
     argcomplete.autocomplete(ap, always_complete_options=False)
     args = ap.parse_args(argv)
@@ -118,7 +119,7 @@ def parse_scuba_args(argv):
     return args
 
 
-def run_scuba(scuba_args):
+def run_scuba(scuba_args: argparse.Namespace) -> int:
     # Locate .scuba.yml
     try:
         # top_path is where .scuba.yml is found, and becomes the top of our bind mount.
@@ -164,6 +165,9 @@ def run_scuba(scuba_args):
 
         # Explicitly pass sys.stdin/stdout/stderr so they apply to the
         # child process if overridden (by tests).
+        #
+        # TODO: This doesn't seem to work in all cases with pytest:
+        # _pytest.capture.DontReadFromInput doesn't have fileno()
         return dockerutil.call(
             args=run_args,
             stdin=sys.stdin,
@@ -172,7 +176,7 @@ def run_scuba(scuba_args):
         )
 
 
-def main(argv=None):
+def main(argv: Optional[Sequence[str]] = None) -> None:
     scuba_args = parse_scuba_args(argv)
 
     try:
diff --git a/scuba/argcomplete_stub.py b/scuba/argcomplete_stub.py
new file mode 100644
index 0000000..f98ab9f
--- /dev/null
+++ b/scuba/argcomplete_stub.py
@@ -0,0 +1,7 @@
+# A stub implementation of argcomplete which does nothing.
+import argparse
+from typing import Any
+
+
+def autocomplete(parser: argparse.ArgumentParser, **kw: Any) -> None:
+    pass
diff --git a/scuba/config.py b/scuba/config.py
index 280059e..fa9c389 100644
--- a/scuba/config.py
+++ b/scuba/config.py
@@ -1,13 +1,22 @@
-import collections
+from __future__ import annotations
+import dataclasses
 import os
-import yaml
 import re
 import shlex
+from typing import Any, List, Dict, Optional, TextIO, Tuple, Type, TypeVar, Union
+
+import yaml
+import yaml.nodes
 
 from .constants import *
 from .utils import *
 from .dockerutil import make_vol_opt
 
+CfgNode = Any
+CfgData = Dict[str, CfgNode]
+Environment = Dict[str, str]
+_T = TypeVar("_T")
+
 
 class ConfigError(Exception):
     pass
@@ -25,19 +34,17 @@ class OverrideMixin:
     and any object can be checked if it is an OverrideMixin using isinstance().
     """
 
-    pass
-
 
 class OverrideNone(OverrideMixin):
     """
     Represents a None value that also has Override behavior
     """
 
-    def __bool__(self):
+    def __bool__(self) -> bool:
         return False
 
 
-class OverrideList(collections.UserList, OverrideMixin):
+class OverrideList(list, OverrideMixin):
     pass
 
 
@@ -47,15 +54,26 @@ class OverrideStr(str, OverrideMixin):
 
 # http://stackoverflow.com/a/9577670
 class Loader(yaml.SafeLoader):
-    def __init__(self, stream, root=None):
-        if root is None:
+    _root: str  # directory containing the loaded document
+    _cache: Dict[str, Any]  # document path => document
+
+    def __init__(self, stream: TextIO):
+        if not hasattr(self, "_root"):
             self._root = os.path.split(stream.name)[0]
-        else:
-            self._root = root
         self._cache = dict()
         super().__init__(stream)
 
-    def from_yaml(self, node):
+    @staticmethod
+    def _rooted_loader(root: str) -> Type[Loader]:
+        """Get a Loader class with _root set to root"""
+
+        class RootedLoader(Loader):
+            pass
+
+        RootedLoader._root = root
+        return RootedLoader
+
+    def from_yaml(self, node: yaml.nodes.Node) -> Any:
         """
         Implementes a !from_yaml constructor with the following syntax:
             !from_yaml filename key
@@ -73,7 +91,9 @@ class Loader(yaml.SafeLoader):
         """
 
         # Load the content from the node, as a scalar
+        assert isinstance(node, yaml.nodes.ScalarNode)
         content = self.construct_scalar(node)
+        assert isinstance(content, str)
 
         # Split on unquoted spaces
         parts = shlex.split(content)
@@ -103,15 +123,17 @@ class Loader(yaml.SafeLoader):
             raise yaml.YAMLError(f"Key {key!r} not found in {filename}")
         return cur
 
-    def override(self, node):
+    def override(self, node: yaml.nodes.Node) -> OverrideMixin:
         """
         Implements !override constructor
         """
         # Load the content from the node, as a scalar
+        assert isinstance(node, yaml.nodes.ScalarNode)
         content = self.construct_scalar(node)
+        assert isinstance(content, str)
 
         # Dynamically add an OverrideMixin to the resulting object's type
-        obj = yaml.load(content, lambda s: Loader(s, root=self._root))
+        obj = yaml.load(content, self._rooted_loader(root=self._root))
         if obj is None:
             obj = OverrideNone()
         else:
@@ -126,6 +148,7 @@ class Loader(yaml.SafeLoader):
                 # Primitive classes (e.g., int, str) don't support __class__ assignment
                 obj = mixin_type(obj)
 
+        assert isinstance(obj, OverrideMixin)
         return obj
 
 
@@ -133,7 +156,7 @@ Loader.add_constructor("!from_yaml", Loader.from_yaml)
 Loader.add_constructor("!override", Loader.override)
 
 
-def find_config():
+def find_config() -> Tuple[str, str, ScubaConfig]:
     """Search up the directory hierarchy for .scuba.yml
 
     Returns: path, rel, config on success, or None if not found
@@ -169,11 +192,16 @@ def find_config():
         rel = os.path.join(rest, rel)
 
 
-def _process_script_node(node, name):
+def _process_script_node(node: CfgNode, name: str) -> List[str]:
     """Process a script-type node
 
-    This handles nodes that follow the *Common script schema*,
-    as outlined in doc/yaml-reference.md.
+    Args:
+      node: A node of data retrieved from a YAML document. Should be a "common
+        script schema" node as described in docs/configuration.rst.
+      name: The name of the node.
+
+    Returns:
+      A script; a list of command strings.
     """
     if isinstance(node, str):
         # The script is just the text itself
@@ -196,7 +224,7 @@ def _process_script_node(node, name):
     raise ConfigError(f"{name}: must be string or dict")
 
 
-def _process_environment(node, name):
+def _process_environment(node: CfgNode, name: str) -> Environment:
     # Environment can be either a list of strings ("KEY=VALUE") or a mapping
     # Environment keys and values are always strings
     result = {}
@@ -220,7 +248,7 @@ def _process_environment(node, name):
     return result
 
 
-def _get_nullable_str(data, key):
+def _get_nullable_str(data: Dict[str, Any], key: str) -> Optional[str]:
     # N.B. We can't use data.get() here, because that might return
     # None, leading to ambiguity between the key being absent or set
     # to a null value.
@@ -232,52 +260,62 @@ def _get_nullable_str(data, key):
     if not key in data:
         return None
 
-    ep = data[key]
+    value = data[key]
 
     # We represent a null value as an empty string.
-    if isinstance(ep, OverrideNone):
-        ep = OverrideStr("")
-    elif ep is None:
-        ep = ""
+    if isinstance(value, OverrideNone):
+        value = OverrideStr("")
+    elif value is None:
+        value = ""
 
-    if not isinstance(ep, str):
-        raise ConfigError(f"{key!r} must be a string, not {type(ep).__name__}")
-    return ep
+    if not isinstance(value, str):
+        raise ConfigError(f"{key!r} must be a string, not {type(value).__name__}")
+    return value
 
 
-def _get_entrypoint(data):
+def _get_entrypoint(data: CfgData) -> Optional[str]:
     return _get_nullable_str(data, "entrypoint")
 
 
-def _get_docker_args(data):
-    args = _get_nullable_str(data, "docker_args")
+def _get_docker_args(data: CfgData) -> Optional[List[str]]:
+    args_str = _get_nullable_str(data, "docker_args")
+    if args_str is None:
+        return None
 
-    if args is not None:
-        override = isinstance(args, OverrideMixin)
-        args = shlex.split(args)
-        if override:
-            args = OverrideList(args)
+    override = isinstance(args_str, OverrideMixin)
+    args = shlex.split(args_str)
+    if override:
+        args = OverrideList(args)
 
     return args
 
 
-def _get_typed_val(data, key, type_):
-    v = data.get(key)
+def _get_typed_val(
+    data: CfgData,
+    key: str,
+    type_: Type[_T],
+    default: Optional[_T] = None,
+) -> Optional[_T]:
+    v = data.get(key, default)
     if v is not None and not isinstance(v, type_):
         raise ConfigError(f"{key!r} must be a {type_.__name__}, not {type(v).__name__}")
     return v
 
 
-def _get_dict(data, key):
+def _get_str(data: CfgData, key: str, default: Optional[str] = None) -> Optional[str]:
+    return _get_typed_val(data, key, str, default)
+
+
+def _get_dict(data: CfgData, key: str) -> Optional[dict]:
     return _get_typed_val(data, key, dict)
 
 
-def _get_delimited_str_list(data, key, sep):
+def _get_delimited_str_list(data: CfgData, key: str, sep: str) -> List[str]:
     s = _get_typed_val(data, key, str)
     return s.split(sep) if s else []
 
 
-def _get_volumes(data):
+def _get_volumes(data: CfgData) -> Optional[Dict[str, ScubaVolume]]:
     voldata = _get_dict(data, "volumes")
     if voldata is None:
         return None
@@ -289,7 +327,7 @@ def _get_volumes(data):
     return vols
 
 
-def _expand_path(in_str):
+def _expand_path(in_str: str) -> str:
     try:
         output = expand_env_vars(in_str)
     except KeyError as ke:
@@ -305,14 +343,14 @@ def _expand_path(in_str):
     return output
 
 
+@dataclasses.dataclass(frozen=True)
 class ScubaVolume:
-    def __init__(self, container_path, host_path=None, options=None):
-        self.container_path = container_path
-        self.host_path = host_path
-        self.options = options or []
+    container_path: str
+    host_path: str  # TODO: Optional for anonymous volume
+    options: List[str] = dataclasses.field(default_factory=list)
 
     @classmethod
-    def from_dict(cls, cpath, node):
+    def from_dict(cls, cpath: str, node: CfgNode) -> ScubaVolume:
         # Treat a null node as an empty dict
         if node is None:
             node = {}
@@ -343,37 +381,24 @@ class ScubaVolume:
 
         raise ConfigError(f"{cpath}: must be string or dict")
 
-    def get_vol_opt(self):
-        if not self.host_path:
-            raise NotImplementedError("No anonymous volumes for now")
+    def get_vol_opt(self) -> str:
         return make_vol_opt(self.host_path, self.container_path, self.options)
 
 
+@dataclasses.dataclass(frozen=True)
 class ScubaAlias:
-    def __init__(
-        self,
-        name,
-        script,
-        image=None,
-        entrypoint=None,
-        environment=None,
-        shell=None,
-        as_root=None,
-        docker_args=None,
-        volumes=None,
-    ):
-        self.name = name
-        self.script = script
-        self.image = image
-        self.entrypoint = entrypoint
-        self.environment = environment
-        self.shell = shell
-        self.as_root = bool(as_root)
-        self.docker_args = docker_args
-        self.volumes = volumes
+    name: str
+    script: List[str]
+    image: Optional[str] = None
+    entrypoint: Optional[str] = None
+    environment: Optional[Dict[str, str]] = None
+    shell: Optional[str] = None
+    as_root: bool = False
+    docker_args: Optional[List[str]] = None
+    volumes: Optional[Dict[str, ScubaVolume]] = None
 
     @classmethod
-    def from_dict(cls, name, node):
+    def from_dict(cls, name: str, node: CfgNode) -> ScubaAlias:
         script = _process_script_node(node, name)
 
         if isinstance(node, dict):  # Rich alias
@@ -386,7 +411,7 @@ class ScubaAlias:
                     node.get("environment"), f"{name}.environment"
                 ),
                 shell=node.get("shell"),
-                as_root=node.get("root"),
+                as_root=bool(node.get("root")),
                 docker_args=_get_docker_args(node),
                 volumes=_get_volumes(node),
             )
@@ -395,7 +420,15 @@ class ScubaAlias:
 
 
 class ScubaConfig:
-    def __init__(self, **data):
+    shell: str
+    entrypoint: Optional[str]
+    docker_args: Optional[List[str]]  # TODO: drop Optional?
+    volumes: Optional[Dict[str, ScubaVolume]]  # TODO: drop Optional? Dict?
+    aliases: Dict[str, ScubaAlias]
+    hooks: Dict[str, List[str]]
+    environment: Environment
+
+    def __init__(self, **data: CfgNode) -> None:
         optional_nodes = (
             "image",
             "aliases",
@@ -415,46 +448,46 @@ class ScubaConfig:
                 + ", ".join(extra)
             )
 
-        self._image = data.get("image")
-        self.shell = data.get("shell", DEFAULT_SHELL)
+        self._image = _get_str(data, "image")
+
+        shell = _get_str(data, "shell", DEFAULT_SHELL)
+        assert shell is not None  # Guaranteed by default arg. TODO: express via type?
+        self.shell = shell
+
         self.entrypoint = _get_entrypoint(data)
         self.docker_args = _get_docker_args(data)
         self.volumes = _get_volumes(data)
-        self._load_aliases(data)
-        self._load_hooks(data)
-        self._load_environment(data)
-
-    def _load_aliases(self, data):
-        self.aliases = {}
+        self.aliases = self._load_aliases(data)
+        self.hooks = self._load_hooks(data)
+        self.environment = _process_environment(data.get("environment"), "environment")
 
+    def _load_aliases(self, data: CfgData) -> Dict[str, ScubaAlias]:
+        aliases = {}
         for name, node in data.get("aliases", {}).items():
             if " " in name:
                 raise ConfigError("Alias names cannot contain spaces")
-            self.aliases[name] = ScubaAlias.from_dict(name, node)
-
-    def _load_hooks(self, data):
-        self.hooks = {}
+            aliases[name] = ScubaAlias.from_dict(name, node)
+        return aliases
 
+    def _load_hooks(self, data: CfgData) -> Dict[str, List[str]]:
+        hooks = {}
         for name in (
             "user",
             "root",
         ):
             node = data.get("hooks", {}).get(name)
             if node:
-                hook = _process_script_node(node, name)
-                self.hooks[name] = hook
-
-    def _load_environment(self, data):
-        self.environment = _process_environment(data.get("environment"), "environment")
+                hooks[name] = _process_script_node(node, name)
+        return hooks
 
     @property
-    def image(self):
-        if not self._image:
+    def image(self) -> str:
+        if self._image is None:
             raise ConfigError("Top-level 'image' not set")
         return self._image
 
 
-def load_config(path):
+def load_config(path: str) -> ScubaConfig:
     try:
         with open(path, "r") as f:
             data = yaml.load(f, Loader)
diff --git a/scuba/dockerutil.py b/scuba/dockerutil.py
index a87097d..8ca33bc 100644
--- a/scuba/dockerutil.py
+++ b/scuba/dockerutil.py
@@ -1,6 +1,11 @@
+from __future__ import annotations
 import subprocess
-import errno
 import json
+from typing import Any, Dict, IO, Optional, Sequence, Union
+
+# https://github.com/python/typeshed/blob/main/stdlib/subprocess.pyi
+_CMD = Union[str, bytes, Sequence[Union[str, bytes]]]
+_FILE = Union[None, int, IO[Any]]
 
 
 class DockerError(Exception):
@@ -8,43 +13,44 @@ class DockerError(Exception):
 
 
 class DockerExecuteError(DockerError):
-    pass
+    def __init__(self) -> None:
+        super().__init__("Failed to execute docker. Is it installed?")
 
 
 class NoSuchImageError(DockerError):
-    def __init__(self, image):
+    def __init__(self, image: str):
         self.image = image
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"No such image: {self.image}"
 
 
-def __wrap_docker_exec(func):
-    """Wrap a function to raise DockerExecuteError on ENOENT"""
-
-    def wrapper(*args, **kwargs):
-        try:
-            return func(*args, **kwargs)
-        except FileNotFoundError:
-            raise DockerExecuteError("Failed to execute docker. Is it installed?")
-
-    return wrapper
-
-
-call = __wrap_docker_exec(subprocess.call)
+def call(
+    args: _CMD,
+    stdin: _FILE,
+    stdout: _FILE,
+    stderr: _FILE,
+) -> int:
+    try:
+        return subprocess.call(args, stdin=stdin, stdout=stdout, stderr=stderr)
+    except FileNotFoundError as err:
+        raise DockerExecuteError() from err
 
 
-def _run_docker(*args, capture=False):
+def _run_docker(*args: str, capture: bool = False) -> subprocess.CompletedProcess[str]:
     """Run docker and raise DockerExecuteError on ENOENT"""
-    args = ["docker"] + list(args)
-    kw = dict(text=True)
+    docker_args = ["docker"] + list(args)
+    kw: Dict[str, Any] = dict(text=True)
     if capture:
         kw.update(capture_output=True)
 
-    return __wrap_docker_exec(subprocess.run)(args, **kw)
+    try:
+        return subprocess.run(docker_args, **kw)
+    except FileNotFoundError as err:
+        raise DockerExecuteError() from err
 
 
-def docker_inspect(image):
+def docker_inspect(image: str) -> dict:
     """Inspects a docker image
 
     Returns: Parsed JSON data
@@ -56,10 +62,12 @@ def docker_inspect(image):
             raise NoSuchImageError(image)
         raise DockerError(f"Failed to inspect image: {cp.stderr.strip()}")
 
-    return json.loads(cp.stdout)[0]
+    result = json.loads(cp.stdout)[0]
+    assert isinstance(result, dict)
+    return result
 
 
-def docker_pull(image):
+def docker_pull(image: str) -> None:
     """Pulls an image"""
     # If this fails, the default docker stdout/stderr looks good to the user.
     cp = _run_docker("pull", image)
@@ -67,7 +75,7 @@ def docker_pull(image):
         raise DockerError(f"Failed to pull image: {image}")
 
 
-def docker_inspect_or_pull(image):
+def docker_inspect_or_pull(image: str) -> dict:
     """Inspects a docker image, pulling it if it doesn't exist"""
     try:
         return docker_inspect(image)
@@ -77,7 +85,7 @@ def docker_inspect_or_pull(image):
         return docker_inspect(image)
 
 
-def get_images():
+def get_images() -> Sequence[str]:
     """Get the current list of docker images
 
     Returns: List of image names
@@ -101,28 +109,35 @@ def get_images():
     return cp.stdout.splitlines()
 
 
-def get_image_command(image):
-    """Gets the default command for an image"""
+def _get_image_config(image: str, key: str) -> Optional[Sequence[str]]:
     info = docker_inspect_or_pull(image)
     try:
-        return info["Config"]["Cmd"]
+        result = info["Config"][key]
     except KeyError as ke:
         raise DockerError(f"Failed to inspect image: JSON result missing key {ke}")
 
+    assert isinstance(result, (type(None), list))
+    return result
 
-def get_image_entrypoint(image):
+
+def get_image_command(image: str) -> Optional[Sequence[str]]:
+    """Gets the default command for an image"""
+    return _get_image_config(image, "Cmd")
+
+
+def get_image_entrypoint(image: str) -> Optional[Sequence[str]]:
     """Gets the image entrypoint"""
-    info = docker_inspect_or_pull(image)
-    try:
-        return info["Config"]["Entrypoint"]
-    except KeyError as ke:
-        raise DockerError(f"Failed to inspect image: JSON result missing key {ke}")
+    return _get_image_config(image, "Entrypoint")
 
 
-def make_vol_opt(hostdir, contdir, options=None):
+def make_vol_opt(
+    hostdir: str,
+    contdir: str,
+    options: Optional[Sequence[str]] = None,
+) -> str:
     """Generate a docker volume option"""
     vol = f"--volume={hostdir}:{contdir}"
-    if options != None:
+    if options is not None:
         if isinstance(options, str):
             options = (options,)
         vol += ":" + ",".join(options)
diff --git a/scuba/scuba.py b/scuba/scuba.py
index 4322826..58d6aba 100644
--- a/scuba/scuba.py
+++ b/scuba/scuba.py
@@ -1,58 +1,68 @@
+from __future__ import annotations
+import copy
+import dataclasses
 import os
 import shutil
 import sys
 import tempfile
-from collections.abc import Mapping
 from grp import getgrgid
 from io import StringIO
 from pwd import getpwuid
+from typing import cast, Any, Dict, Iterable, List, Optional, Sequence, Tuple, Union
+from typing import TextIO
 
 from .config import find_config, ScubaConfig, OverrideMixin
-from .config import ConfigError, ConfigNotFoundError
+from .config import ConfigError, ConfigNotFoundError, ScubaVolume
 from .dockerutil import get_image_command
 from .dockerutil import get_image_entrypoint
 from .dockerutil import make_vol_opt
 from .utils import shell_quote_cmd, flatten_list, get_umask, writeln
 
+VolumeTuple = Tuple[str, str, List[str]]
+
 
 class ScubaError(Exception):
     pass
 
 
 class ScubaDive:
+    context: ScubaContext
+    env_vars: Dict[str, str]
+    volumes: List[VolumeTuple]
+    options: List[str]
+    docker_args: List[str]
+
+    docker_cmd: List[str]
+
     def __init__(
         self,
-        user_command,
-        config,
-        top_path,
-        top_rel,
-        docker_args=None,
-        env=None,
-        as_root=False,
-        verbose=False,
-        image_override=None,
-        entrypoint=None,
-        shell_override=None,
-        keep_tempfiles=False,
+        user_command: List[str],
+        config: ScubaConfig,
+        top_path: str,
+        top_rel: str,
+        docker_args: Optional[List[str]] = None,
+        env: Optional[Dict[str, str]] = None,
+        as_root: bool = False,
+        verbose: bool = False,
+        image_override: Optional[str] = None,
+        entrypoint: Optional[str] = None,
+        shell_override: Optional[str] = None,
+        keep_tempfiles: bool = False,
     ):
-        env = env or {}
-        if not isinstance(env, Mapping):
-            raise ValueError("Argument env must be dict-like")
-
         self.as_root = as_root
         self.verbose = verbose
         self.entrypoint_override = entrypoint
         self.keep_tempfiles = keep_tempfiles
 
         # These will be added to docker run cmdline
-        self.env_vars = env
+        self.env_vars = env or {}
         self.volumes = []
         self.options = []
         self.docker_args = docker_args or []
-        self.workdir = None
+        self.workdir: Optional[str] = None
 
-        self.__scubadir_hostpath = None
-        self.__scubadir_contpath = None
+        self.__scubadir_hostpath: Optional[str] = None
+        self.__scubadir_contpath: Optional[str] = None
         self.config = config
 
         # Mount scuba root directory at the same path in the container...
@@ -68,8 +78,8 @@ class ScubaDive:
             self.context = ScubaContext.process_command(
                 cfg=self.config,
                 command=user_command,
-                image=image_override,
-                shell=shell_override,
+                image_override=image_override,
+                shell_override=shell_override,
             )
 
             # Apply environment vars from .scuba.yml
@@ -95,28 +105,31 @@ class ScubaDive:
             self._cleanup()
             raise
 
-    def __enter__(self):
+    def __enter__(self) -> ScubaDive:
         return self
 
-    def __exit__(self, *exc_info):
+    def __exit__(self, *exc_info: Any) -> None:
         self._cleanup()
 
-    def _cleanup(self):
+    def _cleanup(self) -> None:
         if self.__scubadir_hostpath and not self.keep_tempfiles:
             shutil.rmtree(self.__scubadir_hostpath)
 
-    def __str__(self):
+    def __str__(self) -> str:
         s = StringIO()
 
         indent = "  "
         level = 0
 
-        def writelist(name, vals):
+        def writehdr(name: str) -> None:
             writeln(s, f"{indent * level}{name}:")
+
+        def writelist(name: str, vals: Optional[Iterable[Any]]) -> None:
+            writehdr(name)
             for val in vals or ():
                 writeln(s, f"{indent * (level + 1)}{val}")
 
-        def writescl(name, val=""):
+        def writescl(name: str, val: Union[None, bool, float, str]) -> None:
             writeln(s, f"{indent * level}{name + ':':<14s}{val}")
 
         writeln(s, "ScubaDive")
@@ -133,9 +146,9 @@ class ScubaDive:
             "volumes", (f"{hp} => {cp} {opt}" for hp, cp, opt in self.__get_vol_opts())
         )
 
-        writescl("context")
+        writehdr("context")
         level += 1
-        writescl("script", self.context.script)
+        writelist("script", self.context.script)
         writescl("image", self.context.image)
         writelist("docker_args", self.context.docker_args)
         writelist("volumes", self.context.volumes)
@@ -143,22 +156,27 @@ class ScubaDive:
         return s.getvalue()
 
     @property
-    def is_remote_docker(self):
+    def is_remote_docker(self) -> bool:
         return "DOCKER_HOST" in os.environ
 
-    def add_env(self, name, val):
+    def add_env(self, name: str, val: Union[str, int]) -> None:
         """Add an environment variable to the docker run invocation"""
         if name in self.env_vars:
             raise KeyError(name)
-        self.env_vars[name] = val
+        self.env_vars[name] = str(val)
 
-    def add_volume(self, hostpath, contpath, options=None):
+    def add_volume(
+        self,
+        hostpath: str,
+        contpath: str,
+        options: Optional[List[str]] = None,
+    ) -> None:
         """Add a volume (bind-mount) to the docker run invocation"""
         if options is None:
             options = []
         self.volumes.append((hostpath, contpath, options))
 
-    def try_create_volumes(self):
+    def try_create_volumes(self) -> None:
         """Try to create non-existent host paths prior to docker run invocation
 
         This only creates user-defined volumes from configuration. The scubadir
@@ -182,14 +200,14 @@ class ScubaDive:
             except OSError as err:
                 raise ScubaError(f"Error creating volume host path: {err}") from err
 
-    def add_option(self, option):
+    def add_option(self, option: str) -> None:
         """Add another option to the docker run invocation"""
         self.options.append(option)
 
-    def set_workdir(self, workdir):
+    def set_workdir(self, workdir: str) -> None:
         self.workdir = workdir
 
-    def __locate_scubainit(self):
+    def __locate_scubainit(self) -> str:
         """Determine path to scubainit binary"""
         pkg_path = os.path.dirname(__file__)
 
@@ -198,13 +216,13 @@ class ScubaDive:
             raise ScubaError(f"scubainit not found at {scubainit_path!r}")
         return scubainit_path
 
-    def __make_scubadir(self):
+    def __make_scubadir(self) -> None:
         """Make temp directory where all ancillary files are bind-mounted"""
         self.__scubadir_hostpath = tempfile.mkdtemp(prefix="scubadir")
         self.__scubadir_contpath = "/.scuba"
         self.add_volume(self.__scubadir_hostpath, self.__scubadir_contpath)
 
-    def __setup_native_run(self):
+    def __setup_native_run(self) -> None:
         # These options are appended to mounted volume arguments
         # NOTE: This tells Docker to re-label the directory for compatibility
         # with SELinux. See `man docker-run` for more information.
@@ -272,50 +290,59 @@ class ScubaDive:
         else:
             ep = get_image_entrypoint(self.context.image)
             if ep:
-                self.docker_cmd = ep
+                self.docker_cmd = list(ep)
 
         # The user command is executed via a generated shell script
-        with self.open_scubadir_file("command.sh", "wt") as f:
-            self.docker_cmd += [self.context.shell, f.container_path]
-            writeln(f, "# Auto-generated from scuba")
-            writeln(f, "set -e")
+        with self.open_scubadir_file("command.sh") as cmd_script:
+            self.docker_cmd += [self.context.shell, cmd_script.container_path]
+            writeln(cmd_script, "# Auto-generated from scuba")
+            writeln(cmd_script, "set -e")
             for cmd in self.context.script:
-                writeln(f, cmd)
+                writeln(cmd_script, cmd)
 
-    def open_scubadir_file(self, name, mode):
-        """Opens a file in the 'scubadir'
+    def open_scubadir_file(self, name: str) -> Any:
+        """Opens a text file in the 'scubadir' for writing
 
         This file will automatically be bind-mounted into the container,
         at a path given by the 'container_path' property on the returned file object.
         """
+        assert self.__scubadir_hostpath is not None
+        assert self.__scubadir_contpath is not None
+
         path = os.path.join(self.__scubadir_hostpath, name)
         assert not os.path.exists(path)
 
         # Make any directories required
         os.makedirs(os.path.dirname(path), exist_ok=True)
 
-        f = open(path, mode)
+        # TODO: How to represent TextIO plus container_path attribute?
+        # Deriving from TextIO seemed to do nothing.
+        f: Any = open(path, "w")
         f.container_path = os.path.join(self.__scubadir_contpath, name)
+
         return f
 
-    def copy_scubadir_file(self, name, source):
+    def copy_scubadir_file(self, name: str, source: str) -> str:
         """Copies source into the scubadir
 
         Returns the container-path of the copied file
         """
+        assert self.__scubadir_hostpath is not None
+        assert self.__scubadir_contpath is not None
+
         dest = os.path.join(self.__scubadir_hostpath, name)
         assert not os.path.exists(dest)
         shutil.copy2(source, dest)
 
         return os.path.join(self.__scubadir_contpath, name)
 
-    def __generate_hook_script(self, name, shell):
+    def __generate_hook_script(self, name: str, shell: str) -> None:
         script = self.config.hooks.get(name)
         if not script:
             return
 
         # Generate the hook script, mount it into the container, and tell scubainit
-        with self.open_scubadir_file(f"hooks/{name}.sh", "wt") as f:
+        with self.open_scubadir_file(f"hooks/{name}.sh") as f:
             self.add_env(f"SCUBAINIT_HOOK_{name.upper()}", f.container_path)
 
             writeln(f, f"#!{shell}")
@@ -324,11 +351,11 @@ class ScubaDive:
             for cmd in script:
                 writeln(f, cmd)
 
-    def __get_vol_opts(self):
+    def __get_vol_opts(self) -> Iterable[VolumeTuple]:
         for hostpath, contpath, options in self.volumes:
             yield hostpath, contpath, options + self.vol_opts
 
-    def get_docker_cmdline(self):
+    def get_docker_cmdline(self) -> Sequence[str]:
         args = [
             "docker",
             "run",
@@ -369,79 +396,73 @@ class ScubaDive:
         return args
 
 
+@dataclasses.dataclass
 class ScubaContext:
-    def __init__(
-        self,
-        image=None,
-        script=None,
-        entrypoint=None,
-        environment=None,
-        shell=None,
-        docker_args=None,
-        volumes=None,
-    ):
-        self.image = image
-        self.script = script
-        self.as_root = False
-        self.entrypoint = entrypoint
-        self.environment = environment
-        self.shell = shell
-        self.docker_args = docker_args
-        self.volumes = volumes or {}
+    image: str
+    environment: Dict[str, str]  # key: value
+    volumes: Dict[str, ScubaVolume]
+    shell: str
+    docker_args: List[str]
+    script: Optional[List[str]] = None  # TODO: drop Optional?
+    entrypoint: Optional[str] = None
+    as_root: bool = False
 
     @classmethod
-    def process_command(cls, cfg, command, image=None, shell=None):
+    def process_command(
+        cls,
+        cfg: ScubaConfig,
+        command: Sequence[str],
+        image_override: Optional[str] = None,
+        shell_override: Optional[str] = None,
+    ) -> ScubaContext:
         """Processes a user command using aliases
 
         Arguments:
             cfg         ScubaConfig object
             command     A user command list (e.g. argv)
-            image       Override the image from .scuba.yml
-            shell       Override the shell from .scuba.yml
+            image_override       Override the image from .scuba.yml
+            shell_override       Override the shell from .scuba.yml
 
-        Returns: A ScubaContext object with the following attributes:
-            script: a list of command line strings
-            image: the docker image name to use
+        Returns: A ScubaContext object
         """
-        result = cls(
-            entrypoint=cfg.entrypoint,
-            environment=cfg.environment.copy(),
-            shell=cfg.shell,
-            docker_args=cfg.docker_args,
-            volumes=cfg.volumes,
-        )
+
+        image = None
+        shell = None
+        script = None
+        entrypoint = cfg.entrypoint
+        environment = copy.copy(cfg.environment)
+        docker_args = copy.copy(cfg.docker_args) or []
+        volumes: Dict[str, ScubaVolume] = copy.copy(cfg.volumes or {})
+        as_root = False
 
         if command:
             alias = cfg.aliases.get(command[0])
             if not alias:
                 # Command is not an alias; use it as-is.
-                result.script = [shell_quote_cmd(command)]
+                script = [shell_quote_cmd(command)]
             else:
                 # Using an alias
                 # Does this alias override the image and/or entrypoint?
                 if alias.image:
-                    result.image = alias.image
+                    image = alias.image
                 if alias.entrypoint is not None:
-                    result.entrypoint = alias.entrypoint
+                    entrypoint = alias.entrypoint
                 if alias.shell is not None:
-                    result.shell = alias.shell
+                    shell = alias.shell
                 if alias.as_root:
-                    result.as_root = True
+                    as_root = True
 
-                if (
-                    isinstance(alias.docker_args, OverrideMixin)
-                    or result.docker_args is None
-                ):
-                    result.docker_args = alias.docker_args
+                if isinstance(alias.docker_args, OverrideMixin) or docker_args is None:
+                    docker_args = cast(List[str], alias.docker_args)
                 elif alias.docker_args is not None:
-                    result.docker_args.extend(alias.docker_args)
+                    docker_args.extend(alias.docker_args)
 
                 if alias.volumes is not None:
-                    result.volumes.update(alias.volumes)
+                    volumes.update(alias.volumes)
 
                 # Merge/override the environment
                 if alias.environment:
-                    result.environment.update(alias.environment)
+                    environment.update(alias.environment)
 
                 if len(alias.script) > 1:
                     # Alias is a multiline script; no additional
@@ -450,28 +471,39 @@ class ScubaContext:
                         raise ConfigError(
                             "Additional arguments not allowed with multi-line aliases"
                         )
-                    result.script = alias.script
+                    script = alias.script
 
                 else:
                     # Alias is a single-line script; perform substituion
                     # and add user arguments.
-                    command.pop(0)
-                    result.script = [alias.script[0] + " " + shell_quote_cmd(command)]
+                    script = [alias.script[0] + " " + shell_quote_cmd(command[1:])]
 
-            result.script = flatten_list(result.script)
+            script = flatten_list(script)
 
         # If a shell was given on the CLI, it should override the shell set by
         # the alias or top-level config
-        if shell:
-            result.shell = shell
+        if shell_override:
+            shell = shell_override
 
         # If an image was given, it overrides what might have been set by an alias
-        if image:
-            result.image = image
+        if image_override:
+            image = image_override
 
         # If the image was still not set, then try to get it from the confg,
         # which will raise a ConfigError if it is not set
-        if not result.image:
-            result.image = cfg.image
-
-        return result
+        if image is None:
+            image = cfg.image
+
+        if shell is None:
+            shell = cfg.shell
+
+        return cls(
+            image=image,
+            script=script,
+            entrypoint=entrypoint,
+            environment=environment,
+            shell=shell,
+            docker_args=docker_args,
+            volumes=volumes,
+            as_root=as_root,
+        )
diff --git a/scuba/utils.py b/scuba/utils.py
index 22437ef..79c8978 100644
--- a/scuba/utils.py
+++ b/scuba/utils.py
@@ -1,14 +1,14 @@
-import errno
 import os
 from shlex import quote as shell_quote
 import string
+from typing import Iterable, TextIO, Tuple
 
 
-def shell_quote_cmd(cmdlist):
+def shell_quote_cmd(cmdlist: Iterable[str]) -> str:
     return " ".join(map(shell_quote, cmdlist))
 
 
-def format_cmdline(args, maxwidth=80):
+def format_cmdline(args: Iterable[str], maxwidth: int = 80) -> str:
     """Format args into a shell-quoted command line.
 
     The result will be wrapped to maxwidth characters where possible,
@@ -18,7 +18,7 @@ def format_cmdline(args, maxwidth=80):
     # Leave room for the space and backslash at the end of each line
     maxwidth -= 2
 
-    def lines():
+    def lines() -> Iterable[str]:
         line = ""
         for a in (shell_quote(a) for a in args):
             # If adding this argument will make the line too long,
@@ -39,7 +39,7 @@ def format_cmdline(args, maxwidth=80):
     return " \\\n".join(lines())
 
 
-def parse_env_var(s):
+def parse_env_var(s: str) -> Tuple[str, str]:
     """Parse an environment variable string
 
     Returns a key-value tuple
@@ -58,7 +58,7 @@ def parse_env_var(s):
     return (k, os.getenv(k, ""))
 
 
-def flatten_list(x):
+def flatten_list(x: list) -> list:
     if not isinstance(x, list):
         raise ValueError("argument is not a list")
     result = []
@@ -71,18 +71,18 @@ def flatten_list(x):
     return result
 
 
-def get_umask():
+def get_umask() -> int:
     # Same logic as bash/builtins/umask.def
     val = os.umask(0o22)
     os.umask(val)
     return val
 
 
-def writeln(f, line):
+def writeln(f: TextIO, line: str) -> None:
     f.write(line + "\n")
 
 
-def expand_env_vars(in_str):
+def expand_env_vars(in_str: str) -> str:
     """Expand environment variables in a string
 
     Can raise `KeyError` if a variable is referenced but not defined, similar to
diff --git a/scuba/version.py b/scuba/version.py
index a056f57..3fbcb79 100644
--- a/scuba/version.py
+++ b/scuba/version.py
@@ -1,4 +1,6 @@
+import dataclasses
 from os.path import abspath, dirname, exists, join, normpath
+import subprocess
 import sys
 
 # This logic has been adapted from that of PyInstaller
@@ -17,38 +19,48 @@ BASE_VERSION = "2.10.1"
 git_archive_rev = "$Format:%h$"
 
 
-def git_describe():
-    from subprocess import check_call, check_output
+@dataclasses.dataclass(frozen=True)
+class GitDescribe:
+    tag: str
+    commits: int
+    rev: str
 
-    # Get the version from the local Git repository
-    check_call(["git", "update-index", "-q", "--refresh"], cwd=PROJPATH)
-
-    desc = check_output(["git", "describe", "--long", "--dirty", "--tag"], cwd=PROJPATH)
-    desc = desc.decode("utf-8").strip()
-
-    tag, commits, rev = desc.split("-", 2)
-    tag = tag.lstrip("v")
-    commits = int(commits)
 
-    return tag, commits, rev
-
-
-def get_version():
+def git_describe() -> GitDescribe:
+    # Get the version from the local Git repository
+    subprocess.run(["git", "update-index", "-q", "--refresh"], check=True, cwd=PROJPATH)
+
+    result = subprocess.run(
+        ["git", "describe", "--long", "--dirty", "--tag"],
+        check=True,
+        cwd=PROJPATH,
+        stdout=subprocess.PIPE,
+        encoding="utf-8",
+    )
+    parts = result.stdout.rstrip().split("-", 2)
+    return GitDescribe(
+        tag=parts[0].lstrip("v"),
+        commits=int(parts[1]),
+        rev=parts[2],
+    )
+
+
+def get_version() -> str:
     # Git repo
     # If a local git repository is present, use `git describe` to provide a rich version
     gitdir = normpath(join(PROJPATH, ".git"))
     if exists(gitdir):
-        tag, commits, rev = git_describe()
+        desc = git_describe()
 
         # Ensure the base version matches the Git tag
-        if tag != BASE_VERSION:
+        if desc.tag != BASE_VERSION:
             raise Exception("Git revision different from base version")
 
         # No local version if we're on a tag
-        if commits == 0 and not rev.endswith("dirty"):
+        if desc.commits == 0 and not desc.rev.endswith("dirty"):
             return BASE_VERSION
 
-        return f"{BASE_VERSION}+{commits}-{rev}"
+        return f"{BASE_VERSION}+{desc.commits}-{desc.rev}"
 
     # Git archive
     # If this was produced via `git archive`, we'll use the version it provides
diff --git a/static_analysis.sh b/static_analysis.sh
new file mode 100755
index 0000000..acdab05
--- /dev/null
+++ b/static_analysis.sh
@@ -0,0 +1,4 @@
+#!/bin/bash
+set -e
+python3 -m mypy --version
+python3 -m mypy


Test patch:
diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml
index 3d49be7..08ea942 100644
--- a/.github/workflows/build-test.yml
+++ b/.github/workflows/build-test.yml
@@ -38,6 +38,10 @@ jobs:
 
         ci/test_setup.sh
 
+    - name: Lint
+      run: |
+        mypy
+
     - name: Build
       run: |
         python -m build
diff --git a/tests/test_scuba.py b/tests/test_scuba.py
index fda1847..6864102 100644
--- a/tests/test_scuba.py
+++ b/tests/test_scuba.py
@@ -182,7 +182,9 @@ class TestScubaContext:
         cfg = ScubaConfig(
             image="test_image",
         )
-        result = ScubaContext.process_command(cfg, [], image=override_image_name)
+        result = ScubaContext.process_command(
+            cfg, [], image_override=override_image_name
+        )
         assert result.image == override_image_name
 
     def test_process_command_image_override_missing(self):
@@ -190,7 +192,9 @@ class TestScubaContext:
         override_image_name = "override_image"
 
         cfg = ScubaConfig()
-        result = ScubaContext.process_command(cfg, [], image=override_image_name)
+        result = ScubaContext.process_command(
+            cfg, [], image_override=override_image_name
+        )
         assert result.image == override_image_name
 
     def test_process_command_image_override_alias(self):
@@ -208,7 +212,9 @@ class TestScubaContext:
                 ),
             )
         )
-        result = ScubaContext.process_command(cfg, [], image=override_image_name)
+        result = ScubaContext.process_command(
+            cfg, [], image_override=override_image_name
+        )
         assert result.image == override_image_name
 
     def test_env_merge(self):