"""Class for warehouses without Version Control System."""
from __future__ import annotations
from os import makedirs, remove, scandir, rename
from os.path import exists, isdir, dirname, normpath, splitext
from shutil import rmtree, copy
from chrysalio.lib.utils import copy_content
from ..models.dbwarehouse import DBWarehouse
from .utils import file_move_check
from .ciopath import CioPath
from .i18n import _
# =============================================================================
[docs]
class VcsNone():
"""Class to manage warehouses without VCS.
:param str uid:
ID of this VCS i.e. ID of the corresponding warehouse.
:param str root:
Absolute path to the warehouse directory.
:type dbwarehouse: .models.dbwarehouse.DBWarehouse
:param dbwarehouse: (optional)
DBWarehouse defining the warehouse managed by this VCS.
"""
engine = 'none'
# -------------------------------------------------------------------------
def __init__(
self, uid: str, root: str, dbwarehouse: DBWarehouse | None = None):
"""Constructor method."""
self.uid = uid
self.root = root
self._source = str(dbwarehouse.vcs_url) \
if dbwarehouse is not None and dbwarehouse.vcs_url else None
# -------------------------------------------------------------------------
[docs]
def init(self):
"""Possibly create the root directory.
:rtype: None
"""
makedirs(self.root, exist_ok=True)
# -------------------------------------------------------------------------
[docs]
@classmethod
def is_dirty(cls) -> bool:
"""Return ``False``.
:rtype: bool
"""
return False
# -------------------------------------------------------------------------
[docs]
def clone(self):
"""Call meth:`init` and possibly fill the warehouse.
:rtype: None
"""
self.init()
if self._source and exists(self._source) \
and not tuple(scandir(self.root)):
copy_content(self._source, self.root)
# -------------------------------------------------------------------------
[docs]
def pull(self):
"""Do a clone if source exists and root directory is empty.
:rtype: None
"""
if self._source and exists(self._source) \
and len(tuple(scandir(self.root))) < 2:
copy_content(self._source, self.root)
# -------------------------------------------------------------------------
[docs]
def add(self, ciopath: CioPath | None = None):
"""Do nothing.
:rtype: None
"""
# -------------------------------------------------------------------------
[docs]
def remove(self, ciopath: CioPath) -> str | None:
"""Remove a file or a directory.
:type ciopath: .lib.ciopath.CioPath
:param ciopath:
`CioPath` of the file to remove.
:rtype: :class:`pyramid.i18n.TranslationString` or ``None``
"""
abs_path = ciopath.absolute_path(self.root)
if ciopath.wid != self.uid or not abs_path:
return _('remove: incorrect path')
if not exists(abs_path):
return None
# Remove information file
info_file = ciopath.absolute_info(self.root)
if info_file and exists(info_file):
remove(info_file)
if info_file and ciopath.is_directory():
info_file = splitext(info_file)[0]
if isdir(info_file):
rmtree(info_file)
if info_file:
info_file = normpath(dirname(info_file))
while info_file != self.root and exists(info_file) and not tuple(
scandir(info_file)):
rmtree(info_file)
info_file = dirname(info_file)
# Remove the file
if isdir(abs_path):
rmtree(abs_path)
elif exists(abs_path):
remove(abs_path)
return None
# -------------------------------------------------------------------------
[docs]
def move(
self,
ciopath1: CioPath,
ciopath2: CioPath,
overwrite_ok: bool = False,
mode: str = 'move') -> tuple[CioPath | None, str | None]:
"""Move or copy a file inside the warehouse.
:type ciopath1: .lib.ciopath.CioPath
:param ciopaht1:
`CioPath` of the first file.
:type ciopath2: .lib.ciopath.CioPath
:param ciopath2:
`CioPath` of the srecond file.
:param bool overwrite_ok: (default=False)
If ``True``, do nothing if files are the same.
:param str mode: (``'move'``, ``'copy'``, ``'rename'``)
The way the move must operate.
:rtype: tuple
:return:
A tuple such as ``(final_ciopath2, error)``.
"""
return self.move_(ciopath1, ciopath2, overwrite_ok, mode)
# -------------------------------------------------------------------------
[docs]
def move_(
self, ciopath1: CioPath, ciopath2: CioPath, overwrite_ok: bool,
mode: str) -> tuple[CioPath | None, str | None]:
"""Method used by ``VcsNone`` and ``VcsGit`` to move files."""
# Prepare files
if ciopath1.wid != self.uid or ciopath2.wid != self.uid:
return None, _('File outside the warehouse')
final_ciopath, err = file_move_check(
self.root, ciopath1, self.root, ciopath2, overwrite_ok, mode)
if final_ciopath is None:
return final_ciopath, err
# Action
action = copy_content if mode == 'copy' and ciopath1.is_directory() \
else copy if mode == 'copy' \
else rename
# Move/copy file
abs_path = final_ciopath.absolute_path(self.root)
if abs_path:
makedirs(dirname(abs_path), exist_ok=True)
action(ciopath1.absolute_path(self.root) or '.', abs_path)
# Move/copy information file
info_file1 = ciopath1.absolute_info(self.root)
info_file2 = final_ciopath.absolute_info(self.root)
if info_file1 and info_file2 and exists(info_file1):
makedirs(dirname(info_file2), exist_ok=True)
action = copy if mode == 'copy' else rename
action(info_file1, info_file2)
if info_file1 and ciopath1.is_directory():
info_file1 = splitext(info_file1)[0]
if isdir(info_file1) and info_file2:
info_file2 = splitext(info_file2)[0]
action = copy_content if mode == 'copy' else rename
action(info_file1, info_file2)
return final_ciopath, None
# -------------------------------------------------------------------------
[docs]
def copy(
self,
ciopath1: CioPath,
ciopath2: CioPath,
overwrite_ok: bool = False) -> tuple[CioPath | None, str | None]:
"""Copy a file.
:type ciopath1: .lib.ciopath.CioPath
:param ciopaht1:
`CioPath` of the first file.
:type ciopath2: .lib.ciopath.CioPath
:param ciopath2:
`CioPath` of the second file.
:param bool overwrite_ok: (default=False)
If ``True``, do nothing if files are the same.
:rtype: :class:`pyramid.i18n.TranslationString` or ``None``
"""
return self.move(
ciopath1, ciopath2, overwrite_ok=overwrite_ok, mode='copy')
# -------------------------------------------------------------------------
[docs]
def commit(self, message: str, name: str, email: str = ''):
"""Do nothing.
:rtype: None
"""
# -------------------------------------------------------------------------
[docs]
def log(self, ciopath: CioPath, limit: int = 20):
"""Do nothing.
:rtype: None
"""