Safe Annotations

Author: chadrik

src/rez/package_repository.py

@@ -27,7 +27,7 @@
 PackageFamilyResourceT = TypeVar("PackageFamilyResourceT", bound=PackageFamilyResource)
 
 
-def get_package_repository_types():
+def get_package_repository_types() -> list[str]:
     """Returns the available package repository implementations."""
     return plugin_manager.get_plugins('package_repository')
 
@@ -116,7 +116,7 @@ def clear_caches(self) -> None:
         self.pool.clear_caches()
 
     @cached_property
-    def uid(self) -> tuple[str, str]:
+    def uid(self) -> tuple:
         """Returns a unique identifier for this repository.
 
         This must be a persistent identifier, for example a filepath, or
@@ -435,7 +435,7 @@ def get_last_release_time(self, package_family_resource: PackageFamilyResourceT
         """
         return 0
 
-    def make_resource_handle(self, resource_key: str, **variables) -> ResourceHandle:
+    def make_resource_handle(self, resource_key: str, **variables: Any) -> ResourceHandle:
         """Create a `ResourceHandle`
 
         Nearly all `ResourceHandle` creation should go through here, because it
@@ -460,14 +460,14 @@ def make_resource_handle(self, resource_key: str, **variables) -> ResourceHandle
         return ResourceHandle(resource_key, variables)
 
     @overload
-    def get_resource(self, resource_key: type[ResourceT], **variables) -> ResourceT:
+    def get_resource(self, resource_key: type[ResourceT], **variables: Any) -> ResourceT:
         pass
 
     @overload
-    def get_resource(self, resource_key: str, **variables) -> Resource:
+    def get_resource(self, resource_key: str, **variables: Any) -> Resource:
         pass
 
-    def get_resource(self, resource_key: str | type[Resource], **variables) -> Resource:
+    def get_resource(self, resource_key: str | type[Resource], **variables: Any) -> Resource:
         """Get a resource.
 
         Attempts to get and return a cached version of the resource if
@@ -528,7 +528,7 @@ def get_package_payload_path(self, package_name: str, package_version: str | Ver
         """
         raise NotImplementedError
 
-    def _uid(self) -> tuple[str, str]:
+    def _uid(self) -> tuple:
         """Unique identifier implementation.
 
         You may need to provide your own implementation. For example, consider
@@ -624,7 +624,7 @@ def are_same(self, path_1: str, path_2: str) -> bool:
         return (repo_1.uid == repo_2.uid)
 
     def get_resource(self, resource_key: str, repository_type: str,
-                     location: str, **variables) -> Resource:
+                     location: str, **variables: Any) -> Resource:
         """Get a resource.
 
         Attempts to get and return a cached version of the resource if
@@ -671,7 +671,7 @@ def clear_caches(self) -> None:
         self.repositories.clear()
         self.pool.clear_caches()
 
-    def _get_repository(self, path: str, **repo_args) -> PackageRepository:
+    def _get_repository(self, path: str, **repo_args: Any) -> PackageRepository:
         repo_type, location = path.split('@', 1)
         cls = plugin_manager.get_plugin_class('package_repository', repo_type, PackageRepository)
         repo = cls(location, self.pool, **repo_args)

src/rez/rex_bindings.py

@@ -13,7 +13,7 @@
 from __future__ import annotations
 
 from rez.rex import ActionInterpreter
-from rez.version import VersionRange
+from rez.version import VersionRange, Version, VersionToken
 from rez.version import Requirement
 
 
@@ -57,7 +57,7 @@ class VersionBinding(Binding):
         >>> v.as_tuple():
         (1, 2, '3alpha')
     """
-    def __init__(self, version) -> None:
+    def __init__(self, version: Version) -> None:
         super(VersionBinding, self).__init__()
         self.__version = version
 
@@ -86,8 +86,8 @@ def __getitem__(self, i):
         except IndexError:
             return None
 
-    def __getitem(self, i):
-        def _convert(t):
+    def __getitem(self, i: int | slice):
+        def _convert(t: VersionToken) -> str | int:
             s = str(t)
             if s.isdigit() and (s[0] != '0' or s == '0'):
                 return int(s)

src/rez/utils/memcached.py

@@ -39,7 +39,7 @@ def __bool__(self) -> bool:
 
     logger = config.debug_printer("memcache")
 
-    def __init__(self, servers, debug: bool = False) -> None:
+    def __init__(self, servers: str | list[str], debug: bool = False) -> None:
         """Create a memcached client.
 
         Args:
@@ -58,7 +58,7 @@ def __bool__(self) -> bool:
         return bool(self.servers)
 
     @property
-    def client(self):
+    def client(self) -> Client_:
         """Get the native memcache client.
 
         Returns:
@@ -68,7 +68,7 @@ def client(self):
             self._client = Client_(self.servers)
         return self._client
 
-    def test_servers(self):
+    def test_servers(self) -> set[str]:
         """Test that memcached servers are servicing requests.
 
         Returns:
@@ -83,7 +83,7 @@ def test_servers(self):
                 responders.add(server)
         return responders
 
-    def set(self, key, val, time: int=0, min_compress_len: int=0) -> None:
+    def set(self, key: str, val: Any, time: int=0, min_compress_len: int=0) -> None:
         """See memcache.Client."""
         if not self.servers:
             return
@@ -98,7 +98,7 @@ def set(self, key, val, time: int=0, min_compress_len: int=0) -> None:
                         min_compress_len=min_compress_len)
         self.logger("SET: %s", key)
 
-    def get(self, key):
+    def get(self, key: str) -> Any | Client._Miss:
         """See memcache.Client.
 
         Returns:
@@ -122,7 +122,7 @@ def get(self, key):
         self.logger("MISS: %s", key)
         return self.miss
 
-    def delete(self, key) -> None:
+    def delete(self, key: str) -> None:
         """See memcache.Client."""
         if self.servers:
             key = self._qualified_key(key)
@@ -149,7 +149,7 @@ def flush(self, hard: bool = False) -> None:
                 tag = "flushed" + tag
             self.current = tag
 
-    def get_stats(self):
+    def get_stats(self) -> list[tuple]:
         """Get server statistics.
 
         Returns:
@@ -167,7 +167,7 @@ def disconnect(self) -> None:
             self._client.disconnect_all()
         # print("Disconnected memcached client %s" % str(self))
 
-    def _qualified_key(self, key) -> str:
+    def _qualified_key(self, key: str) -> str:
         """
         Qualify cache key so that:
         * changes to schemas don't break compatibility (cache_interface_version)
@@ -181,15 +181,15 @@ def _qualified_key(self, key) -> str:
             key
         )
 
-    def _get_stats(self, stat_args=None):
+    def _get_stats(self, stat_args=None) -> list[tuple]:
         return self.client.get_stats(stat_args=stat_args)
 
     @classmethod
-    def _key_hash(cls, key):
+    def _key_hash(cls, key: str) -> str:
         return md5(key.encode("utf-8")).hexdigest()
 
     @classmethod
-    def _debug_key_hash(cls, key):
+    def _debug_key_hash(cls, key: str) -> str:
         import re
         h = cls._key_hash(key)[:16]
         value = "%s:%s" % (h, key)
@@ -213,7 +213,7 @@ def acquire(self, servers, debug: bool = False) -> tuple[Client, tuple[tuple, bo
             self.clients[key] = [client, 1]
             return client, key
 
-    def release(self, key) -> None:
+    def release(self, key: tuple[tuple, bool]) -> None:
         entry = self.clients.get(key)
         assert entry
 
@@ -405,5 +405,5 @@ def forget() -> None:
 
 
 class DoNotCache(object):
-    def __init__(self, result) -> None:
+    def __init__(self, result: Any) -> None:
         self.result = result

src/rez/version/_version.py

@@ -10,7 +10,7 @@
 import copy
 import string
 import re
-from typing import cast, Any, Callable, Generic, Iterable, TypeVar, TYPE_CHECKING
+from typing import cast, Any, Callable, Generic, Iterable, TypeVar, TYPE_CHECKING, overload
 
 if TYPE_CHECKING:
     from typing_extensions import Self
@@ -410,7 +410,15 @@ def as_tuple(self) -> tuple[str, ...]:
     def __len__(self) -> int:
         return len(self.tokens or [])
 
+    @overload
     def __getitem__(self, index: int) -> VersionToken:
+        pass
+
+    @overload
+    def __getitem__(self, index: slice) -> list[VersionToken]:
+        pass
+
+    def __getitem__(self, index: int | slice) -> VersionToken | list[VersionToken]:
         try:
             return (self.tokens or [])[index]
         except IndexError:

src/rezplugins/package_repository/filesystem.py

@@ -59,7 +59,7 @@
 format_version = 2
 
 
-def check_format_version(filename, data) -> None:
+def check_format_version(filename: str, data: dict[str, Any]) -> None:
     format_version_ = data.pop("format_version", None)
 
     if format_version_ is not None:
@@ -165,7 +165,7 @@ def state_handle(self) -> int | None:
         return None
 
     @property
-    def base(self) -> str | None:
+    def base(self) -> str:
         # Note: '_redirected_base' is a special attribute set by the build
         # process in order to perform pre-install/release package testing. See
         # `LocalBuildProcess._run_tests()`
@@ -497,7 +497,7 @@ class FileSystemPackageRepository(
     def name(cls) -> str:
         return "filesystem"
 
-    def __init__(self, location, resource_pool, disable_memcache=None,
+    def __init__(self, location: str, resource_pool: ResourcePool, disable_memcache: bool | None = None,
                  disable_pkg_ignore: bool = False) -> None:
         """Create a filesystem package repository.
 
@@ -562,7 +562,7 @@ def __init__(self, location, resource_pool, disable_memcache=None,
             )
             self._get_version_dirs = decorator2(self._get_version_dirs)
 
-    def _uid(self):
+    def _uid(self) -> tuple:
         t = ["filesystem", self.location]
         if os.path.exists(self.location):
             st = os.stat(self.location)
@@ -720,7 +720,7 @@ def unignore_package(self, pkg_name: str, pkg_version: Version) -> int:
         else:
             return -1
 
-    def remove_package(self, pkg_name: str, pkg_version) -> bool:
+    def remove_package(self, pkg_name: str, pkg_version: Version) -> bool:
         # ignore it first, so a partially deleted pkg is not visible
         i = self.ignore_package(pkg_name, pkg_version)
         if i == -1:
@@ -980,7 +980,7 @@ def _copy(self, **kwargs) -> Self:
         return repo_copy
 
     @contextmanager
-    def _lock_package(self, package_name: str, package_version: str | Version | None = None):
+    def _lock_package(self, package_name: str, package_version: str | Version | None = None) -> Iterator[None]:
         from rez.vendor.lockfile import NotLocked
 
         if _settings.file_lock_type == 'default':
@@ -1079,7 +1079,7 @@ def _get_version_dirs__key(self, root: str) -> str:
 
     def _get_version_dirs(self, root: str) -> list[str]:
         # Ignore a version if there is a .ignore<version> file next to it
-        def ignore_dir(name):
+        def ignore_dir(name: str) -> bool:
             if self.disable_pkg_ignore:
                 return False
             else:
@@ -1552,5 +1552,5 @@ def _delete_stale_build_tagfiles(self, family_path: str) -> None:
                     os.remove(tagfilepath)
 
 
-def register_plugin():
+def register_plugin() -> type[FileSystemPackageRepository]:
     return FileSystemPackageRepository