bast
c5c0d42eac
All checks were successful
MT5 Helm GitOps Deploy / build-and-deploy (push) Has been skipped
406 lines
12 KiB
Python
406 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
__lazy_modules__ = {
|
|
"contextlib",
|
|
"glob",
|
|
"grp",
|
|
"plumbum.lib",
|
|
"plumbum.path.remote",
|
|
"pwd",
|
|
"shutil",
|
|
"urllib",
|
|
"urllib.parse",
|
|
"urllib.request",
|
|
}
|
|
|
|
import errno
|
|
import glob
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import urllib.parse as urlparse
|
|
import urllib.request as urllib
|
|
from contextlib import contextmanager
|
|
from typing import IO, TYPE_CHECKING
|
|
|
|
from plumbum.lib import IS_WIN32
|
|
from plumbum.path.base import FSUser, Path
|
|
from plumbum.path.remote import RemotePath
|
|
|
|
if TYPE_CHECKING:
|
|
import builtins
|
|
from collections.abc import Generator, Iterator
|
|
|
|
from plumbum._compat.typing import Self
|
|
|
|
if not sys.platform.startswith("win32"):
|
|
from grp import getgrgid, getgrnam
|
|
from pwd import getpwnam, getpwuid
|
|
else:
|
|
|
|
def getpwuid(_x: int) -> tuple[None]:
|
|
return (None,)
|
|
|
|
def getgrgid(_: int) -> tuple[None]:
|
|
return (None,)
|
|
|
|
def getpwnam(_x: str) -> None: # not NoReturn to avoid unreachable warnings
|
|
raise OSError("`getpwnam` not supported")
|
|
|
|
def getgrnam(_x: str) -> None: # not NoReturn to avoid unreachable warnings
|
|
raise OSError("`getgrnam` not supported")
|
|
|
|
|
|
logger = logging.getLogger("plumbum.local")
|
|
|
|
_EMPTY = object()
|
|
|
|
|
|
# ===================================================================================================
|
|
# Local Paths
|
|
# ===================================================================================================
|
|
class LocalPath(Path):
|
|
"""The class implementing local-machine paths"""
|
|
|
|
__slots__ = ()
|
|
|
|
CASE_SENSITIVE = not IS_WIN32
|
|
|
|
def __new__(cls, *parts: str | Path) -> Self:
|
|
if (
|
|
len(parts) == 1
|
|
and isinstance(parts[0], cls)
|
|
and not isinstance(parts[0], LocalWorkdir)
|
|
):
|
|
return parts[0]
|
|
if not parts:
|
|
raise TypeError("At least one path part is required (none given)")
|
|
if any(isinstance(path, RemotePath) for path in parts):
|
|
raise TypeError(f"LocalPath cannot be constructed from {parts!r}")
|
|
return super().__new__(
|
|
cls, os.path.normpath(os.path.join(*(str(p) for p in parts)))
|
|
)
|
|
|
|
@property
|
|
def _path(self) -> str:
|
|
return str(self)
|
|
|
|
def _get_info(self) -> str:
|
|
return self._path
|
|
|
|
def _form(self, *parts: str | Path) -> LocalPath:
|
|
return LocalPath(*parts)
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return os.path.basename(str(self))
|
|
|
|
@property
|
|
def dirname(self) -> LocalPath:
|
|
return LocalPath(os.path.dirname(str(self)))
|
|
|
|
@property
|
|
def suffix(self) -> str:
|
|
return os.path.splitext(str(self))[1]
|
|
|
|
@property
|
|
def suffixes(self) -> list[str]:
|
|
exts = []
|
|
base = str(self)
|
|
while True:
|
|
base, ext = os.path.splitext(base)
|
|
if ext:
|
|
exts.append(ext)
|
|
else:
|
|
return list(reversed(exts))
|
|
|
|
@property
|
|
def uid(self) -> FSUser:
|
|
uid = self.stat().st_uid
|
|
name = getpwuid(uid)[0]
|
|
return FSUser(uid, name)
|
|
|
|
@property
|
|
def gid(self) -> FSUser:
|
|
gid = self.stat().st_gid
|
|
name = getgrgid(gid)[0]
|
|
return FSUser(gid, name)
|
|
|
|
def join(self, *others: str) -> LocalPath: # type: ignore[override]
|
|
return LocalPath(self, *others)
|
|
|
|
def list(self) -> list[LocalPath]:
|
|
return [self / fn for fn in os.listdir(str(self))]
|
|
|
|
def iterdir(self) -> Iterator[LocalPath]:
|
|
try:
|
|
return (self / fn.name for fn in os.scandir(str(self)))
|
|
except AttributeError:
|
|
return (self / fn for fn in os.listdir(str(self)))
|
|
|
|
def is_dir(self) -> bool:
|
|
return os.path.isdir(str(self))
|
|
|
|
def is_file(self) -> bool:
|
|
return os.path.isfile(str(self))
|
|
|
|
def is_symlink(self) -> bool:
|
|
return os.path.islink(str(self))
|
|
|
|
def exists(self) -> bool:
|
|
return os.path.exists(str(self))
|
|
|
|
def stat(self) -> os.stat_result:
|
|
return os.stat(str(self))
|
|
|
|
def with_name(self, name: str) -> LocalPath:
|
|
return LocalPath(self.dirname) / name
|
|
|
|
@property
|
|
def stem(self) -> str:
|
|
return os.path.splitext(self.name)[0]
|
|
|
|
def with_suffix(self, suffix: str, depth: int | None = 1) -> LocalPath:
|
|
if (
|
|
suffix and not suffix.startswith(os.path.extsep)
|
|
) or suffix == os.path.extsep:
|
|
raise ValueError(f"Invalid suffix {suffix!r}")
|
|
name = self.name
|
|
depth = len(self.suffixes) if depth is None else min(depth, len(self.suffixes))
|
|
for _ in range(depth):
|
|
name, _ = os.path.splitext(name)
|
|
return LocalPath(self.dirname) / (name + suffix)
|
|
|
|
def glob(self, pattern: str) -> builtins.list[LocalPath]:
|
|
return self._glob(
|
|
pattern,
|
|
lambda pat: [
|
|
LocalPath(m)
|
|
for m in glob.glob(
|
|
os.path.join(glob.escape(str(self)), pat), recursive=True
|
|
)
|
|
],
|
|
)
|
|
|
|
def delete(self) -> None:
|
|
if not self.exists():
|
|
return
|
|
if self.is_dir():
|
|
shutil.rmtree(str(self))
|
|
else:
|
|
try:
|
|
os.remove(str(self))
|
|
except OSError as ex: # pragma: no cover
|
|
# file might already been removed (a race with other threads/processes)
|
|
if ex.errno != errno.ENOENT:
|
|
raise
|
|
|
|
def move(self, dst: LocalPath | str) -> LocalPath:
|
|
if isinstance(dst, RemotePath):
|
|
raise TypeError(f"Cannot move local path {self} to {dst!r}")
|
|
shutil.move(str(self), str(dst))
|
|
return LocalPath(dst)
|
|
|
|
def copy(self, dst: LocalPath | str, override: bool | None = None) -> LocalPath:
|
|
if isinstance(dst, RemotePath):
|
|
raise TypeError(f"Cannot copy local path {self} to {dst!r}")
|
|
dst = LocalPath(dst)
|
|
if override is False and dst.exists():
|
|
raise TypeError("File exists and override was not specified")
|
|
if override:
|
|
dst.delete()
|
|
if self.is_dir():
|
|
shutil.copytree(str(self), str(dst))
|
|
else:
|
|
dst_dir = LocalPath(dst).dirname
|
|
if not dst_dir.exists():
|
|
dst_dir.mkdir()
|
|
shutil.copy2(str(self), str(dst))
|
|
return dst
|
|
|
|
def mkdir(
|
|
self, mode: int = 0o777, parents: bool = True, exist_ok: bool = True
|
|
) -> None:
|
|
if not self.exists() or not exist_ok:
|
|
try:
|
|
if parents:
|
|
os.makedirs(str(self), mode)
|
|
else:
|
|
os.mkdir(str(self), mode)
|
|
except OSError as ex: # pragma: no cover
|
|
# directory might already exist (a race with other threads/processes)
|
|
if ex.errno != errno.EEXIST or not exist_ok:
|
|
raise
|
|
|
|
def open(self, mode: str = "r", encoding: str | None = None) -> IO[str] | IO[bytes]:
|
|
return open(
|
|
str(self),
|
|
mode,
|
|
encoding=encoding,
|
|
)
|
|
|
|
def read(self, encoding: str | None = None, mode: str = "r") -> str | bytes:
|
|
if encoding and "b" not in mode:
|
|
mode = mode + "b"
|
|
with self.open(mode) as f:
|
|
data = f.read()
|
|
if encoding:
|
|
assert isinstance(data, bytes)
|
|
return data.decode(encoding)
|
|
return data
|
|
|
|
def write(
|
|
self, data: str | bytes, encoding: str | None = None, mode: str | None = None
|
|
) -> None:
|
|
if encoding:
|
|
if isinstance(data, bytes):
|
|
msg = "Must write string data if encoding given"
|
|
raise TypeError(msg)
|
|
data = data.encode(encoding)
|
|
if mode is None:
|
|
mode = "w" if isinstance(data, str) else "wb"
|
|
with self.open(mode) as f:
|
|
f.write(data) # type: ignore[arg-type]
|
|
|
|
def touch(self) -> None:
|
|
with open(str(self), "a", encoding="utf-8"):
|
|
os.utime(str(self), None)
|
|
|
|
def chown(
|
|
self,
|
|
owner: int | str | None = None,
|
|
group: int | str | None = None,
|
|
recursive: bool | None = None,
|
|
) -> None:
|
|
if not hasattr(os, "chown"):
|
|
raise OSError("os.chown() not supported")
|
|
# -1 tells os.chown to leave that side unchanged, so an unspecified
|
|
# owner/group is preserved per-file rather than being forced to the
|
|
# value read off this path (which would re-own children differently).
|
|
uid = (
|
|
-1
|
|
if owner is None
|
|
else (owner if isinstance(owner, int) else getpwnam(owner)[2])
|
|
)
|
|
gid = (
|
|
-1
|
|
if group is None
|
|
else (group if isinstance(group, int) else getgrnam(group)[2])
|
|
)
|
|
os.chown(str(self), uid, gid)
|
|
if recursive or (recursive is None and self.is_dir()):
|
|
for subpath in self.walk():
|
|
os.chown(str(subpath), uid, gid)
|
|
|
|
def chmod(self, mode: int) -> None:
|
|
if not hasattr(os, "chmod"):
|
|
raise OSError("os.chmod() not supported")
|
|
os.chmod(str(self), mode)
|
|
|
|
def access(self, mode: int | str = 0) -> bool:
|
|
return os.access(str(self), self._access_mode_to_flags(mode))
|
|
|
|
def link(self, dst: LocalPath | str) -> None:
|
|
if isinstance(dst, RemotePath):
|
|
raise TypeError(
|
|
f"Cannot create a hardlink from local path {self} to {dst!r}"
|
|
)
|
|
if hasattr(os, "link"):
|
|
os.link(str(self), str(dst))
|
|
else:
|
|
from plumbum.machines.local import local
|
|
|
|
# windows: use mklink
|
|
if self.is_dir():
|
|
local["cmd"]("/C", "mklink", "/D", "/H", str(dst), str(self))
|
|
else:
|
|
local["cmd"]("/C", "mklink", "/H", str(dst), str(self))
|
|
|
|
def symlink(self, dst: LocalPath | str) -> None:
|
|
if isinstance(dst, RemotePath):
|
|
raise TypeError(
|
|
f"Cannot create a symlink from local path {self} to {dst!r}"
|
|
)
|
|
if hasattr(os, "symlink"):
|
|
os.symlink(str(self), str(dst))
|
|
else:
|
|
from plumbum.machines.local import local
|
|
|
|
# windows: use mklink
|
|
if self.is_dir():
|
|
local["cmd"]("/C", "mklink", "/D", str(dst), str(self))
|
|
else:
|
|
local["cmd"]("/C", "mklink", str(dst), str(self))
|
|
|
|
def unlink(self) -> None:
|
|
try:
|
|
if hasattr(os, "symlink") or not self.is_dir():
|
|
os.unlink(str(self))
|
|
else:
|
|
# windows: use rmdir for directories and directory symlinks
|
|
os.rmdir(str(self))
|
|
except OSError as ex: # pragma: no cover
|
|
# file might already been removed (a race with other threads/processes)
|
|
if ex.errno != errno.ENOENT:
|
|
raise
|
|
|
|
def as_uri(self, scheme: str = "file") -> str:
|
|
return urlparse.urljoin(f"{scheme}://", urllib.pathname2url(str(self)))
|
|
|
|
@property
|
|
def drive(self) -> str:
|
|
return os.path.splitdrive(str(self))[0]
|
|
|
|
@property
|
|
def root(self) -> str:
|
|
return os.path.sep
|
|
|
|
|
|
class LocalWorkdir(LocalPath):
|
|
"""Working directory manipulator"""
|
|
|
|
__slots__ = ()
|
|
__hash__ = None # type: ignore[assignment]
|
|
|
|
def __new__(cls) -> Self:
|
|
return super().__new__(cls, os.getcwd())
|
|
|
|
def chdir(self, newdir: LocalPath | str) -> LocalWorkdir:
|
|
"""Changes the current working directory to the given one
|
|
|
|
:param newdir: The destination director (a string or a ``LocalPath``)
|
|
"""
|
|
if isinstance(newdir, RemotePath):
|
|
raise TypeError(f"newdir cannot be {newdir!r}")
|
|
logger.debug("Chdir to %s", newdir)
|
|
os.chdir(str(newdir))
|
|
return self.__class__()
|
|
|
|
def getpath(self) -> LocalPath:
|
|
"""Returns the current working directory as a ``LocalPath`` object"""
|
|
return LocalPath(self._path)
|
|
|
|
@contextmanager
|
|
def __call__(self, newdir: LocalPath | str) -> Generator[LocalWorkdir, None, None]:
|
|
"""A context manager used to ``chdir`` into a directory and then ``chdir`` back to
|
|
the previous location; much like ``pushd``/``popd``.
|
|
|
|
:param newdir: The destination directory (a string or a ``LocalPath``)
|
|
"""
|
|
prev = self._path
|
|
newdir = self.chdir(newdir)
|
|
try:
|
|
yield newdir
|
|
finally:
|
|
self.chdir(prev)
|
|
|
|
|
|
__all__ = [
|
|
"LocalPath",
|
|
"LocalWorkdir",
|
|
]
|
|
|
|
|
|
def __dir__() -> list[str]:
|
|
return list(__all__)
|