Add _iris module for IRIS namespace management and refactor imports

Author: grongierisc

src/iop/_async_request.py

@@ -1,5 +1,5 @@
 import asyncio
-import iris
+from . import _iris
 from typing import Any, Optional, Union
 from iop._dispatch import dispatch_deserializer, dispatch_serializer
 from iop._message import _Message as Message
@@ -24,6 +24,7 @@ def __init__(self, target: str, request: Union[Message, Any],
 
     async def send(self) -> None:
         # init parameters
+        iris = _iris.get_iris()
         message_header_id = iris.ref()
         queue_name = iris.ref()
         end_time = iris.ref()
@@ -46,6 +47,7 @@ async def send(self) -> None:
         self.set_result(self._response)
 
     def is_done(self) -> None:
+        iris = _iris.get_iris()
         response = iris.ref()
         status = self._iris_handle.dispatchIsRequestDone(self.timeout, self._end_time,
                                                        self._queue_name, self._message_header_id,

src/iop/_business_host.py

@@ -1,7 +1,7 @@
 from inspect import getsource
 from typing import Any,List, Optional, Tuple, Union
 
-import iris
+from . import _iris
 
 from iop._common import _Common
 from iop._message import _Message as Message
@@ -109,6 +109,7 @@ def _validate_target_request(self, target_request: List[Tuple[str, Union[Message
 
     def _create_call_structure(self, target: str, request: Union[Message, Any]) -> Any:
         """Create an Ens.CallStructure object for the request."""
+        iris = _iris.get_iris()
         call = iris.cls("Ens.CallStructure")._New()
         call.TargetDispatchName = target
         call.Request = dispatch_serializer(request)

src/iop/_cli.py

@@ -7,10 +7,10 @@
 import sys
 from typing import Optional, Callable
 from importlib.metadata import version
-
 from iop._director import _Director
 from iop._utils import _Utils
 
+
 class CommandType(Enum):
     DEFAULT = auto()
     LIST = auto()
@@ -46,11 +46,16 @@ class CommandArgs:
     test: Optional[str] = None
     classname: Optional[str] = None
     body: Optional[str] = None
+    namespace: Optional[str] = None
 
 class Command:
     def __init__(self, args: CommandArgs):
         self.args = args
 
+        if self.args.namespace and self.args.namespace != 'not_set':
+            # set environment variable IRISNAMESPACE
+            os.environ['IRISNAMESPACE'] = self.args.namespace
+
     def execute(self) -> None:
         command_type = self._determine_command_type()
         command_handlers = {
@@ -182,6 +187,9 @@ def create_parser() -> argparse.ArgumentParser:
     test = main_parser.add_argument_group('test arguments')
     test.add_argument('-C', '--classname', help='test classname', nargs='?', const='not_set')
     test.add_argument('-B', '--body', help='test body', nargs='?', const='not_set')
+
+    namespace = main_parser.add_argument_group('namespace arguments')
+    namespace.add_argument('-n', '--namespace', help='set namespace', nargs='?', const='not_set')
 
     return main_parser
 

src/iop/_common.py

@@ -3,7 +3,7 @@
 import traceback
 from typing import Any, ClassVar, List, Optional, Tuple
 
-import iris
+from . import _iris
 
 from iop._log_manager import LogManager, logging
 
@@ -285,6 +285,7 @@ def log_assert(self, message: str) -> None:
         Parameters:
         message: a string that is written to the log.
         """
+        iris = _iris.get_iris()
         current_class, current_method = self._log()
         iris.cls("Ens.Util.Log").LogAssert(current_class, current_method, message)
 

src/iop/_director.py

@@ -1,7 +1,7 @@
 import asyncio
 import datetime
 import functools
-import iris
+from . import _iris
 import intersystems_iris.dbapi._DBAPI as irisdbapi
 import signal
 from dataclasses import dataclass
@@ -55,7 +55,7 @@ def create_business_service(target):
         Returns:
             an object that contains an instance of IRISBusinessService
         """
-        iris_object = iris.cls("IOP.Director").dispatchCreateBusinessService(target)
+        iris_object = _iris.get_iris().cls("IOP.Director").dispatchCreateBusinessService(target)
         return iris_object
 
     @staticmethod
@@ -69,7 +69,7 @@ def create_python_business_service(target):
         Returns:
             an object that contains an instance of IRISBusinessService
         """
-        iris_object = iris.cls("IOP.Director").dispatchCreateBusinessService(target)
+        iris_object = _iris.get_iris().cls("IOP.Director").dispatchCreateBusinessService(target)
         return iris_object.GetClass()
 
     ### List of function to manage the production
@@ -102,37 +102,37 @@ async def _start_production_async(production_name=None, handler=None):
     def start_production(production_name=None):
         if production_name is None or production_name == '':
             production_name = _Director.get_default_production()
-        iris.cls('Ens.Director').StartProduction(production_name)
+        _iris.get_iris().cls('Ens.Director').StartProduction(production_name)
 
     ### stop production
     @staticmethod
     def stop_production():
-        iris.cls('Ens.Director').StopProduction()
+        _iris.get_iris().cls('Ens.Director').StopProduction()
 
     ### restart production
     @staticmethod
     def restart_production():
-        iris.cls('Ens.Director').RestartProduction()
+        _iris.get_iris().cls('Ens.Director').RestartProduction()
 
     ### shutdown production
     @staticmethod
     def shutdown_production():
-        iris.cls('Ens.Director').StopProduction(10,1)
+        _iris.get_iris().cls('Ens.Director').StopProduction(10,1)
 
     ### update production
     @staticmethod
     def update_production():
-        iris.cls('Ens.Director').UpdateProduction()
+        _iris.get_iris().cls('Ens.Director').UpdateProduction()
 
     ### list production
     @staticmethod
     def list_productions():
-        return iris.cls('IOP.Director').dispatchListProductions()
+        return _iris.get_iris().cls('IOP.Director').dispatchListProductions()
 
     ### status production
     @staticmethod
     def status_production():
-        dikt = iris.cls('IOP.Director').StatusProduction()
+        dikt = _iris.get_iris().cls('IOP.Director').StatusProduction()
         if dikt['Production'] is None or dikt['Production'] == '':
             dikt['Production'] = _Director.get_default_production()
         return dikt
@@ -141,13 +141,13 @@ def status_production():
     @staticmethod
     def set_default_production(production_name=''):
         #set ^Ens.Configuration("SuperUser","LastProduction")
-        glb = iris.gref("^Ens.Configuration")
+        glb = _iris.get_iris().gref("^Ens.Configuration")
         glb['csp', "LastProduction"] = production_name
 
     ### get default production
     @staticmethod
     def get_default_production():
-        glb = iris.gref("^Ens.Configuration")
+        glb = _iris.get_iris().gref("^Ens.Configuration")
         default_production_name = glb['csp', "LastProduction"]
         if default_production_name is None or default_production_name == '':
             default_production_name = 'Not defined'
@@ -257,28 +257,28 @@ def test_component(target,message=None,classname=None,body=None):
         body: the body of the message
         """
         if not message:
-            message = iris.cls('Ens.Request')._New()
+            message = _iris.get_iris().cls('Ens.Request')._New()
         if classname:
             # if classname start with 'iris.' then create an iris object
             if classname.startswith('iris.'):
                 # strip the iris. prefix
                 classname = classname[5:]
                 if body:
-                    message = iris.cls(classname)._New(body)
+                    message = _iris.get_iris().cls(classname)._New(body)
                 else:
-                    message = iris.cls(classname)._New()
+                    message = _iris.get_iris().cls(classname)._New()
             # else create a python object
             else:
                 # python message are casted to Grongier.PEX.Message
-                message = iris.cls("IOP.Message")._New()
+                message = _iris.get_iris().cls("IOP.Message")._New()
                 message.classname = classname
                 if body:
                     message.json = body
                 else:
                     message.json = _Utils.string_to_stream("{}")
         # serialize the message
         serial_message = dispatch_serializer(message)
-        response = iris.cls('IOP.Utils').dispatchTestComponent(target,serial_message)
+        response = _iris.get_iris().cls('IOP.Utils').dispatchTestComponent(target,serial_message)
         try:
             deserialized_response = dispatch_deserializer(response)
         except ImportError as e:

src/iop/_iris.py

@@ -0,0 +1,7 @@
+import os
+
+def get_iris(namespace:str=None)->'iris':
+    if namespace:
+        os.environ['IRISNAMESPACE'] = namespace
+    import iris
+    return iris

src/iop/_log_manager.py

@@ -1,5 +1,5 @@
 import traceback
-import iris
+from . import _iris
 import logging
 from typing import Optional, Tuple
 
@@ -41,11 +41,11 @@ def __init__(self, to_console: bool = False):
 
         # Map Python logging levels to IRIS logging methods
         self.level_map = {
-            logging.DEBUG: iris.cls("Ens.Util.Log").LogTrace,
-            logging.INFO: iris.cls("Ens.Util.Log").LogInfo,
-            logging.WARNING: iris.cls("Ens.Util.Log").LogWarning,
-            logging.ERROR: iris.cls("Ens.Util.Log").LogError,
-            logging.CRITICAL: iris.cls("Ens.Util.Log").LogAlert,
+            logging.DEBUG: _iris.get_iris().cls("Ens.Util.Log").LogTrace,
+            logging.INFO: _iris.get_iris().cls("Ens.Util.Log").LogInfo,
+            logging.WARNING: _iris.get_iris().cls("Ens.Util.Log").LogWarning,
+            logging.ERROR: _iris.get_iris().cls("Ens.Util.Log").LogError,
+            logging.CRITICAL: _iris.get_iris().cls("Ens.Util.Log").LogAlert,
         }
         # Map Python logging levels to IRIS logging Console level
         self.level_map_console = {
@@ -76,8 +76,8 @@ def emit(self, record: logging.LogRecord) -> None:
         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):
-            iris.cls("%SYS.System").WriteToConsoleLog(self.format(record),
+            _iris.get_iris().cls("%SYS.System").WriteToConsoleLog(self.format(record),
                 0,self.level_map_console.get(record.levelno, 0),class_name+"."+method_name)
         else:
-            log_func = self.level_map.get(record.levelno, iris.cls("Ens.Util.Log").LogInfo)
+            log_func = self.level_map.get(record.levelno, _iris.get_iris().cls("Ens.Util.Log").LogInfo)
             log_func(class_name, method_name, self.format(record))

src/iop/_serialization.py

@@ -7,7 +7,7 @@
 from dataclasses import asdict, is_dataclass
 from typing import Any, Dict, Type
 
-import iris
+from . import _iris
 from pydantic import BaseModel, TypeAdapter, ValidationError
 
 from iop._message import _PydanticPickleMessage
@@ -37,17 +37,17 @@ def _convert_to_json_safe(obj: Any) -> Any:
             raise SerializationError(f"Object {obj} must be a Pydantic model or dataclass")
 
     @staticmethod
-    def serialize(message: Any, use_pickle: bool = False) -> iris.cls:
+    def serialize(message: Any, use_pickle: bool = False) -> Any:
         """Serializes a message to IRIS format."""
         if isinstance(message, _PydanticPickleMessage) or use_pickle:
             return MessageSerializer._serialize_pickle(message)
         return MessageSerializer._serialize_json(message)
 
     @staticmethod
-    def _serialize_json(message: Any) -> iris.cls:
+    def _serialize_json(message: Any) -> Any:
         json_string = MessageSerializer._convert_to_json_safe(message)
 
-        msg = iris.cls('IOP.Message')._New()
+        msg = _iris.get_iris().cls('IOP.Message')._New()
         msg.classname = f"{message.__class__.__module__}.{message.__class__.__name__}"
 
         if hasattr(msg, 'buffer') and len(json_string) > msg.buffer:
@@ -57,13 +57,13 @@ def _serialize_json(message: Any) -> iris.cls:
         return msg
 
     @staticmethod
-    def deserialize(serial: iris.cls, use_pickle: bool = False) -> Any:
+    def deserialize(serial: Any, use_pickle: bool = False) -> Any:
         if use_pickle:
             return MessageSerializer._deserialize_pickle(serial)
         return MessageSerializer._deserialize_json(serial)
 
     @staticmethod
-    def _deserialize_json(serial: iris.cls) -> Any:
+    def _deserialize_json(serial: Any) -> Any:
         if not serial.classname:
             raise SerializationError("JSON message malformed, must include classname")
 
@@ -88,15 +88,15 @@ def _deserialize_json(serial: iris.cls) -> Any:
             raise SerializationError(f"Failed to deserialize JSON: {str(e)}")
 
     @staticmethod
-    def _serialize_pickle(message: Any) -> iris.cls:
+    def _serialize_pickle(message: Any) -> Any:
         pickle_string = codecs.encode(pickle.dumps(message), "base64").decode()
-        msg = iris.cls('IOP.PickleMessage')._New()
+        msg = _iris.get_iris().cls('IOP.PickleMessage')._New()
         msg.classname = f"{message.__class__.__module__}.{message.__class__.__name__}"
         msg.jstr = _Utils.string_to_stream(pickle_string)
         return msg
 
     @staticmethod
-    def _deserialize_pickle(serial: iris.cls) -> Any:
+    def _deserialize_pickle(serial: Any) -> Any:
         string = _Utils.stream_to_string(serial.jstr)
         return pickle.loads(codecs.decode(string.encode(), "base64"))