[744] Add mypy type checking

Author: stsnel

.github/workflows/main.yml

@@ -0,0 +1,34 @@
+# This action is based on Github's default starter workflow for Python at
+# https://github.com/actions/starter-workflows/blob/master/ci/python-package.yml
+# (C) Github, MIT License
+
+name: "Python type checking"
+
+on: [push, pull_request]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-24.04
+
+    strategy:
+      matrix:
+        python-version: ['3.9']
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install dependencies
+      run: |
+        pip install ./
+        pip install mypy==1.16.1
+
+    - name: Run type checks
+      run: |
+        mypy --ignore-missing-imports irods
+

irods/exception.py

@@ -6,6 +6,7 @@
 import numbers
 import os
 import sys
+from typing import Dict
 
 
 class PycommandsException(Exception):
@@ -91,7 +92,8 @@ def __repr__(self):
 
 
 class iRODSExceptionMeta(type):
-    codes = {}
+    # Use "Any" type for values because Codacy fails when using a more specific type.
+    codes: "Dict[int, iRODSException]" = {}
     positive_code_error_message = (
         "For {name}, a positive code of {attrs[code]} was declared."
     )

irods/manager/data_object_manager.py

@@ -5,6 +5,7 @@
 import logging
 import os
 import weakref
+from typing import Any, List, Type
 from irods.models import DataObject, Collection
 from irods.manager import Manager
 from irods.manager._internal import _api_impl, _logical_path
@@ -38,8 +39,8 @@
 
 logger = logging.getLogger(__name__)
 
-_update_types = []
-_update_functions = weakref.WeakKeyDictionary()
+_update_types: List[Type] = []
+_update_functions: weakref.WeakKeyDictionary[Type, Any] = weakref.WeakKeyDictionary()
 
 
 def register_update_instance(object_, updater):  # updater
@@ -414,10 +415,10 @@ def chksum(self, path, **options):
                         "[",
                         "{",
                     ):  # in iRODS 4.2.11 and later, myStr is in JSON format.
-                        exc = Server_Checksum_Warning(checksum)
+                        exception = Server_Checksum_Warning(checksum)
                         if not r_error_stack:
-                            r_error_stack.fill(exc.response)
-                        raise exc
+                            r_error_stack.fill(exception.response)
+                        raise exception
                 except iRODSMessage.ResponseNotParseable:
                     # response.msg is None when VERIFY_CHKSUM_KW is used
                     pass

irods/manager/metadata_manager.py

@@ -1,6 +1,7 @@
 import logging
 import copy
 from os.path import dirname, basename
+from typing import Any, Dict
 
 from irods.manager import Manager
 from irods.message import MetadataRequest, iRODSMessage, JSON_Message
@@ -32,7 +33,7 @@ class MetadataManager(Manager):
     def use_timestamps(self):
         return getattr(self, "_use_ts", False)
 
-    __kw = {}  # default (empty) keywords
+    __kw : Dict[str, Any] = {}  # default (empty) keywords
 
     def _updated_keywords(self, opts):
         kw_ = self.__kw.copy()

irods/message/__init__.py

@@ -5,6 +5,7 @@
 import socket
 import json
 import irods.exception as ex
+from typing import Optional
 import xml.etree.ElementTree as ET_xml
 import defusedxml.ElementTree as ET_secure_xml
 from . import quasixml as ET_quasi_xml
@@ -85,15 +86,15 @@ class BadXMLSpec(RuntimeError):
 
 _XML_strings = {k: v for k, v in vars(XML_Parser_Type).items() if k.endswith("_XML")}
 
-_default_XML = os.environ.get(
+_default_XML_env = os.environ.get(
     "PYTHON_IRODSCLIENT_DEFAULT_XML", globals().get("_default_XML")
 )
 
-if not _default_XML:
+if not _default_XML_env:
     _default_XML = XML_Parser_Type.STANDARD_XML
 else:
     try:
-        _default_XML = _XML_strings[_default_XML]
+        _default_XML = _XML_strings[_default_XML_env]
     except KeyError:
         raise BadXMLSpec("XML parser type not recognized")
 
@@ -866,7 +867,7 @@ class VersionResponse(Message):
 
 class _admin_request_base(Message):
 
-    _name = None
+    _name : Optional[str] = None
 
     def __init__(self, *args):
         if self.__class__._name is None:
@@ -1153,7 +1154,7 @@ class ModDataObjMeta(Message):
 #  -- A tuple-descended class which facilitates filling in a
 #     quasi-RError stack from a JSON formatted list.
 
-_Server_Status_Message = namedtuple("server_status_msg", ("msg", "status"))
+_Server_Status_Message = namedtuple("_Server_Status_Message", ("msg", "status"))
 
 
 class RErrorStack(list):

irods/message/ordered.py

@@ -2,11 +2,7 @@
 # http://askawizard.blogspot.com/2008/10/ordered-properties-python-saga-part-5.html
 from itertools import count
 
-try:
-    next_counter = count().__next__
-except AttributeError:
-    next_counter = count().next
-
+next_counter = count().__next__
 
 class OrderedProperty:
 

irods/message/quasixml.py

@@ -176,14 +176,8 @@ def parse_elem(tokens):
     return elem
 
 
-try:
-    unicode  # Python 2
-except NameError:
-    unicode = str
-
-
 def fromstring(s):
-    if type(s) is unicode:
+    if type(s) is str:
         s = s.encode("utf-8")
     if type(s) is not bytes:
         raise TypeError("expected a bytes-object, got {}".format(type(s).__name__))

irods/models.py

@@ -1,9 +1,10 @@
+from typing import Dict, List, Tuple
 from irods.column import Column, Integer, String, DateTime, Keyword
 
 
 class ModelBase(type):
-    column_items = []
-    column_dict = {}
+    column_items : List[Tuple[int, Column]] = []
+    column_dict : Dict[int, Column] = {}
 
     @classmethod
     def columns(cls):

irods/parallel.py

@@ -9,6 +9,7 @@
 import concurrent.futures
 import threading
 import multiprocessing
+from typing import List, Union
 
 from irods.data_object import iRODSDataObject
 from irods.exception import DataObjectDoesNotExist
@@ -603,7 +604,8 @@ def setupLoggingWithDateTimeHeader(name, level=logging.DEBUG):
     sess = iRODSSession(irods_env_file=env_file, **ssl_settings)
     atexit.register(lambda: sess.cleanup())
 
-    opt, arg = getopt.getopt(sys.argv[1:], "vL:l:aR:N:")
+    opt, arg_ = getopt.getopt(sys.argv[1:], "vL:l:aR:N:")
+    arg: List[Union[str, int]] = list(arg_)
 
     opts = dict(opt)
 
@@ -622,8 +624,9 @@ def setupLoggingWithDateTimeHeader(name, level=logging.DEBUG):
 
     kwarg = {k.lstrip("-"): v for k, v in opts.items()}
 
-    arg[1] = Oper.PUT if arg[1].lower() in ("w", "put", "a") else Oper.GET
-    if async_xfer is not None:
+    # The purpose of the isinstance calls in the lines below is to allow mypy to infer the current type
+    arg[1] = Oper.PUT if isinstance(arg[1], str) and arg[1].lower() in ("w", "put", "a") else Oper.GET
+    if isinstance(arg[1], int) and async_xfer is not None:
         arg[1] |= Oper.NONBLOCKING
 
     ret = io_main(sess, *arg, **kwarg)  # arg[0] = data object or path

irods/prc_write_irodsA.py

@@ -3,6 +3,7 @@
 import getopt
 import textwrap
 import sys
+from typing import Callable, Dict
 
 from irods.auth.pam_password import _get_pam_password_from_stdin as get_password
 from irods.client_init import write_pam_irodsA_file, write_native_irodsA_file
@@ -21,7 +22,7 @@
     """
     )
 
-    vector = {"pam_password": write_pam_irodsA_file, "native": write_native_irodsA_file}
+    vector : Dict[str, Callable] = {"pam_password": write_pam_irodsA_file, "native": write_native_irodsA_file}
     opts, args = getopt.getopt(sys.argv[1:], "hi:", ["ttl=", "help"])
     optD = dict(opts)
     help_selected = {*optD} & {"-h", "--help"}
@@ -45,10 +46,12 @@
         inp_stream = optD.get("-i", None)
         if "--ttl" in optD:
             options["ttl"] = optD["--ttl"]
-        pw = get_password(
-            sys.stdin if inp_stream in ("-", None) else open(inp_stream, "r"),
-            prompt=f"Enter current password for scheme {scheme!r}: ",
-        )
+        if inp_stream is None or inp_stream == "-":
+            pw = get_password(sys.stdin,
+                              prompt=f"Enter current password for scheme {scheme!r}: ",)
+        else:
+            pw = get_password(open(inp_stream, "r", encoding='utf-8'),
+                              prompt=f"Enter current password for scheme {scheme!r}: ",)
         vector[scheme](pw, **options)
     else:
         print("did not recognize authentication scheme argument", file=sys.stderr)