mypy plugin and typing info for koji.MultiCallSession (#173)

provides a mypy plugin that allows for static analysis of the methods of koji.MultiCallSession

kojismokydingo/types.py

@@ -31,7 +31,8 @@
     ClientSession, PathInfo, )
 from optparse import Values
 from typing import (
-    Any, Callable, Dict, Iterable, List, Optional, Tuple, Union, )
+    Any, Callable, Dict, Generic, Iterable, List, Optional,
+    Tuple, TypeVar, Union, )
 
 
 try:
@@ -40,14 +41,18 @@
     # Python < 3.10 doesn't have typing.TypedDict
     from typing_extensions import TypedDict
 
-
 try:
     from typing import Protocol
 except ImportError:
     # Python < 3.8 doesn't have typing.Protocol
     class Protocol:  # type: ignore
         ...
 
+try:
+    from contextlib import AbstractContextManager as ContextManager
+except ImportError:
+    from typing import ContextManager
+
 
 __all__ = (
     "ArchiveInfo",

kojismokydingo/users.py

@@ -64,46 +64,40 @@ def collect_userstats(
 
     userinfo = as_userinfo(session, user)
 
-    with session.multicall() as mc:  # type: ignore
-        calls = (
-            ('build_count',
-             mc.listBuilds(userID=userinfo['id'],
-                           queryOpts={'countOnly': True})),
-
-            ('last_build',
-             mc.listBuilds(userID=userinfo['id'],
-                           queryOpts={'limit': 1, 'order': '-build_id'})),
-
-            ('package_count',
-             mc.listPackages(userID=userinfo['id'], with_dups=True,
-                             queryOpts={'countOnly': True})),
-
-            ('task_count',
-             mc.listTasks(opts={'owner': userinfo['id'], 'parent': None},
-                          queryOpts={'countOnly': True})),
-
-            ('last_task',
-             mc.listTasks(opts={'owner': userinfo['id'], 'parent': None},
-                          queryOpts={'limit': 1, 'order': '-id'})),
-        )
-
-    stats = {k: v.result for k, v in calls}
-
-    # unwrap the last_build
-    lst = stats['last_build']
-    if lst:
-        stats['last_build'] = lst[0]
-    else:
-        stats['last_build'] = None
-
-    # and also the last_task
-    lst = stats['last_task']
-    if lst:
-        stats['last_task'] = lst[0]
-    else:
-        stats['last_task'] = None
-
-    return cast(UserStatistics, stats)
+    with session.multicall() as mc:
+        build_count = mc.listBuilds(userID=userinfo['id'],
+                                    queryOpts={'countOnly': True})
+
+        package_count = mc.listPackages(userID=userinfo['id'],
+                                        with_dups=True,
+                                        queryOpts={'countOnly': True})
+
+        task_count = mc.listTasks(opts={'owner': userinfo['id'],
+                                        'parent': None},
+                                  queryOpts={'countOnly': True})
+
+        last_build = mc.listBuilds(userID=userinfo['id'],
+                                   queryOpts={'limit': 1,
+                                              'order': '-build_id'})
+
+        last_task = mc.listTasks(opts={'owner': userinfo['id'],
+                                       'parent': None},
+                                 queryOpts={'limit': 1,
+                                            'order': '-id'})
+
+    stats: UserStatistics = {
+        # I haven't figured out how to indicate that the queryOpts
+        # with countOnly set changes the return type to int.
+        'build_count': build_count.result,      # type: ignore
+        'package_count': package_count.result,  # type: ignore
+        'task_count': task_count.result,        # type: ignore
+
+        # just need to unwrap the list if any
+        'last_build': last_build.result[0] if last_build.result else None,
+        'last_task': last_task.result[0] if last_task.result else None,
+    }
+
+    return stats
 
 
 def get_user_groups(

stubs/koji/__init__.pyi → mypy/koji/__init__.pyi

@@ -25,20 +25,37 @@ calls are being used correctly.
 """
 
 
+from __future__ import annotations
+
+
 from configparser import ConfigParser, RawConfigParser
 from datetime import datetime
 from typing import (
-    Any, Dict, Iterable, List, Optional, Tuple, TypedDict, TypeVar,
-    Union, Set, overload, )
+    Any, Dict, Generic, Iterable, List, Optional, Tuple,
+    TypedDict, TypeVar, Union, Set, overload, )
 from xmlrpc.client import DateTime
 
 from kojismokydingo.types import (
     ArchiveInfo, ArchiveTypeInfo, BuildInfo, BuildrootInfo, BuildState,
     BTypeInfo, ChannelInfo, CGInfo, HostInfo, ListTasksOptions,
     PackageInfo, PermInfo, QueryOptions, RepoInfo, RepoState, RPMInfo,
     RPMSignature, SearchResult, TagBuildInfo, TagInfo, TagGroupInfo,
-    TagInheritance, TagPackageInfo, TargetInfo, TaskInfo, UserGroup,
-    UserInfo, )
+    TagInheritance, TagPackageInfo, TargetInfo, TaskInfo,
+    UserGroup, UserInfo, )
+
+# local mypy plugin and special decorator
+from proxytype import proxytype
+
+
+try:
+    from contextlib import AbstractContextManager as ContextManager
+except ImportError:
+    from typing import ContextManager
+
+try:
+    from typing import Self  # type: ignore
+except ImportError:
+    from typing_extensions import Self
 
 
 # Koji 1.34.0 intentionally broke API compatibility and removed these.
@@ -151,7 +168,7 @@ class TagError(GenericError):
 class ClientSession:
 
     baseurl: str
-    multicall: bool
+    multicall: "MultiCallHack"
     opts: Dict[str, Any]
 
     def __init__(
@@ -1050,4 +1067,47 @@ def read_config_files(
     ...
 
 
+# === MultiCallSession ===
+
+
+VirtualResultType = TypeVar("VirtualResultType")
+
+
+class VirtualCall(Generic[VirtualResultType]):
+    result: VirtualResultType
+
+
+@proxytype(ClientSession, VirtualCall)
+class MultiCallSession:
+    """
+    All of the same methods from a `ClientSession`, but wrapped to
+    return `VirtualCall` instances instead.
+
+    KSD doesn't use this type directly and I didn't want the proxytype
+    plugin to become a runtime dependency of KSD itself, so I left its
+    definition here rather than in `kojismokydingo.types` where it will
+    only be utilized when running mypy.
+    """
+    ...
+
+
+class MultiCallHack:
+
+    def __set__(self, obj: Any, value: bool) -> None:
+        # assignment to bool, eg. `session.multicall = True`
+        ...
+
+    def __bool__(self) -> bool:
+        ...
+
+    def __nonzero__(self) -> bool:
+        ...
+
+    def __call__(
+            self,
+            strict: Optional[bool] = False,
+            batch: Optional[int] = None) -> ContextManager[MultiCallSession]:
+        ...
+
+
 # The end.