Source code for ciowarehouse2.lib.ciopath

"""A class that uniquely represents a file."""

from __future__ import annotations
from os.path import dirname, basename, join, normpath, exists, isdir
from pathlib import Path
from hashlib import sha1
from mimetypes import guess_type

from pyramid.request import Request

LOCAL_DIR = '.local'
INFO_DIR = '.info'
LOCKS_DIR = join(LOCAL_DIR, 'Locks')
MYSELF = '__myself__'


# =============================================================================
[docs] class CioPath(): """A class to manage `CioPath`. :param str wid: Warehouse ID. :param str path: (optional) Relative path inside the warehouse. :param bool is_dir: (default=False) ``True`` if the given `CioPath` is a directory. """ # ------------------------------------------------------------------------- def __init__( self, wid: str | None, path: str | None = None, is_dir: bool = False): """Constructor method.""" if not wid: self.wid: str | None = None self.path: str = '' self._is_dir: bool = False else: self.wid = wid self._is_dir = bool( not path or path == '.' or path.endswith('/') or (path and path != '.' and is_dir)) self.path = '.' if not path else normpath( path[:-1] if path.endswith('/') else path) if self.path.startswith('..'): self.wid = None self.path = '' self._is_dir = False # -------------------------------------------------------------------------
[docs] @classmethod def from_str(cls, string: str | None, is_dir: bool = False) -> CioPath: """Return a new `CioPath` created from a string. :param str string: String representing a `CioPath` """ if not string or ':' not in string: return CioPath(None) wid, path = string.partition(':')[::2] return CioPath(wid, path, is_dir)
# -------------------------------------------------------------------------
[docs] @classmethod def from_paths(cls, wid: str, root: str, paths: list[str] | set[str]) -> list[CioPath]: """Return a list or `CioPath` from a list of relative paths. :param str wid: Warehouse ID. :param str root: Absolute path to the root of the warehouse. :param list paths: List of relative paths inside the given warehouse. """ ciopaths = [] for path in paths: abs_path = join(root, path) if exists(abs_path): ciopaths.append(CioPath(wid, path, isdir(abs_path))) return ciopaths
# -------------------------------------------------------------------------
[docs] @classmethod def from_request(cls, request: Request, is_dir: bool = False) -> CioPath: """Look for a `CioPath` in a request. :type request: pyramid.request.Request :param request: Current request. :param bool is_dir: (default=False) ``True`` if the given `CioPath` is a directory. """ ciopath_list = request.matchdict.get('ciopath') if not ciopath_list: return CioPath(None) if len(ciopath_list) == 1: return CioPath(ciopath_list[0]) return CioPath(ciopath_list[0], '/'.join(ciopath_list[1:]), is_dir)
# ------------------------------------------------------------------------- def __repr__(self) -> str: """Return a string representation of the `CioPath`. :rtype: str """ if self.wid is None: return '' tail = '/' if self._is_dir else '' return f'{self.wid}:{self.path}{tail}' # ------------------------------------------------------------------------- def __eq__(self, other) -> bool: """Override the default implementation.""" if isinstance(other, CioPath): return self.wid == other.wid and self.path == other.path \ and self._is_dir == other.is_directory() return False # ------------------------------------------------------------------------- def __bool__(self) -> bool: """Override the default implementation.""" return self.wid is not None # ------------------------------------------------------------------------- def __hash__(self): """Override the default implementation.""" return hash(str(self)) # -------------------------------------------------------------------------
[docs] def uid(self) -> str: """Return a Unique ID for this `CioPath`. :rtype: str """ if self.wid is None: return '' ciohash = sha1(str(self).encode('utf8'), usedforsecurity=False) return f"cio{ciohash.hexdigest()}"
# -------------------------------------------------------------------------
[docs] def is_root(self) -> bool: """`True` if it is the root of the warehouse.""" return not self.path or self.path == '.'
# -------------------------------------------------------------------------
[docs] def is_directory(self) -> bool: """`True` if the path points to a directory.""" return self._is_dir
# -------------------------------------------------------------------------
[docs] def directory(self) -> str | None: """Return the path of the directory containing the file.""" if self.wid is None: return None return dirname(self.path) or '.'
# -------------------------------------------------------------------------
[docs] def facet(self, of_file: bool = False) -> str | None: """Return the facet repesentation of the directory containing the file or of the file itself.""" if self.wid is None: return None directory = self.path if of_file else dirname(self.path) return f'/{self.wid}/{directory}' \ if directory and directory != '.' else f'/{self.wid}'
# -------------------------------------------------------------------------
[docs] def file_name(self) -> str | None: """Return the file name.""" if self.wid is None: return None if self.is_root(): return '' return basename(self.path)
# -------------------------------------------------------------------------
[docs] def absolute_path(self, root: str | None) -> str | None: """Return the absolute path to the file.""" if root is None or self.wid is None: return None abs_path = normpath(join(root, self.path)) return abs_path if abs_path.startswith(root) else None
# -------------------------------------------------------------------------
[docs] def absolute_info(self, root: str | None) -> str | None: """Return the absolute path to the information file linked to the given `CioPath`.""" if root is None or self.wid is None or self.is_root(): return None info_dir = join(root, INFO_DIR) abs_path = normpath( join(info_dir, self.directory() or '.', f'{self.file_name()}.xml')) return abs_path if abs_path.startswith(info_dir) else None
# -------------------------------------------------------------------------
[docs] def absolute_lock(self, root: str | None) -> str | None: """Return the absolute path to the lock file linked to the given `CioPath`.""" if root is None or self.wid is None: return None locks_dir = join(root, LOCKS_DIR) abs_path = normpath(join(locks_dir, self.path, MYSELF)) \ if self.is_directory() else normpath(join(locks_dir, self.path)) return abs_path if abs_path.startswith(locks_dir) else None
# -------------------------------------------------------------------------
[docs] def route(self) -> str: """Return a piece of route representing the `CioPath`.""" if self.wid is None: return '' if self.is_root(): return f'{self.wid}/' return f"{self.wid}/{self.path}{'/' if self._is_dir else ''}"
# -------------------------------------------------------------------------
[docs] def mimetype(self) -> str | None: """Return the MIME type of the `CioPath`.""" if self.wid is None: return None if self.is_directory(): return 'inode/directory' return guess_type(self.path, False)[0]
# -------------------------------------------------------------------------
[docs] def contains(self, other: CioPath) -> bool: """`True` if the given `CioPath` is strictly contained in current one.""" return self._is_dir and other != self \ and str(other).startswith(str(self))
# -------------------------------------------------------------------------
[docs] def join(self, path: str, is_dir: bool = False) -> CioPath: """Creates a {CioPath} with path adjoined to self.""" if self.wid is None or not self._is_dir: return self return CioPath(self.wid, join(self.path, path), is_dir)
# -------------------------------------------------------------------------
[docs] def parent(self) -> CioPath: """Return the `CioPath`` without its final component, if there is one.""" if self.wid is None or self.is_root(): return self return CioPath(self.wid, dirname(self.path), True)
# -------------------------------------------------------------------------
[docs] def touch(self, root: str | None) -> CioPath: """Update the date of modification of the file.""" if self.wid is None or self.is_root(): return self abs_path = self.absolute_path(root) if abs_path is not None and exists(abs_path): Path(abs_path).touch() return self