dbapi: add some typing

Signed-off-by: Andrew Udvare <[email protected]>

lib/portage/dbapi/__init__.py

@@ -5,6 +5,7 @@
 
 import re
 import warnings
+from typing import Any, Dict, List, Optional, Sequence, Tuple
 
 import portage
 
@@ -29,7 +30,7 @@
 
 class dbapi:
     _category_re = re.compile(r"^\w[-.+\w]*$", re.UNICODE)
-    _categories = None
+    _categories: Optional[Tuple[str, ...]] = None
     _use_mutable = False
     _known_keys = frozenset(auxdbkeys)
     _pkg_str_aux_keys = ("EAPI", "KEYWORDS", "SLOT", "repository")
@@ -38,7 +39,7 @@ def __init__(self):
         pass
 
     @property
-    def categories(self):
+    def categories(self) -> Tuple[str, ...]:
         """
         Use self.cp_all() to generate a category list. Mutable instances
         can delete the self._categories attribute in cases when the cached
@@ -52,11 +53,11 @@ def categories(self):
     def close_caches(self):
         pass
 
-    def cp_list(self, cp, use_cache=1):
+    def cp_list(self, cp: str, use_cache: int = 1) -> Any:
         raise NotImplementedError(self)
 
     @staticmethod
-    def _cmp_cpv(cpv1, cpv2):
+    def _cmp_cpv(cpv1, cpv2) -> int:
         result = vercmp(cpv1.version, cpv2.version)
         if result == 0 and cpv1.build_time is not None and cpv2.build_time is not None:
             result = (cpv1.build_time > cpv2.build_time) - (
@@ -65,7 +66,7 @@ def _cmp_cpv(cpv1, cpv2):
         return result
 
     @staticmethod
-    def _cpv_sort_ascending(cpv_list):
+    def _cpv_sort_ascending(cpv_list: Sequence[Any]) -> None:
         """
         Use this to sort self.cp_list() results in ascending
         order. It sorts in place and returns None.
@@ -76,7 +77,7 @@ def _cpv_sort_ascending(cpv_list):
             # dict to map strings back to their original values.
             cpv_list.sort(key=cmp_sort_key(dbapi._cmp_cpv))
 
-    def cpv_all(self):
+    def cpv_all(self) -> List[str]:
         """Return all CPVs in the db
         Args:
                 None
@@ -93,16 +94,16 @@ def cpv_all(self):
             cpv_list.extend(self.cp_list(cp))
         return cpv_list
 
-    def cp_all(self, sort=False):
+    def cp_all(self, sort: bool = False) -> List[str]:
         """Implement this in a child class
         Args
                 sort - return sorted results
         Returns:
                 A list of strings 1 per CP in the datastore
         """
-        return NotImplementedError
+        raise NotImplementedError
 
-    def aux_get(self, mycpv, mylist, myrepo=None):
+    def aux_get(self, mycpv: str, mylist: str, myrepo: Optional[str] = None) -> List[str]:
         """Return the metadata keys in mylist for mycpv
         Args:
                 mycpv - "sys-apps/foo-1.0"
@@ -114,7 +115,7 @@ def aux_get(self, mycpv, mylist, myrepo=None):
         """
         raise NotImplementedError
 
-    def aux_update(self, cpv, metadata_updates):
+    def aux_update(self, cpv: str, metadata_updates: Dict[str, Any]) -> None:
         """
         Args:
           cpv - "sys-apps/foo-1.0"
@@ -124,7 +125,7 @@ def aux_update(self, cpv, metadata_updates):
         """
         raise NotImplementedError
 
-    def match(self, origdep, use_cache=1):
+    def match(self, origdep: str, use_cache: int = 1):
         """Given a dependency, try to find packages that match
         Args:
                 origdep - Depend atom
@@ -138,7 +139,7 @@ def match(self, origdep, use_cache=1):
             self._iter_match(mydep, self.cp_list(mydep.cp, use_cache=use_cache))
         )
 
-    def _iter_match(self, atom, cpv_iter):
+    def _iter_match(self, atom: str, cpv_iter):
         cpv_iter = iter(match_from_list(atom, cpv_iter))
         if atom.repo:
             cpv_iter = self._iter_match_repo(atom, cpv_iter)

lib/portage/dbapi/porttree.py

@@ -49,6 +49,7 @@
 
 import collections
 from collections import OrderedDict
+from typing import List, Optional, Sequence, Type, Tuple, Union
 from urllib.parse import urlparse
 
 
@@ -435,7 +436,7 @@ def findLicensePath(self, license_name):
                 return license_path
         return None
 
-    def findname(self, mycpv, mytree=None, myrepo=None):
+    def findname(self, mycpv: str, mytree: Optional[str] = None, myrepo: Optional[str] = None) -> str:
         return self.findname2(mycpv, mytree, myrepo)[0]
 
     def getRepositoryPath(self, repository_id):
@@ -494,7 +495,12 @@ def getIgnoredRepos(self):
         """
         return self.settings.repositories.ignored_repos
 
-    def findname2(self, mycpv, mytree=None, myrepo=None):
+    def findname2(
+            self,
+            mycpv: str,
+            mytree: Optional[str] = None,
+            myrepo: Optional[str] = None,
+        ) -> Union[Tuple[None, int], Tuple[str, str], Tuple[str, None]]:
         """
         Returns the location of the CPV, and what overlay it was in.
         Searches overlays first, then PORTDIR; this allows us to return the first
@@ -643,7 +649,13 @@ def _pull_valid_cache(self, cpv, ebuild_path, repo_path):
 
         return (metadata, ebuild_hash)
 
-    def aux_get(self, mycpv, mylist, mytree=None, myrepo=None):
+    def aux_get(
+            self,
+            mycpv: str,
+            mylist: Sequence[str],
+            mytree: Optional[str] = None,
+            myrepo: Optional[str] = None,
+        ) -> List[str]:
         "stub code for returning auxilliary db information, such as SLOT, DEPEND, etc."
         'input: "sys-apps/foo-1.0",["SLOT","DEPEND","HOMEPAGE"]'
         'return: ["0",">=sys-libs/bar-1.0","http://www.foo.com"] or raise PortageKeyError if error'
@@ -1200,12 +1212,12 @@ def melt(self):
 
     def xmatch(
         self,
-        level,
-        origdep,
-        mydep=DeprecationWarning,
-        mykey=DeprecationWarning,
-        mylist=DeprecationWarning,
-    ):
+        level: str,
+        origdep: str,
+        mydep: Type[DeprecationWarning] = DeprecationWarning,
+        mykey: Type[DeprecationWarning] = DeprecationWarning,
+        mylist: Type[DeprecationWarning] = DeprecationWarning,
+    ) -> Union[Sequence[str], str]:
         """
         Caching match function.
 
@@ -1381,7 +1393,7 @@ async def async_xmatch(self, level, origdep, loop=None):
 
         return myval
 
-    def match(self, mydep, use_cache=1):
+    def match(self, mydep: str, use_cache: int = 1) -> Union[Sequence[str], str]:
         return self.xmatch("match-visible", mydep)
 
     def gvisible(self, mylist):