fix: clear the type checker and linter findings
flake8 reported 49 findings and pyright 10, and two of them were real bugs rather than matters of style. The rest were unused imports and four functions over the complexity limit. - Unlock a DOI folder with the id the DOI response carries, rather than an attribute the model does not have, which raised `AttributeError` on every locked DOI folder. - Size the `ls` Size column from the sub-folder on the row rather than from its parent, which pushed the later columns out of line. - Share the path resolution and the destination checks between `cp` and `mv`, and split the recursive download and the `ls` row printing, bringing all four functions under the complexity limit. - Accept a client built without a connection, which `config` and `version` rely on, and report the reason if one is then asked for. - Declare the config protocol's constructor for the type checker alone, so the protocol keeps its guard against being instantiated. - Remove 41 unused imports, and let flake8 accept black's spacing. - Point pyright at the project's own environment, without which it resolved no dependency and reported 124 findings that were not real. - Cover the two fixes and the shared `cp`/`mv` paths with new tests.
This commit is contained in:
+94
-80
@@ -1,21 +1,40 @@
|
||||
import os
|
||||
from typing import Any
|
||||
from dataclasses import dataclass
|
||||
from unicodedata import normalize
|
||||
|
||||
from mdrsclient.api import DoiApi, FilesApi, FoldersApi, LaboratoriesApi, UsersApi
|
||||
from mdrsclient.api import FilesApi, FoldersApi
|
||||
from mdrsclient.cache import CacheInterface
|
||||
from mdrsclient.config import ConfigInterface
|
||||
from mdrsclient.connection import MDRSConnection
|
||||
from mdrsclient.exceptions import IllegalArgumentException, MDRSException, UnauthorizedException, UnexpectedException
|
||||
from mdrsclient.models import File, Folder, Laboratory, Token, User
|
||||
from mdrsclient.exceptions import IllegalArgumentException
|
||||
from mdrsclient.models import File, Folder, FolderSimple, Laboratory
|
||||
from mdrsclient.models.file import find_file
|
||||
from mdrsclient.services import MdrsService
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TransferEndpoints:
|
||||
"""Both ends of a `cp` or `mv`, resolved: the parents, their contents, and the names."""
|
||||
|
||||
laboratory: Laboratory
|
||||
s_path: str
|
||||
s_basename: str
|
||||
s_parent_folder: Folder
|
||||
s_parent_files: list[File]
|
||||
d_path: str
|
||||
d_basename: str
|
||||
d_parent_folder: Folder
|
||||
d_parent_files: list[File]
|
||||
|
||||
def is_same_place(self) -> bool:
|
||||
"""Whether the destination names what the source already is, leaving nothing to do."""
|
||||
return self.s_parent_folder.id == self.d_parent_folder.id and self.s_basename == self.d_basename
|
||||
|
||||
|
||||
class MdrsClient(MdrsService):
|
||||
"""Service layer client for MDRS."""
|
||||
|
||||
def __init__(self, connection: MDRSConnection, config_class: type[ConfigInterface] | None = None):
|
||||
def __init__(self, connection: MDRSConnection | None, config_class: type[ConfigInterface] | None = None):
|
||||
super().__init__(connection, config_class)
|
||||
|
||||
@classmethod
|
||||
@@ -63,7 +82,8 @@ class MdrsClient(MdrsService):
|
||||
files = self.find_files(folder.id)
|
||||
return folder, files
|
||||
|
||||
def cp(self, src_path: str, dest_path: str, is_recursive: bool = False) -> None:
|
||||
def __resolve_transfer(self, src_path: str, dest_path: str) -> TransferEndpoints:
|
||||
"""Resolve both ends of a transfer, refusing one that crosses a remote or a laboratory."""
|
||||
s_remote, s_laboratory_name, s_path = self.parse_remote_host_with_path(src_path)
|
||||
d_remote, d_laboratory_name, d_path = self.parse_remote_host_with_path(dest_path)
|
||||
if s_remote != d_remote:
|
||||
@@ -71,8 +91,9 @@ class MdrsClient(MdrsService):
|
||||
if s_laboratory_name != d_laboratory_name:
|
||||
raise IllegalArgumentException("Laboratory mismatched.")
|
||||
s_path = s_path.rstrip("/")
|
||||
s_dirname = os.path.dirname(s_path)
|
||||
s_basename = os.path.basename(s_path)
|
||||
# A destination ending in a separator names a folder to put the source into, and
|
||||
# keeps the source's own name.
|
||||
if d_path.endswith("/"):
|
||||
d_dirname = d_path
|
||||
d_basename = s_basename
|
||||
@@ -80,84 +101,77 @@ class MdrsClient(MdrsService):
|
||||
d_dirname = os.path.dirname(d_path)
|
||||
d_basename = os.path.basename(d_path)
|
||||
laboratory = self.find_laboratory(s_laboratory_name)
|
||||
s_parent_folder = self.find_folder(laboratory, s_dirname)
|
||||
s_parent_files = self.find_files(s_parent_folder.id)
|
||||
s_parent_folder = self.find_folder(laboratory, os.path.dirname(s_path))
|
||||
d_parent_folder = self.find_folder(laboratory, d_dirname)
|
||||
d_parent_files = self.find_files(d_parent_folder.id)
|
||||
s_file = find_file(s_parent_files, s_basename)
|
||||
return TransferEndpoints(
|
||||
laboratory=laboratory,
|
||||
s_path=s_path,
|
||||
s_basename=s_basename,
|
||||
s_parent_folder=s_parent_folder,
|
||||
s_parent_files=self.find_files(s_parent_folder.id),
|
||||
d_path=d_path,
|
||||
d_basename=d_basename,
|
||||
d_parent_folder=d_parent_folder,
|
||||
d_parent_files=self.find_files(d_parent_folder.id),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def __check_file_destination(ends: TransferEndpoints) -> None:
|
||||
"""Refuse a destination that a file cannot take the place of."""
|
||||
if find_file(ends.d_parent_files, ends.d_basename) is not None:
|
||||
raise IllegalArgumentException(f"File `{ends.d_basename}` already exists.")
|
||||
if ends.d_parent_folder.find_sub_folder(ends.d_basename) is not None:
|
||||
raise IllegalArgumentException(
|
||||
f"Cannot overwrite non-folder `{ends.d_basename}` with folder `{ends.d_path}`."
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def __check_folder_destination(ends: TransferEndpoints, s_folder: FolderSimple) -> None:
|
||||
"""Refuse a destination that a folder cannot take the place of."""
|
||||
if find_file(ends.d_parent_files, ends.d_basename) is not None:
|
||||
raise IllegalArgumentException(
|
||||
f"Cannot overwrite non-folder `{ends.d_basename}` with folder `{ends.s_path}`."
|
||||
)
|
||||
d_folder = ends.d_parent_folder.find_sub_folder(ends.d_basename)
|
||||
if d_folder is not None:
|
||||
if d_folder.id == s_folder.id:
|
||||
raise IllegalArgumentException(f"`{ends.s_path}` and `{ends.s_path}` are the same folder.")
|
||||
raise IllegalArgumentException(f"Cannot move `{ends.s_path}` to `{ends.d_path}`: Folder not empty.")
|
||||
|
||||
@staticmethod
|
||||
def __find_source_folder(ends: TransferEndpoints) -> FolderSimple:
|
||||
s_folder = ends.s_parent_folder.find_sub_folder(ends.s_basename)
|
||||
if s_folder is None:
|
||||
raise IllegalArgumentException(f"File or folder `{ends.s_basename}` not found.")
|
||||
return s_folder
|
||||
|
||||
def cp(self, src_path: str, dest_path: str, is_recursive: bool = False) -> None:
|
||||
ends = self.__resolve_transfer(src_path, dest_path)
|
||||
s_file = find_file(ends.s_parent_files, ends.s_basename)
|
||||
if s_file is not None:
|
||||
d_file = find_file(d_parent_files, d_basename)
|
||||
if d_file is not None:
|
||||
raise IllegalArgumentException(f"File `{d_basename}` already exists.")
|
||||
d_sub_folder = d_parent_folder.find_sub_folder(d_basename)
|
||||
if d_sub_folder is not None:
|
||||
raise IllegalArgumentException(f"Cannot overwrite non-folder `{d_basename}` with folder `{d_path}`.")
|
||||
file_api = FilesApi(self.connection)
|
||||
if s_parent_folder.id != d_parent_folder.id or d_basename != s_basename:
|
||||
file_api.copy(s_file, d_parent_folder.id, normalize("NFC", d_basename))
|
||||
else:
|
||||
s_folder = s_parent_folder.find_sub_folder(s_basename)
|
||||
if s_folder is None:
|
||||
raise IllegalArgumentException(f"File or folder `{s_basename}` not found.")
|
||||
if not is_recursive:
|
||||
raise IllegalArgumentException(f"Cannot copy `{s_path}`: Is a folder.")
|
||||
if find_file(d_parent_files, d_basename) is not None:
|
||||
raise IllegalArgumentException(f"Cannot overwrite non-folder `{d_basename}` with folder `{s_path}`.")
|
||||
d_folder = d_parent_folder.find_sub_folder(d_basename)
|
||||
if d_folder is not None:
|
||||
if d_folder.id == s_folder.id:
|
||||
raise IllegalArgumentException(f"`{s_path}` and `{s_path}` are the same folder.")
|
||||
raise IllegalArgumentException(f"Cannot move `{s_path}` to `{d_path}`: Folder not empty.")
|
||||
folder_api = FoldersApi(self.connection)
|
||||
if s_parent_folder.id != d_parent_folder.id or s_basename != d_basename:
|
||||
folder_api.copy(s_folder, d_parent_folder.id, normalize("NFC", d_basename))
|
||||
self.__check_file_destination(ends)
|
||||
if not ends.is_same_place():
|
||||
FilesApi(self.connection).copy(s_file, ends.d_parent_folder.id, normalize("NFC", ends.d_basename))
|
||||
return
|
||||
s_folder = self.__find_source_folder(ends)
|
||||
if not is_recursive:
|
||||
raise IllegalArgumentException(f"Cannot copy `{ends.s_path}`: Is a folder.")
|
||||
self.__check_folder_destination(ends, s_folder)
|
||||
if not ends.is_same_place():
|
||||
FoldersApi(self.connection).copy(s_folder, ends.d_parent_folder.id, normalize("NFC", ends.d_basename))
|
||||
|
||||
def mv(self, src_path: str, dest_path: str) -> None:
|
||||
s_remote, s_laboratory_name, s_path = self.parse_remote_host_with_path(src_path)
|
||||
d_remote, d_laboratory_name, d_path = self.parse_remote_host_with_path(dest_path)
|
||||
if s_remote != d_remote:
|
||||
raise IllegalArgumentException("Remote host mismatched.")
|
||||
if s_laboratory_name != d_laboratory_name:
|
||||
raise IllegalArgumentException("Laboratory mismatched.")
|
||||
s_path = s_path.rstrip("/")
|
||||
s_dirname = os.path.dirname(s_path)
|
||||
s_basename = os.path.basename(s_path)
|
||||
if d_path.endswith("/"):
|
||||
d_dirname = d_path
|
||||
d_basename = s_basename
|
||||
else:
|
||||
d_dirname = os.path.dirname(d_path)
|
||||
d_basename = os.path.basename(d_path)
|
||||
laboratory = self.find_laboratory(s_laboratory_name)
|
||||
s_parent_folder = self.find_folder(laboratory, s_dirname)
|
||||
s_parent_files = self.find_files(s_parent_folder.id)
|
||||
d_parent_folder = self.find_folder(laboratory, d_dirname)
|
||||
d_parent_files = self.find_files(d_parent_folder.id)
|
||||
s_file = find_file(s_parent_files, s_basename)
|
||||
ends = self.__resolve_transfer(src_path, dest_path)
|
||||
s_file = find_file(ends.s_parent_files, ends.s_basename)
|
||||
if s_file is not None:
|
||||
d_file = find_file(d_parent_files, d_basename)
|
||||
if d_file is not None:
|
||||
raise IllegalArgumentException(f"File `{d_basename}` already exists.")
|
||||
d_sub_folder = d_parent_folder.find_sub_folder(d_basename)
|
||||
if d_sub_folder is not None:
|
||||
raise IllegalArgumentException(f"Cannot overwrite non-folder `{d_basename}` with folder `{d_path}`.")
|
||||
file_api = FilesApi(self.connection)
|
||||
if s_parent_folder.id != d_parent_folder.id or d_basename != s_basename:
|
||||
file_api.move(s_file, d_parent_folder.id, normalize("NFC", d_basename))
|
||||
else:
|
||||
s_folder = s_parent_folder.find_sub_folder(s_basename)
|
||||
if s_folder is None:
|
||||
raise IllegalArgumentException(f"File or folder `{s_basename}` not found.")
|
||||
if find_file(d_parent_files, d_basename) is not None:
|
||||
raise IllegalArgumentException(f"Cannot overwrite non-folder `{d_basename}` with folder `{s_path}`.")
|
||||
d_folder = d_parent_folder.find_sub_folder(d_basename)
|
||||
if d_folder is not None:
|
||||
if d_folder.id == s_folder.id:
|
||||
raise IllegalArgumentException(f"`{s_path}` and `{s_path}` are the same folder.")
|
||||
raise IllegalArgumentException(f"Cannot move `{s_path}` to `{d_path}`: Folder not empty.")
|
||||
folder_api = FoldersApi(self.connection)
|
||||
if s_parent_folder.id != d_parent_folder.id or d_basename != s_basename:
|
||||
folder_api.move(s_folder, d_parent_folder.id, normalize("NFC", d_basename))
|
||||
self.__check_file_destination(ends)
|
||||
if not ends.is_same_place():
|
||||
FilesApi(self.connection).move(s_file, ends.d_parent_folder.id, normalize("NFC", ends.d_basename))
|
||||
return
|
||||
s_folder = self.__find_source_folder(ends)
|
||||
self.__check_folder_destination(ends, s_folder)
|
||||
if not ends.is_same_place():
|
||||
FoldersApi(self.connection).move(s_folder, ends.d_parent_folder.id, normalize("NFC", ends.d_basename))
|
||||
|
||||
def chacl(
|
||||
self, remote_path: str, access_level: int, is_recursive: bool = False, password: str | None = None
|
||||
|
||||
Reference in New Issue
Block a user