Bump mypy to 1.16.1 (#6858)

* mypy: use --pretty for nicer error messages!
* Fix bug in existing archive import validation!
* Use cast
* Fix typing of shared_options in src/aiida/cmdline/groups/dynamic.py
* Mark ctx positional-only in profile_configure_rabbitmq

Author: danielhollas

.pre-commit-config.yaml

@@ -78,7 +78,7 @@ repos:
   - id: mypy
     name: mypy
     entry: mypy
-    args: [--config-file=pyproject.toml]
+    args: [--config-file=pyproject.toml, --pretty]
     language: python
     types: [python]
     require_serial: true

pyproject.toml

@@ -229,7 +229,7 @@ notebook = [
 ]
 pre-commit = [
   'aiida-core[atomic_tools,rest,tests,tui]',
-  'mypy~=1.13.0',
+  'mypy~=1.16.0',
   'packaging~=23.0',
   'pre-commit~=3.5',
   'sqlalchemy[mypy]~=2.0',

src/aiida/cmdline/commands/cmd_profile.py

@@ -131,7 +131,7 @@ def profile_setup():
     """Set up a new profile."""
 
 
-@verdi_profile.command('configure-rabbitmq')  # type: ignore[arg-type]
+@verdi_profile.command('configure-rabbitmq')
 @arguments.PROFILE(default=defaults.get_default_profile)
 @options.FORCE()
 @setup.SETUP_BROKER_PROTOCOL()
@@ -142,7 +142,7 @@ def profile_setup():
 @setup.SETUP_BROKER_VIRTUAL_HOST()
 @options.NON_INTERACTIVE(default=True, show_default='--non-interactive')
 @click.pass_context
-def profile_configure_rabbitmq(ctx, profile, non_interactive, force, **kwargs):
+def profile_configure_rabbitmq(ctx, /, profile, non_interactive, force, **kwargs):
     """Configure RabbitMQ for a profile.
 
     Enable RabbitMQ for a profile that was created without a broker, or reconfigure existing connection details.

src/aiida/cmdline/groups/dynamic.py

@@ -47,7 +47,7 @@ def __init__(
         command: t.Callable,
         entry_point_group: str,
         entry_point_name_filter: str = r'.*',
-        shared_options: list[click.Option] | None = None,
+        shared_options: list[t.Callable[[t.Any], t.Any]] | None = None,
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -143,7 +143,7 @@ def apply_options(func):
 
         return apply_options
 
-    def list_options(self, entry_point: str) -> list:
+    def list_options(self, entry_point: str) -> list[t.Callable[[t.Any], t.Any]]:
         """Return the list of options that should be applied to the command for the given entry point.
 
         :param entry_point: The entry point.

src/aiida/engine/processes/functions.py

@@ -355,11 +355,11 @@ def build(func: FunctionType, node_class: t.Type['ProcessNode']) -> t.Type['Func
         """
         if (
             not issubclass(node_class, ProcessNode)  # type: ignore[redundant-expr]
-            or not issubclass(node_class, FunctionCalculationMixin)  # type: ignore[unreachable]
+            or not issubclass(node_class, FunctionCalculationMixin)
         ):
             raise TypeError('the node_class should be a sub class of `ProcessNode` and `FunctionCalculationMixin`')
 
-        signature = inspect.signature(func)  # type: ignore[unreachable]
+        signature = inspect.signature(func)
 
         args: list[str] = []
         var_positional: str | None = None
@@ -373,17 +373,21 @@ def build(func: FunctionType, node_class: t.Type['ProcessNode']) -> t.Type['Func
             LOGGER.warning(f'function `{func.__name__}` has invalid type hints: {exception}')
             annotations = {}
 
-        try:
-            parsed_docstring = docstring_parser.parse(func.__doc__)
-        except Exception as exception:
-            LOGGER.warning(f'function `{func.__name__}` has a docstring that could not be parsed: {exception}')
-            param_help_string = {}
+        if func.__doc__ is None:
+            param_help_string: dict[str, str | None] = {}
             namespace_help_string = None
         else:
-            param_help_string = {param.arg_name: param.description for param in parsed_docstring.params}
-            namespace_help_string = parsed_docstring.short_description if parsed_docstring.short_description else ''
-            if parsed_docstring.long_description is not None:
-                namespace_help_string += f'\n\n{parsed_docstring.long_description}'
+            try:
+                parsed_docstring = docstring_parser.parse(func.__doc__)
+            except Exception as exception:
+                LOGGER.warning(f'function `{func.__name__}` has a docstring that could not be parsed: {exception}')
+                param_help_string = {}
+                namespace_help_string = None
+            else:
+                param_help_string = {param.arg_name: param.description for param in parsed_docstring.params}
+                namespace_help_string = parsed_docstring.short_description if parsed_docstring.short_description else ''
+                if parsed_docstring.long_description is not None:
+                    namespace_help_string += f'\n\n{parsed_docstring.long_description}'
 
         for key, parameter in signature.parameters.items():
             if parameter.kind in [parameter.POSITIONAL_ONLY, parameter.POSITIONAL_OR_KEYWORD, parameter.KEYWORD_ONLY]:
@@ -435,7 +439,7 @@ def define(cls, spec):
                     def indirect_default(value=default):
                         return to_aiida_type(value)
                 else:
-                    indirect_default = default
+                    indirect_default = default  # type: ignore[assignment]
 
                 spec.input(
                     parameter.name,

src/aiida/engine/processes/process.py

@@ -83,7 +83,7 @@ class Process(PlumpyProcess):
     class SaveKeys(enum.Enum):
         """Keys used to identify things in the saved instance state bundle."""
 
-        CALC_ID: str = 'calc_id'
+        CALC_ID = 'calc_id'
 
     @classmethod
     def spec(cls) -> ProcessSpec:
@@ -379,7 +379,7 @@ def kill(self, msg_text: str | None = None, force_kill: bool = False) -> Union[b
                     continue
                 try:
                     result = self.runner.controller.kill_process(child.pk, msg_text=f'Killed by parent<{self.node.pk}>')
-                    result = asyncio.wrap_future(result)  # type: ignore[arg-type]
+                    result = asyncio.wrap_future(result)
                     if asyncio.isfuture(result):
                         killing.append(result)
                 except ConnectionClosed:

src/aiida/orm/nodes/data/code/installed.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import pathlib
+from typing import cast
 
 from pydantic import field_serializer, field_validator
 
@@ -47,15 +48,15 @@ class Model(AbstractCode.Model):
             ...,
             title='Computer',
             description='The remote computer on which the executable resides.',
-            orm_to_model=lambda node, _: node.computer.label,
+            orm_to_model=lambda node, _: cast('InstalledCode', node).computer.label,
             short_name='-Y',
             priority=2,
         )
         filepath_executable: str = MetadataField(
             ...,
             title='Filepath executable',
             description='Filepath of the executable on the remote computer.',
-            orm_to_model=lambda node, _: str(node.filepath_executable),  # type: ignore[attr-defined]
+            orm_to_model=lambda node, _: str(cast('InstalledCode', node).filepath_executable),
             short_name='-X',
             priority=1,
         )
@@ -83,7 +84,7 @@ def __init__(self, computer: Computer, filepath_executable: str, **kwargs):
         """
         super().__init__(**kwargs)
         self.computer = computer
-        self.filepath_executable = filepath_executable  # type: ignore[assignment]
+        self.filepath_executable = filepath_executable
 
     def _validate(self):
         """Validate the instance by checking that a computer has been defined.
@@ -159,7 +160,7 @@ def get_executable(self) -> pathlib.PurePath:
         """
         return self.filepath_executable
 
-    @property  # type: ignore[override]
+    @property
     def computer(self) -> Computer:
         """Return the computer of this code."""
         assert self.backend_entity.computer is not None

src/aiida/orm/nodes/data/code/legacy.py

@@ -231,7 +231,7 @@ def get_code_helper(cls, label, machinename=None, backend=None):
             return result[0]
 
     @classmethod
-    def get(cls, pk=None, label=None, machinename=None):
+    def get(cls, pk=None, label=None, machinename=None):  # type: ignore[override]
         """Get a Computer object with given identifier string, that can either be
         the numeric ID (pk), or the label (and computername) (if unique).
 

src/aiida/orm/nodes/data/code/portable.py

@@ -104,7 +104,7 @@ def __init__(
         if not filepath_files_path.is_dir():
             raise ValueError(f'The filepath `{filepath_files}` is not a directory.')
 
-        self.filepath_executable = filepath_executable  # type: ignore[assignment] # should be fixed in mypy 1.15 see mypy/commit/1eb9d4c
+        self.filepath_executable = filepath_executable
         self.base.repository.put_object_from_tree(str(filepath_files))
 
     def _validate(self):

src/aiida/orm/nodes/process/calculation/calcfunction.py

@@ -47,7 +47,7 @@ def validate_outgoing(self, target: 'Node', link_type: LinkType, link_label: str
             )
 
 
-class CalcFunctionNode(FunctionCalculationMixin, CalculationNode):  # type: ignore[misc]
+class CalcFunctionNode(FunctionCalculationMixin, CalculationNode):
     """ORM class for all nodes representing the execution of a calcfunction."""
 
     _CLS_NODE_LINKS = CalcFunctionNodeLinks