Improve namedtuples in aiida/engine (#4688)

This commit replaces old-style namedtuples with `typing.NamedTuple` sub-classes.
This allows for typing of fields and better default value assignment.

Note this feature requires python>=3.6.1,
but it is anyhow intended that python 3.6 be dropped for the next release.

aiida/engine/processes/calcjobs/calcjob.py

@@ -370,7 +370,7 @@ def parse(self, retrieved_temporary_folder: Optional[str] = None) -> ExitCode:
         for entry in self.node.get_outgoing():
             self.out(entry.link_label, entry.node)
 
-        return exit_code or ExitCode(0)  # type: ignore[call-arg]
+        return exit_code or ExitCode(0)
 
     def parse_scheduler_output(self, retrieved: orm.Node) -> Optional[ExitCode]:
         """Parse the output of the scheduler if that functionality has been implemented for the plugin."""

aiida/engine/processes/exit_code.py

@@ -8,38 +8,36 @@
 # For further information please visit http://www.aiida.net               #
 ###########################################################################
 """A namedtuple and namespace for ExitCodes that can be used to exit from Processes."""
-from collections import namedtuple
+from typing import NamedTuple, Optional
 from aiida.common.extendeddicts import AttributeDict
 
 __all__ = ('ExitCode', 'ExitCodesNamespace')
 
 
-class ExitCode(namedtuple('ExitCode', ['status', 'message', 'invalidates_cache'])):
+class ExitCode(NamedTuple):
     """A simple data class to define an exit code for a :class:`~aiida.engine.processes.process.Process`.
 
-    When an instance of this clas is returned from a `Process._run()` call, it will be interpreted that the `Process`
+    When an instance of this class is returned from a `Process._run()` call, it will be interpreted that the `Process`
     should be terminated and that the exit status and message of the namedtuple should be set to the corresponding
     attributes of the node.
 
-    .. note:: this class explicitly sub-classes a namedtuple to not break backwards compatibility and to have it behave
-        exactly as a tuple.
-
     :param status: positive integer exit status, where a non-zero value indicated the process failed, default is `0`
-    :type status: int
-
     :param message: optional message with more details about the failure mode
-    :type message: str
-
     :param invalidates_cache: optional flag, indicating that a process should not be used in caching
-    :type invalidates_cache: bool
     """
 
+    status: int = 0
+    message: Optional[str] = None
+    invalidates_cache: bool = False
+
     def format(self, **kwargs: str) -> 'ExitCode':
         """Create a clone of this exit code where the template message is replaced by the keyword arguments.
 
         :param kwargs: replacement parameters for the template message
-        :return: `ExitCode`
+
         """
+        if self.message is None:
+            raise ValueError('message is None')
         try:
             message = self.message.format(**kwargs)
         except KeyError:
@@ -49,10 +47,6 @@ def format(self, **kwargs: str) -> 'ExitCode':
         return ExitCode(self.status, message, self.invalidates_cache)
 
 
-# Set the defaults for the `ExitCode` attributes
-ExitCode.__new__.__defaults__ = (0, None, False)  # type: ignore[attr-defined]
-
-
 class ExitCodesNamespace(AttributeDict):
     """A namespace of `ExitCode` instances that can be accessed through getattr as well as getitem.
 

aiida/engine/processes/functions.py

@@ -408,4 +408,4 @@ def run(self) -> Optional['ExitCode']:
                 'Must be a Data type or a mapping of {{string: Data}}'.format(result.__class__)
             )
 
-        return ExitCode()  # type: ignore[call-arg]
+        return ExitCode()

aiida/engine/processes/workchains/utils.py

@@ -8,33 +8,36 @@
 # For further information please visit http://www.aiida.net               #
 ###########################################################################
 """Utilities for `WorkChain` implementations."""
-from collections import namedtuple
 from functools import partial
 from inspect import getfullargspec
 from types import FunctionType  # pylint: disable=no-name-in-module
-from typing import List, Optional, Union
+from typing import List, Optional, Union, NamedTuple
 from wrapt import decorator
 
 from ..exit_code import ExitCode
 
 __all__ = ('ProcessHandlerReport', 'process_handler')
 
-ProcessHandlerReport = namedtuple('ProcessHandlerReport', 'do_break exit_code')
-ProcessHandlerReport.__new__.__defaults__ = (False, ExitCode())  # type: ignore[attr-defined,call-arg]
-"""A namedtuple to define a process handler report for a :class:`aiida.engine.BaseRestartWorkChain`.
 
-This namedtuple should be returned by a process handler of a work chain instance if the condition of the handler was
-met by the completed process. If no further handling should be performed after this method the `do_break` field should
-be set to `True`. If the handler encountered a fatal error and the work chain needs to be terminated, an `ExitCode` with
-non-zero exit status can be set. This exit code is what will be set on the work chain itself. This works because the
-value of the `exit_code` field returned by the handler, will in turn be returned by the `inspect_process` step and
-returning a non-zero exit code from any work chain step will instruct the engine to abort the work chain.
+class ProcessHandlerReport(NamedTuple):
+    """A namedtuple to define a process handler report for a :class:`aiida.engine.BaseRestartWorkChain`.
 
-:param do_break: boolean, set to `True` if no further process handlers should be called, default is `False`
-:param exit_code: an instance of the :class:`~aiida.engine.processes.exit_code.ExitCode` tuple. If not explicitly set,
-    the default `ExitCode` will be instantiated which has status `0` meaning that the work chain step will be considered
-    successful and the work chain will continue to the next step.
-"""
+    This namedtuple should be returned by a process handler of a work chain instance if the condition of the handler was
+    met by the completed process. If no further handling should be performed after this method the `do_break` field
+    should be set to `True`.
+    If the handler encountered a fatal error and the work chain needs to be terminated, an `ExitCode` with
+    non-zero exit status can be set. This exit code is what will be set on the work chain itself. This works because the
+    value of the `exit_code` field returned by the handler, will in turn be returned by the `inspect_process` step and
+    returning a non-zero exit code from any work chain step will instruct the engine to abort the work chain.
+
+    :param do_break: boolean, set to `True` if no further process handlers should be called, default is `False`
+    :param exit_code: an instance of the :class:`~aiida.engine.processes.exit_code.ExitCode` tuple.
+        If not explicitly set, the default `ExitCode` will be instantiated,
+        which has status `0` meaning that the work chain step will be considered
+        successful and the work chain will continue to the next step.
+    """
+    do_break: bool = False
+    exit_code: ExitCode = ExitCode()
 
 
 def process_handler(
@@ -108,7 +111,9 @@ def wrapper(wrapped, instance, args, kwargs):
         # When the handler will be called by the `BaseRestartWorkChain` it will pass the node as the only argument
         node = args[0]
 
-        if exit_codes is not None and node.exit_status not in [exit_code.status for exit_code in exit_codes]:
+        if exit_codes is not None and node.exit_status not in [
+            exit_code.status for exit_code in exit_codes  # type: ignore[union-attr]
+        ]:
             result = None
         else:
             result = wrapped(*args, **kwargs)

aiida/engine/processes/workchains/workchain.py

@@ -214,7 +214,7 @@ def _do_step(self) -> Any:
         else:
             # Set result to None unless stepper_result was non-zero positive integer or ExitCode with similar status
             if isinstance(stepper_result, int) and stepper_result > 0:
-                result = ExitCode(stepper_result)  # type: ignore[call-arg]
+                result = ExitCode(stepper_result)
             elif isinstance(stepper_result, ExitCode) and stepper_result.status > 0:
                 result = stepper_result
             else:

aiida/engine/transports.py

@@ -8,7 +8,6 @@
 # For further information please visit http://www.aiida.net               #
 ###########################################################################
 """A transport queue to batch process multiple tasks that require a Transport."""
-from collections import namedtuple
 import contextlib
 import logging
 import traceback
@@ -41,7 +40,6 @@ class TransportQueue:
     up to that point.  This way opening of transports (a costly operation) can
     be minimised.
     """
-    AuthInfoEntry = namedtuple('AuthInfoEntry', ['authinfo', 'transport', 'callbacks', 'callback_handle'])
 
     def __init__(self, loop: Optional[asyncio.AbstractEventLoop] = None):
         """

setup.json

@@ -7,7 +7,7 @@
     "author_email": "[email protected]",
     "description": "AiiDA is a workflow manager for computational science with a strong focus on provenance, performance and extensibility.",
     "include_package_data": true,
-    "python_requires": ">=3.6",
+    "python_requires": ">=3.6.1",
     "classifiers": [
         "Framework :: AiiDA",
         "License :: OSI Approved :: MIT License",