diff --git a/ssh_utilities/abstract/__init__.py b/ssh_utilities/abstract/__init__.py index d650095..2d4efb7 100644 --- a/ssh_utilities/abstract/__init__.py +++ b/ssh_utilities/abstract/__init__.py @@ -2,7 +2,8 @@ from ._builtins import BuiltinsABC from ._connection import ConnectionABC -from ._os import OsABC, OsPathABC, DirEntryABC +from ._os import OsABC, DirEntryABC +from ._os_path import OsPathABC from ._pathlib import PathlibABC from ._shutil import ShutilABC from ._subprocess import SubprocessABC diff --git a/ssh_utilities/abstract/_os.py b/ssh_utilities/abstract/_os.py index d2c45be..31f88ed 100644 --- a/ssh_utilities/abstract/_os.py +++ b/ssh_utilities/abstract/_os.py @@ -1,4 +1,4 @@ -"""Template module for all os classes.""" +"""Template module for all os classes and methods.""" import logging from abc import ABC, abstractmethod @@ -8,7 +8,7 @@ from ..typeshed import _ONERROR, _SPATH from . import _ATTRIBUTES -__all__ = ["OsPathABC", "OsABC", "DirEntryABC"] +__all__ = ["OsABC", "DirEntryABC"] logging.getLogger(__name__) @@ -117,162 +117,6 @@ def stat(self, *, follow_symlinks: bool = True) -> "_ATTRIBUTES": raise NotImplementedError -class OsPathABC(ABC): - """`os.path` module drop-in replacement base.""" - - __name__: str - __abstractmethods__: FrozenSet[str] - - @abstractmethod - def isfile(self, path: "_SPATH") -> bool: - """Check if path points to a file. - - Parameters - ---------- - path: :const:`ssh_utilities.typeshed._SPATH` - path to check - - Returns - ------- - bool - check result - - Raises - ------ - IOError - if file could not be accessed - """ - raise NotImplementedError - - @abstractmethod - def isdir(self, path: "_SPATH") -> bool: - """Check if path points to directory. - - Parameters - ---------- - path: :const:`ssh_utilities.typeshed._SPATH` - path to check - - Returns - ------- - bool - check result - - Raises - ------ - IOError - if dir could not be accessed - """ - raise NotImplementedError - - @abstractmethod - def exists(self, path: "_SPATH") -> bool: - """Check if path exists in filesystem. - - Parameters - ---------- - path: :const:`ssh_utilities.typeshed._SPATH` - path to check - - Returns - ------- - bool - check result - """ - raise NotImplementedError - - @abstractmethod - def islink(self, path: "_SPATH") -> bool: - """Check if path points to symbolic link. - - Parameters - ---------- - path: :const:`ssh_utilities.typeshed._SPATH` - path to check - - Returns - ------- - bool - check result - - Raises - ------ - IOError - if dir could not be accessed - """ - raise NotImplementedError - - @abstractmethod - def realpath(self, path: "_SPATH") -> str: - """Return the canonical path of the specified filename. - - Eliminates any symbolic links encountered in the path. - - Parameters - ---------- - path : :const:`ssh_utilities.typeshed._SPATH` - path to resolve - - Returns - ------- - str - string representation of the resolved path - """ - raise NotImplementedError - - @abstractmethod - def getsize(self, path: "_SPATH") -> int: - """Return the size of path in bytes. - - Parameters - ---------- - path : :const:`ssh_utilities.typeshed._SPATH` - path to file/directory - - Returns - ------- - int - size in bytes - - Raises - ------ - OsError - if the file does not exist or is inaccessible - """ - raise NotImplementedError - - @abstractmethod - def join(self, path: "_SPATH", *paths: "_SPATH") -> str: - """Join one or more path components intelligently. - - The return value is the concatenation of path and any members of - *paths with exactly one directory separator following each non-empty - part except the last, meaning that the result will only end - in a separator if the last part is empty. If a component is - an absolute path, all previous components are thrown away and - joining continues from the absolute path component. On Windows, - the drive letter is not reset when an absolute path component - (e.g., 'foo') is encountered. If a component contains a drive letter, - all previous components are thrown away and the drive letter is reset. - Note that since there is a current directory for each drive, - os.path.join("c:", "foo") represents a path relative to the current - directory on drive C: (c:foo), not c:/foo. - - Parameters - ---------- - path : :const:`ssh_utilities.typeshed._SPATH` - the starting path part - *paths : :const:`ssh_utilities.typeshed._SPATH` - path parts to join to the first one - - Returns - ------- - str - joined path parts - """ - raise NotImplementedError - - class OsABC(ABC, Generic[_Os1, _Os2, _Os3, _Os4, _Os5, _Os6]): """`os` module drop-in replacement base.""" @@ -454,7 +298,7 @@ def replace(self, src: "_SPATH", dst: "_SPATH", *, ------ OsError If dst is a directory, the operation will fail with an OSError, - + Warnings -------- If dst exists and is a file, it will be replaced silently diff --git a/ssh_utilities/abstract/_os_path.py b/ssh_utilities/abstract/_os_path.py new file mode 100644 index 0000000..2d60fef --- /dev/null +++ b/ssh_utilities/abstract/_os_path.py @@ -0,0 +1,168 @@ +"""Template module for all os.path classes and methods.""" + +import logging +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, FrozenSet + +if TYPE_CHECKING: + from ..typeshed import _SPATH + +__all__ = ["OsPathABC"] + +logging.getLogger(__name__) + + +class OsPathABC(ABC): + """`os.path` module drop-in replacement base.""" + + __name__: str + __abstractmethods__: FrozenSet[str] + + @abstractmethod + def isfile(self, path: "_SPATH") -> bool: + """Check if path points to a file. + + Parameters + ---------- + path: :const:`ssh_utilities.typeshed._SPATH` + path to check + + Returns + ------- + bool + check result + + Raises + ------ + IOError + if file could not be accessed + """ + raise NotImplementedError + + @abstractmethod + def isdir(self, path: "_SPATH") -> bool: + """Check if path points to directory. + + Parameters + ---------- + path: :const:`ssh_utilities.typeshed._SPATH` + path to check + + Returns + ------- + bool + check result + + Raises + ------ + IOError + if dir could not be accessed + """ + raise NotImplementedError + + @abstractmethod + def exists(self, path: "_SPATH") -> bool: + """Check if path exists in filesystem. + + Parameters + ---------- + path: :const:`ssh_utilities.typeshed._SPATH` + path to check + + Returns + ------- + bool + check result + """ + raise NotImplementedError + + @abstractmethod + def islink(self, path: "_SPATH") -> bool: + """Check if path points to symbolic link. + + Parameters + ---------- + path: :const:`ssh_utilities.typeshed._SPATH` + path to check + + Returns + ------- + bool + check result + + Raises + ------ + IOError + if dir could not be accessed + """ + raise NotImplementedError + + @abstractmethod + def realpath(self, path: "_SPATH") -> str: + """Return the canonical path of the specified filename. + + Eliminates any symbolic links encountered in the path. + + Parameters + ---------- + path : :const:`ssh_utilities.typeshed._SPATH` + path to resolve + + Returns + ------- + str + string representation of the resolved path + """ + raise NotImplementedError + + @abstractmethod + def getsize(self, path: "_SPATH") -> int: + """Return the size of path in bytes. + + Parameters + ---------- + path : :const:`ssh_utilities.typeshed._SPATH` + path to file/directory + + Returns + ------- + int + size in bytes + + Raises + ------ + OsError + if the file does not exist or is inaccessible + """ + raise NotImplementedError + + @abstractmethod + def join(self, path: "_SPATH", *paths: "_SPATH") -> str: + """Join one or more path components intelligently. + + The return value is the concatenation of path and any members of + *paths with exactly one directory separator following each non-empty + part except the last, meaning that the result will only end + in a separator if the last part is empty. If a component is + an absolute path, all previous components are thrown away and + joining continues from the absolute path component. On Windows, + the drive letter is not reset when an absolute path component + (e.g., 'foo') is encountered. If a component contains a drive letter, + all previous components are thrown away and the drive letter is reset. + Note that since there is a current directory for each drive, + os.path.join("c:", "foo") represents a path relative to the current + directory on drive C: (c:foo), not c:/foo. + + Parameters + ---------- + path : :const:`ssh_utilities.typeshed._SPATH` + the starting path part + *paths : :const:`ssh_utilities.typeshed._SPATH` + path parts to join to the first one + + Returns + ------- + str + joined path parts + """ + raise NotImplementedError diff --git a/ssh_utilities/local/__init__.py b/ssh_utilities/local/__init__.py index 3b91bfa..1396e6a 100644 --- a/ssh_utilities/local/__init__.py +++ b/ssh_utilities/local/__init__.py @@ -2,6 +2,7 @@ # ! preserve this order otherwise import fail from ._os import Os +from ._os_path import OsPath from ._builtins import Builtins from ._pathlib import Pathlib from ._shutil import Shutil @@ -9,4 +10,4 @@ from .local import LocalConnection __all__ = ["LocalConnection", "Builtins", "Os", "Pathlib", "Shutil", - "Subprocess"] + "Subprocess", "OsPath"] diff --git a/ssh_utilities/local/_os.py b/ssh_utilities/local/_os.py index 361be3f..349fd6c 100644 --- a/ssh_utilities/local/_os.py +++ b/ssh_utilities/local/_os.py @@ -10,7 +10,8 @@ except ImportError: from typing_extensions import Literal # python < 3.8 -from ..abstract import OsABC, OsPathABC +from ..abstract import OsABC +from ._os_path import OsPath if TYPE_CHECKING: from ..typeshed import _SPATH @@ -34,10 +35,10 @@ class Os(OsABC): def __init__(self, connection: "LocalConnection") -> None: self.c = connection - self._path = OsPathLocal(connection) # type: ignore + self._path = OsPath(connection) # type: ignore @property - def path(self) -> "OsPathLocal": + def path(self) -> OsPath: return self._path def scandir(self, path: "_SPATH"): @@ -113,32 +114,3 @@ def name(self) -> Literal["nt", "posix", "java"]: def walk(self, top: "_SPATH", topdown: bool = True, onerror=None, followlinks: bool = False) -> os.walk: return os.walk(top, topdown, onerror, followlinks) - - -class OsPathLocal(OsPathABC): - """Drop in replacement for `os.path` module.""" - - def __init__(self, connection: "LocalConnection") -> None: - self.c = connection - - def isfile(self, path: "_SPATH") -> bool: - return os.path.isfile(self.c._path2str(path)) - - def isdir(self, path: "_SPATH") -> bool: - return os.path.isdir(self.c._path2str(path)) - - def exists(self, path: "_SPATH") -> bool: - return os.path.exists(self.c._path2str(path)) - - def islink(self, path: "_SPATH") -> bool: - return os.path.islink(self.c._path2str(path)) - - def realpath(self, path: "_SPATH") -> str: - return os.path.realpath(self.c._path2str(path)) - - def getsize(self, path: "_SPATH") -> int: - return os.path.getsize(self.c._path2str(path)) - - def join(self, path: "_SPATH", *paths: "_SPATH") -> str: - return os.path.join(self.c._path2str(path), - *[self.c._path2str(p) for p in paths]) diff --git a/ssh_utilities/local/_os_path.py b/ssh_utilities/local/_os_path.py new file mode 100644 index 0000000..3bcfab5 --- /dev/null +++ b/ssh_utilities/local/_os_path.py @@ -0,0 +1,44 @@ +"""Local connection os.path methods.""" + +import logging +import os +from typing import TYPE_CHECKING + +from ..abstract import OsPathABC + +if TYPE_CHECKING: + from ..typeshed import _SPATH + from .local import LocalConnection + +__all__ = ["OsPath"] + +logging.getLogger(__name__) + + +class OsPath(OsPathABC): + """Drop in replacement for `os.path` module.""" + + def __init__(self, connection: "LocalConnection") -> None: + self.c = connection + + def isfile(self, path: "_SPATH") -> bool: + return os.path.isfile(self.c._path2str(path)) + + def isdir(self, path: "_SPATH") -> bool: + return os.path.isdir(self.c._path2str(path)) + + def exists(self, path: "_SPATH") -> bool: + return os.path.exists(self.c._path2str(path)) + + def islink(self, path: "_SPATH") -> bool: + return os.path.islink(self.c._path2str(path)) + + def realpath(self, path: "_SPATH") -> str: + return os.path.realpath(self.c._path2str(path)) + + def getsize(self, path: "_SPATH") -> int: + return os.path.getsize(self.c._path2str(path)) + + def join(self, path: "_SPATH", *paths: "_SPATH") -> str: + return os.path.join(self.c._path2str(path), + *[self.c._path2str(p) for p in paths]) diff --git a/ssh_utilities/multi_connection/_delegated.py b/ssh_utilities/multi_connection/_delegated.py index c2cf629..5dc6158 100644 --- a/ssh_utilities/multi_connection/_delegated.py +++ b/ssh_utilities/multi_connection/_delegated.py @@ -5,10 +5,11 @@ import logging from typing import TYPE_CHECKING, Union +from ..abstract import OsPathABC if TYPE_CHECKING: - from ..abstract import (BuiltinsABC, ConnectionABC, OsABC, OsPathABC, - PathlibABC, ShutilABC, SubprocessABC) + from ..abstract import (BuiltinsABC, ConnectionABC, OsABC, + PathlibABC, ShutilABC, SubprocessABC) from ..local import LocalConnection from ..remote import SSHConnection from .multi_connection import MultiConnection diff --git a/ssh_utilities/remote/__init__.py b/ssh_utilities/remote/__init__.py index ebc7835..a95874b 100644 --- a/ssh_utilities/remote/__init__.py +++ b/ssh_utilities/remote/__init__.py @@ -6,6 +6,7 @@ # ! preserve this order otherwise import fail from ._os import Os +from ._os_path import OsPath from ._builtins import Builtins from ._pathlib import Pathlib from ._shutil import Shutil @@ -13,4 +14,4 @@ from .remote import SSHConnection __all__ = ["SSHConnection", "PIPE", "STDOUT", "DEVNULL", "Builtins", "Os", - "Pathlib", "Shutil", "Subprocess"] + "Pathlib", "Shutil", "Subprocess", "OsPath"] diff --git a/ssh_utilities/remote/_os.py b/ssh_utilities/remote/_os.py index 7d85083..37c49ac 100644 --- a/ssh_utilities/remote/_os.py +++ b/ssh_utilities/remote/_os.py @@ -2,8 +2,6 @@ import logging import os -from ntpath import join as njoin -from posixpath import join as pjoin from stat import S_ISDIR, S_ISLNK, S_ISREG from typing import TYPE_CHECKING, Iterator, List, Optional @@ -12,11 +10,12 @@ except ImportError: from typing_extensions import Literal # python < 3.8 -from ..abstract import DirEntryABC, OsABC, OsPathABC +from ..abstract import DirEntryABC, OsABC from ..constants import G, R from ..exceptions import CalledProcessError, UnknownOsError from ..utils import lprint from ._connection_wrapper import check_connections +from ._os_path import OsPath if TYPE_CHECKING: from paramiko.sftp_attr import SFTPAttributes @@ -29,7 +28,6 @@ log = logging.getLogger(__name__) -# TODO get the follow_symlinks arguments working class DirEntryRemote(DirEntryABC): name: str @@ -84,10 +82,10 @@ class Os(OsABC): def __init__(self, connection: "SSHConnection") -> None: self.c = connection - self._path = OsPathRemote(connection) + self._path = OsPath(connection) @property - def path(self) -> "OsPathRemote": + def path(self) -> OsPath: return self._path @check_connections @@ -335,79 +333,6 @@ def walk(self, top: "_SPATH", topdown: bool = True, yield x -# alternative to os.path module -class OsPathRemote(OsPathABC): - """Drop in replacement for `os.path` module.""" - - def __init__(self, connection: "SSHConnection") -> None: - self.c = connection - - @check_connections(exclude_exceptions=IOError) - def isfile(self, path: "_SPATH") -> bool: - # have to call without decorators, otherwise FileNotFoundError - # does not propagate - unwrap = self.c.os.stat.__wrapped__ - try: - return S_ISREG(unwrap(self, self.c._path2str(path)).st_mode) - except FileNotFoundError: - return False - - @check_connections(exclude_exceptions=IOError) - def isdir(self, path: "_SPATH") -> bool: - # have to call without decorators, otherwise FileNotFoundError - # does not propagate - unwrap = self.c.os.stat.__wrapped__ - try: - return S_ISDIR(unwrap(self, self.c._path2str(path)).st_mode) - except FileNotFoundError: - return False - - @check_connections - def exists(self, path: "_SPATH") -> bool: - # have to call without decorators, otherwise FileNotFoundError - # does not propagate - unwrap = self.c.os.stat.__wrapped__ - try: - unwrap(self, self.c._path2str(path)) - except FileNotFoundError: - return False - else: - return True - - @check_connections(exclude_exceptions=IOError) - def islink(self, path: "_SPATH") -> bool: - # have to call without decorators, otherwise FileNotFoundError - # does not propagate - unwrap = self.c.os.stat.__wrapped__ - try: - return S_ISLNK(unwrap(self, self.c._path2str(path)).st_mode) - except FileNotFoundError: - return False - - @check_connections - def realpath(self, path: "_SPATH") -> str: - return self.c.sftp.normalize(self.c._path2str(path)) - - def getsize(self, path: "_SPATH") -> int: - - size = self.c.os.stat(path).st_size - - if size: - return size - else: - raise OSError(f"Could not get size of file: {path}") - - @check_connections - def join(self, path: "_SPATH", *paths: "_SPATH") -> str: - - if self.c.os.name == "nt": - return njoin(self.c._path2str(path), - *[self.c._path2str(p) for p in paths]) - else: - return pjoin(self.c._path2str(path), - *[self.c._path2str(p) for p in paths]) - - class Scandir: """Reads directory contents and yields as DirEntry objects. diff --git a/ssh_utilities/remote/_os_path.py b/ssh_utilities/remote/_os_path.py new file mode 100644 index 0000000..0f7cf66 --- /dev/null +++ b/ssh_utilities/remote/_os_path.py @@ -0,0 +1,92 @@ +"""Remote connection os.path methods.""" + +import logging +from ntpath import join as njoin +from posixpath import join as pjoin +from stat import S_ISDIR, S_ISLNK, S_ISREG +from typing import TYPE_CHECKING + +from ..abstract import OsPathABC +from ._connection_wrapper import check_connections + +if TYPE_CHECKING: + from ..typeshed import _SPATH + from .remote import SSHConnection + + +__all__ = ["OsPath"] + +log = logging.getLogger(__name__) + + +# alternative to os.path module +class OsPath(OsPathABC): + """Drop in replacement for `os.path` module.""" + + def __init__(self, connection: "SSHConnection") -> None: + self.c = connection + + @check_connections(exclude_exceptions=IOError) + def isfile(self, path: "_SPATH") -> bool: + # have to call without decorators, otherwise FileNotFoundError + # does not propagate + unwrap = self.c.os.stat.__wrapped__ + try: + return S_ISREG(unwrap(self, self.c._path2str(path)).st_mode) + except FileNotFoundError: + return False + + @check_connections(exclude_exceptions=IOError) + def isdir(self, path: "_SPATH") -> bool: + # have to call without decorators, otherwise FileNotFoundError + # does not propagate + unwrap = self.c.os.stat.__wrapped__ + try: + return S_ISDIR(unwrap(self, self.c._path2str(path)).st_mode) + except FileNotFoundError: + return False + + @check_connections + def exists(self, path: "_SPATH") -> bool: + # have to call without decorators, otherwise FileNotFoundError + # does not propagate + unwrap = self.c.os.stat.__wrapped__ + try: + unwrap(self, self.c._path2str(path)) + except FileNotFoundError: + return False + else: + return True + + @check_connections(exclude_exceptions=IOError) + def islink(self, path: "_SPATH") -> bool: + # have to call without decorators, otherwise FileNotFoundError + # does not propagate + unwrap = self.c.os.stat.__wrapped__ + try: + return S_ISLNK(unwrap(self, self.c._path2str(path)).st_mode) + except FileNotFoundError: + return False + + @check_connections + def realpath(self, path: "_SPATH") -> str: + return self.c.sftp.normalize(self.c._path2str(path)) + + def getsize(self, path: "_SPATH") -> int: + + size = self.c.os.stat(path).st_size + + if size: + return size + else: + raise OSError(f"Could not get size of file: {path}") + + @check_connections + def join(self, path: "_SPATH", *paths: "_SPATH") -> str: + + if self.c.os.name == "nt": + return njoin(self.c._path2str(path), + *[self.c._path2str(p) for p in paths]) + else: + return pjoin(self.c._path2str(path), + *[self.c._path2str(p) for p in paths])