[OMCSession] update OMCPath to use OMPathABC as baseline and further cleanup

Author: syntron

OMPython/OMCSession.py

@@ -255,13 +255,15 @@ def getClassNames(self, className=None, recursive=False, qualified=False, sort=F
         return self._ask(question='getClassNames', opt=opt)
 
 
-class OMCPathReal(pathlib.PurePosixPath):
+class OMPathABC(pathlib.PurePosixPath, metaclass=abc.ABCMeta):
     """
-    Implementation of a basic (PurePosix)Path object which uses OMC as backend. The connection to OMC is provided via an
-    instances of OMCSession* classes.
+    Implementation of a basic (PurePosix)Path object to be used within OMPython. The derived classes can use OMC as
+    backend and - thus - work on different configurations like docker or WSL. The connection to OMC is provided via an
+    instances of classes derived from BaseSession.
 
-    PurePosixPath is selected to cover usage of OMC in docker or via WSL. Usage of specialised function could result in
-    errors as well as usage on a Windows system due to slightly different definitions (PureWindowsPath).
+    PurePosixPath is selected as it covers all but Windows systems (Linux, docker, WSL). However, the code is written
+    such that possible Windows system are taken into account. Nevertheless, the overall functionality is limited
+    compared to standard pathlib.Path objects.
     """
 
     def __init__(self, *path, session: OMCSession) -> None:
@@ -272,46 +274,122 @@ def with_segments(self, *pathsegments):
         """
         Create a new OMCPath object with the given path segments.
 
-        The original definition of Path is overridden to ensure the OMC session is set.
+        The original definition of Path is overridden to ensure the session data is set.
         """
         return type(self)(*pathsegments, session=self._session)
 
-    def is_file(self, *, follow_symlinks=True) -> bool:
+    @abc.abstractmethod
+    def is_file(self) -> bool:
+        """
+        Check if the path is a regular file.
+        """
+
+    @abc.abstractmethod
+    def is_dir(self) -> bool:
+        """
+        Check if the path is a directory.
+        """
+
+    @abc.abstractmethod
+    def is_absolute(self):
+        """
+        Check if the path is an absolute path.
+        """
+
+    @abc.abstractmethod
+    def read_text(self) -> str:
+        """
+        Read the content of the file represented by this path as text.
+        """
+
+    @abc.abstractmethod
+    def write_text(self, data: str):
+        """
+        Write text data to the file represented by this path.
+        """
+
+    @abc.abstractmethod
+    def mkdir(self, parents: bool = True, exist_ok: bool = False):
+        """
+        Create a directory at the path represented by this class.
+
+        The argument parents with default value True exists to ensure compatibility with the fallback solution for
+        Python < 3.12. In this case, pathlib.Path is used directly and this option ensures, that missing parent
+        directories are also created.
+        """
+
+    @abc.abstractmethod
+    def cwd(self):
+        """
+        Returns the current working directory as an OMPathBase object.
+        """
+
+    @abc.abstractmethod
+    def unlink(self, missing_ok: bool = False) -> None:
+        """
+        Unlink (delete) the file or directory represented by this path.
+        """
+
+    @abc.abstractmethod
+    def resolve(self, strict: bool = False):
+        """
+        Resolve the path to an absolute path.
+        """
+
+    def absolute(self):
+        """
+        Resolve the path to an absolute path. Just a wrapper for resolve().
+        """
+        return self.resolve()
+
+    def exists(self) -> bool:
+        """
+        Semi replacement for pathlib.Path.exists().
+        """
+        return self.is_file() or self.is_dir()
+
+    @abc.abstractmethod
+    def size(self) -> int:
+        """
+        Get the size of the file in bytes - this is an extra function and the best we can do using OMC.
+        """
+
+
+class _OMCPath(OMPathABC):
+    """
+    Implementation of a OMPathBase using OMC as backend. The connection to OMC is provided via an instances of an
+    OMCSession* classes.
+    """
+
+    def is_file(self) -> bool:
         """
         Check if the path is a regular file.
         """
         return self._session.sendExpression(f'regularFileExists("{self.as_posix()}")')
 
-    def is_dir(self, *, follow_symlinks=True) -> bool:
+    def is_dir(self) -> bool:
         """
         Check if the path is a directory.
         """
         return self._session.sendExpression(f'directoryExists("{self.as_posix()}")')
 
     def is_absolute(self):
         """
-        Check if the path is an absolute path considering the possibility that we are running locally on Windows. This
-        case needs special handling as the definition of is_absolute() differs.
+        Check if the path is an absolute path.
         """
         if isinstance(self._session, OMCSessionLocal) and platform.system() == 'Windows':
             return pathlib.PureWindowsPath(self.as_posix()).is_absolute()
         return super().is_absolute()
 
-    def read_text(self, encoding=None, errors=None, newline=None) -> str:
+    def read_text(self) -> str:
         """
         Read the content of the file represented by this path as text.
-
-        The additional arguments `encoding`, `errors` and `newline` are only defined for compatibility with Path()
-        definition.
         """
         return self._session.sendExpression(f'readFile("{self.as_posix()}")')
 
-    def write_text(self, data: str, encoding=None, errors=None, newline=None):
+    def write_text(self, data: str):
         """
         Write text data to the file represented by this path.
-
-        The additional arguments `encoding`, `errors`, and `newline` are only defined for compatibility with Path()
-        definitions.
         """
         if not isinstance(data, str):
             raise TypeError(f"data must be str, not {data.__class__.__name__}")
@@ -321,11 +399,13 @@ def write_text(self, data: str, encoding=None, errors=None, newline=None):
 
         return len(data)
 
-    def mkdir(self, mode=0o777, parents=False, exist_ok=False):
+    def mkdir(self, parents: bool = True, exist_ok: bool = False):
         """
-        Create a directory at the path represented by this OMCPath object.
+        Create a directory at the path represented by this class.
 
-        The additional arguments `mode`, and `parents` are only defined for compatibility with Path() definitions.
+        The argument parents with default value True exists to ensure compatibility with the fallback solution for
+        Python < 3.12. In this case, pathlib.Path is used directly and this option ensures, that missing parent
+        directories are also created.
         """
         if self.is_dir() and not exist_ok:
             raise FileExistsError(f"Directory {self.as_posix()} already exists!")
@@ -334,7 +414,7 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False):
 
     def cwd(self):
         """
-        Returns the current working directory as an OMCPath object.
+        Returns the current working directory as an OMPathBase object.
         """
         cwd_str = self._session.sendExpression('cd()')
         return OMCPath(cwd_str, session=self._session)
@@ -387,19 +467,6 @@ def _omc_resolve(self, pathstr: str) -> str:
 
         return pathstr_resolved
 
-    def absolute(self):
-        """
-        Resolve the path to an absolute path. This is done by calling resolve() as it is the best we can do
-        using OMC functions.
-        """
-        return self.resolve(strict=True)
-
-    def exists(self, follow_symlinks=True) -> bool:
-        """
-        Semi replacement for pathlib.Path.exists().
-        """
-        return self.is_file() or self.is_dir()
-
     def size(self) -> int:
         """
         Get the size of the file in bytes - this is an extra function and the best we can do using OMC.
@@ -414,47 +481,47 @@ def size(self) -> int:
         raise OMCSessionException(f"Error reading file size for path {self.as_posix()}!")
 
 
-if sys.version_info < (3, 12):
+class OMPathCompatibility(pathlib.Path):
+    """
+    Compatibility class for OMPathBase in Python < 3.12. This allows to run all code which uses OMPathBase (mainly
+    ModelicaSystem) on these Python versions. There are remaining limitation as only local execution is possible.
+    """
+
+    # modified copy of pathlib.Path.__new__() definition
+    def __new__(cls, *args, **kwargs):
+        logger.warning("Python < 3.12 - using a version of class OMCPath "
+                       "based on pathlib.Path for local usage only.")
+
+        if cls is OMPathCompatibility:
+            cls = OMPathCompatibilityWindows if os.name == 'nt' else OMPathCompatibilityPosix
+        self = cls._from_parts(args)
+        if not self._flavour.is_supported:
+            raise NotImplementedError(f"cannot instantiate {cls.__name__} on your system")
+        return self
 
-    class OMCPathCompatibility(pathlib.Path):
+    def size(self) -> int:
         """
-        Compatibility class for OMCPath in Python < 3.12. This allows to run all code which uses OMCPath (mainly
-        ModelicaSystem) on these Python versions. There is one remaining limitation: only OMCProcessLocal will work as
-        OMCPathCompatibility is based on the standard pathlib.Path implementation.
+        Needed compatibility function to have the same interface as OMCPathReal
         """
+        return self.stat().st_size
 
-        # modified copy of pathlib.Path.__new__() definition
-        def __new__(cls, *args, **kwargs):
-            logger.warning("Python < 3.12 - using a version of class OMCPath "
-                           "based on pathlib.Path for local usage only.")
-
-            if cls is OMCPathCompatibility:
-                cls = OMCPathCompatibilityWindows if os.name == 'nt' else OMCPathCompatibilityPosix
-            self = cls._from_parts(args)
-            if not self._flavour.is_supported:
-                raise NotImplementedError(f"cannot instantiate {cls.__name__} on your system")
-            return self
 
-        def size(self) -> int:
-            """
-            Needed compatibility function to have the same interface as OMCPathReal
-            """
-            return self.stat().st_size
+class OMPathCompatibilityPosix(pathlib.PosixPath, OMPathCompatibility):
+    """
+    Compatibility class for OMCPath on Posix systems (Python < 3.12)
+    """
 
-    class OMCPathCompatibilityPosix(pathlib.PosixPath, OMCPathCompatibility):
-        """
-        Compatibility class for OMCPath on Posix systems (Python < 3.12)
-        """
 
-    class OMCPathCompatibilityWindows(pathlib.WindowsPath, OMCPathCompatibility):
-        """
-        Compatibility class for OMCPath on Windows systems (Python < 3.12)
-        """
+class OMPathCompatibilityWindows(pathlib.WindowsPath, OMPathCompatibility):
+    """
+    Compatibility class for OMCPath on Windows systems (Python < 3.12)
+    """
 
-    OMCPath = OMCPathCompatibility
 
+if sys.version_info < (3, 12):
+    OMCPath = OMPathCompatibility
 else:
-    OMCPath = OMCPathReal
+    OMCPath = _OMCPath
 
 
 class ModelExecutionException(Exception):
@@ -576,13 +643,13 @@ def escape_str(value: str) -> str:
         """
         return OMCSession.escape_str(value=value)
 
-    def omcpath(self, *path) -> OMCPath:
+    def omcpath(self, *path) -> OMPathABC:
         """
         Create an OMCPath object based on the given path segments and the current OMC process definition.
         """
         return self.omc_process.omcpath(*path)
 
-    def omcpath_tempdir(self, tempdir_base: Optional[OMCPath] = None) -> OMCPath:
+    def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC:
         """
         Get a temporary directory using OMC. It is our own implementation as non-local usage relies on OMC to run all
         filesystem related access.
@@ -755,11 +822,10 @@ def escape_str(value: str) -> str:
         """
         return value.replace("\\", "\\\\").replace('"', '\\"')
 
-    def model_execution_prefix(self, cwd: Optional[OMCPath] = None) -> list[str]:
+    def model_execution_prefix(self, cwd: Optional[OMPathABC] = None) -> list[str]:
         """
         Helper function which returns a command prefix needed for docker and WSL. It defaults to an empty list.
         """
-
         return []
 
     def get_version(self) -> str:
@@ -768,14 +834,14 @@ def get_version(self) -> str:
         """
         return self.sendExpression("getVersion()", parsed=True)
 
-    def set_workdir(self, workdir: OMCPath) -> None:
+    def set_workdir(self, workdir: OMPathABC) -> None:
         """
         Set the workdir for this session.
         """
         exp = f'cd("{workdir.as_posix()}")'
         self.sendExpression(exp)
 
-    def omcpath(self, *path) -> OMCPath:
+    def omcpath(self, *path) -> OMPathABC:
         """
         Create an OMCPath object based on the given path segments and the current OMCSession* class.
         """
@@ -788,7 +854,7 @@ def omcpath(self, *path) -> OMCPath:
             raise OMCSessionException("OMCPath is supported for Python < 3.12 only if OMCSessionLocal is used!")
         return OMCPath(*path, session=self)
 
-    def omcpath_tempdir(self, tempdir_base: Optional[OMCPath] = None) -> OMCPath:
+    def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC:
         """
         Get a temporary directory using OMC. It is our own implementation as non-local usage relies on OMC to run all
         filesystem related access.
@@ -805,10 +871,10 @@ def omcpath_tempdir(self, tempdir_base: Optional[OMCPath] = None) -> OMCPath:
         return self._tempdir(tempdir_base=tempdir_base)
 
     @staticmethod
-    def _tempdir(tempdir_base: OMCPath) -> OMCPath:
+    def _tempdir(tempdir_base: OMPathABC) -> OMPathABC:
         names = [str(uuid.uuid4()) for _ in range(100)]
 
-        tempdir: Optional[OMCPath] = None
+        tempdir: Optional[OMPathABC] = None
         for name in names:
             # create a unique temporary directory name
             tempdir = tempdir_base / name
@@ -1222,15 +1288,15 @@ def get_docker_container_id(self) -> str:
 
         return self._docker_container_id
 
-    def model_execution_prefix(self, cwd: Optional[OMCPath] = None) -> list[str]:
+    def model_execution_prefix(self, cwd: Optional[OMPathABC] = None) -> list[str]:
         """
         Helper function which returns a command prefix needed for docker and WSL. It defaults to an empty list.
         """
         docker_cmd = [
             "docker", "exec",
             "--user", str(self._getuid()),
-            ]
-        if isinstance(cwd, OMCPath):
+        ]
+        if isinstance(cwd, OMPathABC):
             docker_cmd += ["--workdir", cwd.as_posix()]
         docker_cmd += self._docker_extra_args
         if isinstance(self._docker_container_id, str):
@@ -1500,7 +1566,7 @@ def __init__(
         # connect to the running omc instance using ZMQ
         self._omc_port = self._omc_port_get()
 
-    def model_execution_prefix(self, cwd: Optional[OMCPath] = None) -> list[str]:
+    def model_execution_prefix(self, cwd: Optional[OMPathABC] = None) -> list[str]:
         """
         Helper function which returns a command prefix needed for docker and WSL. It defaults to an empty list.
         """
@@ -1510,7 +1576,7 @@ def model_execution_prefix(self, cwd: Optional[OMCPath] = None) -> list[str]:
             wsl_cmd += ['--distribution', self._wsl_distribution]
         if isinstance(self._wsl_user, str):
             wsl_cmd += ['--user', self._wsl_user]
-        if isinstance(cwd, OMCPath):
+        if isinstance(cwd, OMPathABC):
             wsl_cmd += ['--cd', cwd.as_posix()]
         wsl_cmd += ['--']