Refactor command handling and improve type hints across multiple modules

Author: grongierisc

src/iop/_cli.py

@@ -98,7 +98,7 @@ def _determine_command_type(self) -> CommandType:
     def _handle_default(self) -> None:
         if self.args.default == 'not_set':
             print(_Director.get_default_production())
-        else:
+        elif self.args.default is not None:
             _Director.set_default_production(self.args.default)
 
     def _handle_list(self) -> None:
@@ -144,15 +144,16 @@ def _handle_export(self) -> None:
 
     def _handle_migrate(self) -> None:
         migrate_path = self.args.migrate
-        if not os.path.isabs(migrate_path):
-            migrate_path = os.path.join(os.getcwd(), migrate_path)
-        _Utils.migrate(migrate_path)
+        if migrate_path is not None:
+            if not os.path.isabs(migrate_path):
+                migrate_path = os.path.join(os.getcwd(), migrate_path)
+            _Utils.migrate(migrate_path)
 
     def _handle_log(self) -> None:
         if self.args.log == 'not_set':
             print(_Director.log_production())
-        else:
-            print(_Director.log_production_top(self.args.log))
+        elif self.args.log is not None:
+            print(_Director.log_production_top(int(self.args.log)))
 
     def _handle_init(self) -> None:
         _Utils.setup(None)

src/iop/_common.py

@@ -106,7 +106,8 @@ def _get_info(cls) -> List[str]:
                     super_class = classname[1:-1]
                     adapter = cls.get_adapter_type()
                     if adapter is None:
-                        adapter = cls.getAdapterType() # For backwards compatibility
+                        # for retro-compatibility
+                        adapter = cls.getAdapterType() # type: ignore
                     break
                 elif classname in ["'iop.BusinessProcess'","'iop.DuplexProcess'","'iop.InboundAdapter'","'iop.OutboundAdapter'",
                                    "'grongier.pex.BusinessProcess'","'grongier.pex.DuplexProcess'","'grongier.pex.InboundAdapter'","'grongier.pex.OutboundAdapter'"] :

src/iop/_dispatch.py

@@ -1,5 +1,5 @@
-from inspect import signature
-from typing import Any
+from inspect import signature, Parameter
+from typing import Any, List, Tuple, Callable
 
 from ._serialization import serialize_message, serialize_pickle_message, deserialize_message, deserialize_pickle_message
 from ._message_validator import is_message_instance, is_pickle_message_instance, is_iris_object_instance
@@ -59,7 +59,7 @@ def dispatch_deserializer(serial: Any) -> Any:
     else:
         return serial
 
-def dispach_message(host, request: Any) -> Any:
+def dispach_message(host: Any, request: Any) -> Any:
     """Dispatches the message to the appropriate method.
     
     Args:
@@ -79,23 +79,43 @@ def dispach_message(host, request: Any) -> Any:
 
     return getattr(host, call)(request)
 
-def create_dispatch(host) -> None:
-    """Creates a list of tuples, where each tuple contains the name of a class and the name of a method
-    that takes an instance of that class as its only argument.
+def create_dispatch(host: Any) -> None:
+    """Creates a dispatch table mapping class names to their handler methods.
+    The dispatch table consists of tuples of (fully_qualified_class_name, method_name).
+    Only methods that take a single typed parameter are considered as handlers.
     """
-    if len(host.DISPATCH) == 0:
-        method_list = [func for func in dir(host) if callable(getattr(host, func)) and not func.startswith("_")]
-        for method in method_list:
-            try:
-                param = signature(getattr(host, method)).parameters
-            except ValueError as e:
-                param = ''
-            if (len(param) == 1):
-                annotation = str(param[list(param)[0]].annotation)
-                i = annotation.find("'")
-                j = annotation.rfind("'")
-                if j == -1:
-                    j = None
-                classname = annotation[i+1:j]
-                host.DISPATCH.append((classname, method))
-    return
+    if len(host.DISPATCH) > 0:
+        return
+
+    for method_name in get_callable_methods(host):
+        handler_info = get_handler_info(host, method_name)
+        if handler_info:
+            host.DISPATCH.append(handler_info)
+
+def get_callable_methods(host: Any) -> List[str]:
+    """Returns a list of callable method names that don't start with underscore."""
+    return [
+        func for func in dir(host) 
+        if callable(getattr(host, func)) and not func.startswith("_")
+    ]
+
+def get_handler_info(host: Any, method_name: str) -> Tuple[str, str] | None:
+    """Analyzes a method to determine if it's a valid message handler.
+    Returns a tuple of (fully_qualified_class_name, method_name) if valid,
+    None otherwise.
+    """
+    try:
+        params = signature(getattr(host, method_name)).parameters
+        if len(params) != 1:
+            return None
+
+        param: Parameter = next(iter(params.values()))
+        annotation = param.annotation
+        
+        if annotation == Parameter.empty or not isinstance(annotation, type):
+            return None
+
+        return f"{annotation.__module__}.{annotation.__name__}", method_name
+            
+    except ValueError:
+        return None

src/iop/_iris.py

@@ -1,7 +1,7 @@
 import os
 from typing import Optional
 
-def get_iris(namespace: Optional[str]=None)->'iris':
+def get_iris(namespace: Optional[str]=None)->'iris': # type: ignore
     if namespace:
         os.environ['IRISNAMESPACE'] = namespace
     import iris

src/iop/_log_manager.py

@@ -72,9 +72,9 @@ def emit(self, record: logging.LogRecord) -> None:
         Args:
             record: The logging record to emit
         """
-        class_name = record.class_name if hasattr(record, "class_name") else record.name
-        method_name = record.method_name if hasattr(record, "method_name") else record.funcName
-        if self.to_console or (hasattr(record, "to_console") and record.to_console):
+        class_name = record.class_name if hasattr(record, "class_name") else record.name # type: ignore has been added as extra attribute in LogRecord
+        method_name = record.method_name if hasattr(record, "method_name") else record.funcName # type: ignore has been added as extra attribute in LogRecord
+        if self.to_console or (hasattr(record, "to_console") and record.to_console): # type: ignore has been added as extra attribute in LogRecord
             _iris.get_iris().cls("%SYS.System").WriteToConsoleLog(self.format(record),
                 0,self.level_map_console.get(record.levelno, 0),class_name+"."+method_name)
         else:

src/iop/_private_session_duplex.py

@@ -2,6 +2,7 @@
 
 from ._business_host import _BusinessHost
 from ._decorators import input_deserializer, input_serializer_param, output_serializer, input_serializer, output_deserializer
+from ._dispatch import create_dispatch, dispach_message
 
 class _PrivateSessionDuplex(_BusinessHost):
 
@@ -27,7 +28,7 @@ def on_message(self, request):
     @output_serializer
     def _dispatch_on_message(self, request):
         """ For internal use only. """
-        return self._dispach_message(request)
+        return dispach_message(self,request)
 
     def _set_iris_handles(self, handle_current, handle_partner):
         """ For internal use only. """

src/iop/_private_session_process.py

@@ -12,7 +12,7 @@ def _dispatch_on_document(self, host_object,source_config_name, request):
         self._save_persistent_properties(host_object)
         return return_object
 
-    def on_document(source_config_name,request):
+    def on_document(self,source_config_name,request):
         pass
 
 

src/iop/_serialization.py

@@ -107,7 +107,7 @@ def _parse_classname(classname: str) -> tuple[str, str]:
             raise SerializationError(f"Classname must include a module: {classname}")
         return classname[:j], classname[j+1:]
 
-def dataclass_from_dict(klass: Type, dikt: Dict) -> Any:
+def dataclass_from_dict(klass: Type | Any, dikt: Dict) -> Any:
     """Converts a dictionary to a dataclass instance.
     Handles non attended fields and nested dataclasses."""
 

src/iop/_utils.py

@@ -6,6 +6,7 @@
 import importlib.util
 import importlib.resources
 import json
+from typing import Any, Dict, Optional, Union
 
 import xmltodict
 from pydantic import TypeAdapter
@@ -25,13 +26,12 @@ def raise_on_error(sc):
             raise RuntimeError(_iris.get_iris().system.Status.GetOneStatusText(sc))
 
     @staticmethod
-    def setup(path:str = None):
+    def setup(path:Optional[str] = None):
 
         if path is None:
             # get the path of the data folder with importlib.resources
             try:
-                path = importlib.resources.files('iop').joinpath('cls')
-                path = str(path)
+                path = str(importlib.resources.files('iop').joinpath('cls'))
             except ModuleNotFoundError:
                 path = None
 
@@ -40,29 +40,28 @@ def setup(path:str = None):
 
         # for retrocompatibility load grongier.pex
         try:
-            path = importlib.resources.files('grongier').joinpath('cls')
-            path = str(path)
+            path = str(importlib.resources.files('grongier').joinpath('cls'))
         except ModuleNotFoundError:
             path = None
 
         if path:
             _Utils.raise_on_error(_iris.get_iris().cls('%SYSTEM.OBJ').LoadDir(path,'cubk',"*.cls",1))
 
     @staticmethod
-    def register_message_schema(cls):
+    def register_message_schema(msg_cls: type):
         """
         It takes a class and registers the schema
         
         :param cls: The class to register
         """
-        if issubclass(cls,_PydanticMessage):
-            schema = cls.model_json_schema()
-        elif issubclass(cls,_Message):
-            type_adapter = TypeAdapter(cls)
+        if issubclass(msg_cls,_PydanticMessage):
+            schema = msg_cls.model_json_schema()
+        elif issubclass(msg_cls,_Message):
+            type_adapter = TypeAdapter(msg_cls)
             schema = type_adapter.json_schema()
         else:
             raise ValueError("The class must be a subclass of _Message or _PydanticMessage")
-        schema_name = cls.__module__ + '.' + cls.__name__
+        schema_name = msg_cls.__module__ + '.' + msg_cls.__name__
         schema_str = json.dumps(schema)
         categories = schema_name
         _Utils.register_schema(schema_name,schema_str,categories)
@@ -172,10 +171,11 @@ def _register_file(filename:str,path:str,overwrite:int=1,iris_package_name:str='
             for klass in classes:
                 extend = ''
                 if len(klass.bases) == 1:
-                    if hasattr(klass.bases[0],'id'):
-                        extend = klass.bases[0].id
-                    else:
-                        extend = klass.bases[0].attr
+                    base = klass.bases[0]
+                    if isinstance(base, ast.Name):
+                        extend = base.id
+                    elif isinstance(base, ast.Attribute):
+                        extend = base.attr
                 if extend in ('BusinessOperation','BusinessProcess','BusinessService','DuplexService','DuplexProcess','DuplexOperation','InboundAdapter','OutboundAdapter'):
                     module = _Utils.filename_to_module(filename)
                     iris_class_name = f"{iris_package_name}.{module}.{klass.name}"
@@ -283,7 +283,7 @@ def import_module_from_path(module_name, file_path):
             raise ValueError("The file path must be absolute")
 
         spec = importlib.util.spec_from_file_location(module_name, file_path)
-        if spec is None:
+        if spec is None or spec.loader is None:
             raise ImportError(f"Cannot find module named {module_name} at {file_path}")
 
         module = importlib.util.module_from_spec(spec)