Improve mypy precision by moving generics to subclass

It only matters for construction. After construction, we can erase
the type by upcasting, keeping the types simple and improving mypy
precision.

nixops/locks/__init__.py

@@ -1,11 +1,33 @@
 from typing import TypeVar, Type
 from typing_extensions import Protocol
 
+"""
+Interface to a lock driver.
+
+An implementation should inherit from LockDriver in order to for a plugin to be
+able to integrate it.
+"""
+# This separation was introduced to hide the LockOptions details from the
+# LockInterface type. It only matters for construction and clients don't have
+# to know about it.
+class LockInterface(Protocol):
+    # lock: acquire a lock.
+    # Note: no arguments will be passed over kwargs. Making it part of
+    # the type definition allows adding new arguments later.
+    def lock(self, **kwargs) -> None:
+        raise NotImplementedError
+
+    # unlock: release the lock.
+    # Note: no arguments will be passed over kwargs. Making it part of
+    # the type definition allows adding new arguments later.
+    def unlock(self, **kwargs) -> None:
+        raise NotImplementedError
+
 
 LockOptions = TypeVar("LockOptions")
 
 
-class LockDriver(Protocol[LockOptions]):
+class LockDriver(LockInterface, Protocol[LockOptions]):
     # Hack: Make T a mypy invariant. According to PEP-0544, a
     # Protocol[T] whose T is only used in function arguments and
     # returns is "de-facto covariant".
@@ -26,15 +48,3 @@ def options(**kwargs) -> LockOptions:
 
     def __init__(self, args: LockOptions) -> None:
         raise NotImplementedError
-
-    # lock: acquire a lock.
-    # Note: no arguments will be passed over kwargs. Making it part of
-    # the type definition allows adding new arguments later.
-    def lock(self, **kwargs) -> None:
-        raise NotImplementedError
-
-    # unlock: release the lock.
-    # Note: no arguments will be passed over kwargs. Making it part of
-    # the type definition allows adding new arguments later.
-    def unlock(self, **kwargs) -> None:
-        raise NotImplementedError

nixops/script_defs.py

@@ -2,8 +2,8 @@
 
 from nixops.nix_expr import py2nix
 from nixops.parallel import run_tasks
-from nixops.storage import StorageBackend
-from nixops.locks import LockDriver
+from nixops.storage import StorageBackend, StorageInterface
+from nixops.locks import LockDriver, LockInterface
 
 import contextlib
 import nixops.statefile
@@ -75,8 +75,8 @@ def deployment(
         yield depl
 
 
-def get_lock(network: NetworkEval) -> LockDriver:
-    lock: LockDriver
+def get_lock(network: NetworkEval) -> LockInterface:
+    lock: LockInterface
     lock_class: Type[LockDriver]
     lock_drivers = PluginManager.lock_drivers()
     try:
@@ -113,14 +113,14 @@ def network_state(
         )
         raise Exception("Missing storage provider plugin.")
 
-    lock: Optional[LockDriver]
+    lock: Optional[LockInterface]
     if doLock:
         lock = get_lock(network)
     else:
         lock = None
 
     storage_class_options = storage_class.options(**network.storage.configuration)
-    storage: StorageBackend = storage_class(storage_class_options)
+    storage: StorageInterface = storage_class(storage_class_options)
 
     with TemporaryDirectory("nixops") as statedir:
         statefile = statedir + "/state.nixops"

nixops/statefile.py

@@ -5,12 +5,12 @@
 import sqlite3
 import sys
 import threading
-from typing import Any, Optional, List, Type
+from typing import Any, Optional, List, Type, TypeVar, Generic
 from types import TracebackType
 import re
 
 import nixops.deployment
-from nixops.locks import LockDriver
+from nixops.locks import LockInterface
 
 
 class Connection(sqlite3.Connection):
@@ -81,14 +81,17 @@ def get_default_state_file() -> str:
     )
 
 
+LockOptions = TypeVar("LockOptions")
+
+
 class StateFile(object):
     """NixOps state file."""
 
     current_schema = 3
-    lock: Optional[LockDriver]
+    lock: Optional[LockInterface]
 
     def __init__(
-        self, db_file: str, writable: bool, lock: Optional[LockDriver] = None
+        self, db_file: str, writable: bool, lock: Optional[LockInterface] = None
     ) -> None:
         self.db_file: str = db_file
         self.lock = lock

nixops/storage/__init__.py

@@ -3,6 +3,36 @@
 
 from typing_extensions import Protocol
 
+"""
+Interface to a storage driver.
+
+An implementation should inherit from LockDriver in order to for a plugin to be
+able to integrate it.
+"""
+# This separation was introduced to hide the T (options) details from the
+# StorageInterface type. It only matters for construction and clients don't have
+# to know about it.
+class StorageInterface(Protocol):
+    # fetchToFile: download the state file to the local disk.
+    # Note: no arguments will be passed over kwargs. Making it part of
+    # the type definition allows adding new arguments later.
+    def fetchToFile(self, path: str, **kwargs) -> None:
+        raise NotImplementedError
+
+    # onOpen: receive the StateFile object for last-minute, backend
+    # specific changes to the state file.
+    # Note: no arguments will be passed over kwargs. Making it part of
+    # the type definition allows adding new arguments later.
+    def onOpen(self, sf: nixops.statefile.StateFile, **kwargs) -> None:
+        pass
+
+    # uploadFromFile: upload the new version of the state file
+    # Note: no arguments will be passed over kwargs. Making it part of
+    # the type definition allows adding new arguments later.
+    def uploadFromFile(self, path: str, **kwargs) -> None:
+        raise NotImplementedError
+
+
 if TYPE_CHECKING:
     import nixops.statefile
 
@@ -11,7 +41,7 @@
 StorageArgValues = Mapping[str, Any]
 
 
-class StorageBackend(Protocol[T]):
+class StorageBackend(StorageInterface, Protocol[T]):
     # Hack: Make T a mypy invariant. According to PEP-0544, a
     # Protocol[T] whose T is only used in function arguments and
     # returns is "de-facto covariant".
@@ -32,22 +62,3 @@ def options(**kwargs) -> T:
 
     def __init__(self, args: T) -> None:
         raise NotImplementedError
-
-    # fetchToFile: download the state file to the local disk.
-    # Note: no arguments will be passed over kwargs. Making it part of
-    # the type definition allows adding new arguments later.
-    def fetchToFile(self, path: str, **kwargs) -> None:
-        raise NotImplementedError
-
-    # onOpen: receive the StateFile object for last-minute, backend
-    # specific changes to the state file.
-    # Note: no arguments will be passed over kwargs. Making it part of
-    # the type definition allows adding new arguments later.
-    def onOpen(self, sf: nixops.statefile.StateFile, **kwargs) -> None:
-        pass
-
-    # uploadFromFile: upload the new version of the state file
-    # Note: no arguments will be passed over kwargs. Making it part of
-    # the type definition allows adding new arguments later.
-    def uploadFromFile(self, path: str, **kwargs) -> None:
-        raise NotImplementedError