Merge pull request #4 from Sage-Bionetworks-Workflows/bgrande/files-and-targets

Implement `FileType`, `File`, and `Target` to representing the targets of quality control

Author: Bruno Grande

src/dcqc/file.py

@@ -0,0 +1,194 @@
+from __future__ import annotations
+
+import os
+import re
+from collections.abc import Collection, Mapping
+from copy import deepcopy
+from dataclasses import asdict, dataclass
+from pathlib import Path, PurePosixPath
+from typing import Any, Optional
+
+from dcqc.mixins import SerializableMixin
+from dcqc.utils import open_parent_fs
+
+
+@dataclass
+class FileType:
+    # Class attributes
+    # A type hint is omitted so this attribute isn't
+    # picked up by @dataclass as an instance attribute
+    _registry = dict()  # type: ignore
+
+    # Instance attributes
+    name: str
+    file_extensions: tuple[str, ...]
+
+    def __init__(self, name: str, file_extensions: Collection[str]):
+        self.name = name
+        self.file_extensions = tuple(file_extensions)
+        self.register_file_type(self)
+
+    @classmethod
+    def register_file_type(cls, self):
+        name = self.name.lower()
+        if name in cls._registry:
+            message = f"File type ({name}) is already registered ({self._registry})."
+            raise ValueError(message)
+        cls._registry[name] = self
+
+    @classmethod
+    def get_file_type(cls, file_type: str) -> FileType:
+        file_type = file_type.lower()
+        if file_type not in cls._registry:
+            types = list(cls._registry)
+            message = f"File type ({file_type}) not among available options ({types})."
+            raise ValueError(message)
+        return cls._registry[file_type]
+
+
+# TODO: These file types could be moved to an external file
+# Instantiated file types are automatically tracked by the FileType class
+FileType("*", ())  # To represent all file types
+FileType("TXT", (".txt",))
+FileType("TIFF", (".tif", ".tiff"))
+FileType("OME-TIFF", (".ome.tif", ".ome.tiff"))
+
+
+@dataclass
+class File(SerializableMixin):
+    url: str
+    metadata: dict[str, Any]
+    type: str
+
+    LOCAL_REGEX = re.compile(r"((file|osfs)://)?/?[^:]+")
+
+    def __init__(
+        self,
+        url: str,
+        metadata: Mapping[str, Any],
+        relative_to: Optional[Path] = None,
+    ):
+        relative_to = relative_to or Path.cwd()
+        if self.is_local(url):
+            scheme, separator, resource = url.rpartition("://")
+            path = Path(resource)
+            if not path.is_absolute():
+                resource = os.path.relpath(relative_to / resource)
+            url = "".join([scheme, separator, resource])
+        self.url = str(url)
+        self.metadata = dict(metadata)
+        self.type = self._pop_file_type()
+        self.file_name = self._get_file_name()
+        self._fs = None
+
+    @property
+    def fs(self):
+        if self._fs is None:
+            fname = self.file_name
+            fs, bname = open_parent_fs(self.url)
+            if bname != fname:
+                message = f"Inconsistent file names: FS ({bname}) and File ({fname})."
+                raise ValueError(message)
+            self._fs = fs
+        return self._fs
+
+    def _pop_file_type(self) -> str:
+        file_type = self.get_metadata("file_type")
+        del self.metadata["file_type"]
+        return file_type
+
+    def _get_file_name(self):
+        path = PurePosixPath(self.url)
+        return path.name
+
+    def get_file_type(self) -> FileType:
+        return FileType.get_file_type(self.type)
+
+    def get_metadata(self, key: str) -> Any:
+        if key not in self.metadata:
+            url = self.url
+            md = self.metadata
+            message = f"File ({url}) does not have '{key}' in its metadata ({md})."
+            raise ValueError(message)
+        return self.metadata[key]
+
+    def is_local(self, url: Optional[str] = None):
+        url = url or self.url
+        return self.LOCAL_REGEX.fullmatch(url) is not None
+
+    # TODO: Create a new instance attribute `self._local_path` for keeping
+    #       track of the local path instead of overwriting `self.url`
+    def get_local_path(self) -> str:
+        if not self.is_local():
+            message = f"File ({self.url}) should first be downloaded using stage()."
+            raise FileNotFoundError(message)
+        local_path = self.url
+        if local_path.startswith(("osfs://", "file://")):
+            local_path = self.fs.getsyspath(self.file_name)
+        return local_path
+
+    def stage(self, destination: Optional[str] = None, overwrite: bool = False) -> str:
+        """Download remote files and copy local files.
+
+        A destination is required for remote files.
+        Local files aren't moved if a destination is omitted.
+
+        Args:
+            destination (Optional[str]): File or folder path
+                where to store the file. Defaults to None.
+            overwrite (bool): Whether to ignore existing file
+                at the target destination. Defaults to False.
+
+        Raises:
+            ValueError: If a destination is not specified
+                when staging a remote file.
+            ValueError: If the parent directory of the
+                destination does not exist.
+            FileExistsError: If the destination file already
+                exists and ``overwrite`` was not enabled.
+
+        Returns:
+            str: The updated URL (i.e., location) of the file.
+        """
+        if not destination:
+            if self.is_local():
+                return self.url
+            else:
+                message = f"Destination is required for remote files ({self.url})."
+                raise ValueError(message)
+
+        # By this point, destination is defined (not None)
+        file_name = self._get_file_name()
+        destination_path = Path(destination)
+        if destination_path.is_dir():
+            destination_path = destination_path / file_name
+            destination = destination_path.as_posix()
+
+        if not destination_path.parent.exists():
+            message = f"Parent folder of destination ({destination}) does not exist."
+            raise ValueError(message)
+
+        if destination_path.exists() and not overwrite:
+            message = f"Destination ({destination}) already exists. Enable overwrite."
+            raise FileExistsError(message)
+
+        if self.is_local():
+            local_path = self.get_local_path()
+            destination_path.symlink_to(local_path)
+        else:
+            with open(destination, "wb") as dest_file:
+                self.fs.download(self.file_name, dest_file)
+
+        self.url = destination
+        return self.url
+
+    def to_dict(self):
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, dictionary: dict) -> File:
+        dictionary = deepcopy(dictionary)
+        file_type = dictionary.pop("type")
+        dictionary["metadata"]["file_type"] = file_type
+        file = cls(**dictionary)
+        return file

src/dcqc/mixins.py

@@ -0,0 +1,24 @@
+from abc import ABC, abstractmethod
+from typing import Any
+
+SerializedObject = dict[str, Any]
+
+
+class SerializableMixin(ABC):
+    @classmethod
+    def from_dict_prepare(cls, dictionary: SerializedObject) -> SerializedObject:
+        """Validate and prepare dictionary for deserialization."""
+        type_ = dictionary.pop("type")
+        if type_ != cls.__name__:
+            message = f"Type ({type_}) does not match the class ({cls.__name__})."
+            raise ValueError(message)
+        return dictionary
+
+    @abstractmethod
+    def to_dict(self) -> SerializedObject:
+        """"""
+
+    # TODO: Uncomment this once the functions are ready
+    # @abstractmethod
+    # def from_dict(self):
+    #     """"""

src/dcqc/target.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from dataclasses import asdict, dataclass
+
+from dcqc.file import File
+from dcqc.mixins import SerializableMixin
+
+
+# TODO: Eventually, there might be target-specific metadata
+# TODO: Now that Target is much simpler, it might make sense
+#       to rename the class to FileSet since it currently
+#       really is just a wrapper for a group of files
+# TODO: Maybe the Composite pattern would work here?
+@dataclass
+class Target(SerializableMixin):
+    """Construct a multi-file Target.
+
+    Targets ensure support for both single-file
+    and multi-file tests.
+
+    Args:
+        *files (File): Sequence of files objects.
+    """
+
+    type: str
+    files: list[File]
+
+    def __init__(self, *files: File):
+        self.type = self.__class__.__name__
+        self.files = list(files)
+
+    def to_dict(self):
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, dictionary: dict) -> Target:
+        dictionary = deepcopy(dictionary)
+        dictionary = cls.from_dict_prepare(dictionary)
+        files = [File.from_dict(d) for d in dictionary["files"]]
+        target = cls(*files)
+        return target

src/dcqc/utils.py

@@ -0,0 +1,17 @@
+from fs import open_fs
+from fs.base import FS
+
+
+def open_parent_fs(url: str) -> tuple[FS, str]:
+    # Split off prefix to avoid issues with `rpartition("/")`
+    scheme, separator, path = url.rpartition("://")
+    if separator == "":
+        prefix = "osfs://"
+    else:
+        prefix = scheme + separator
+
+    # parent_path can be "" if there is no "/" in the path
+    parent_path, _, base_name = path.rpartition("/")
+    parent_url = prefix + parent_path
+    fs = open_fs(parent_url)
+    return fs, base_name

tests/conftest.py

@@ -9,11 +9,13 @@
 
 from datetime import datetime
 from getpass import getuser
-from pathlib import Path, PurePath, PurePosixPath
+from pathlib import Path
 from uuid import uuid4
 
 import pytest
 
+from dcqc.file import File
+
 CNFPATH = Path(__file__).resolve()
 TESTDIR = CNFPATH.parent
 DATADIR = TESTDIR / "data"
@@ -30,12 +32,36 @@ def pytest_configure():
 
 @pytest.fixture
 def get_data():
-    def _get_data(filename: str, as_posix: bool = False) -> PurePath:
+    def _get_data(filename: str) -> Path:
         path = DATADIR / filename
         if not path.exists():
             raise ValueError(f"Path ({path}) does not exist.")
-        if as_posix:
-            path = PurePosixPath(*path.parts)  # type: ignore
         return path
 
     yield _get_data
+
+
+@pytest.fixture
+def test_files(get_data):
+    txt_path = get_data("test.txt").as_posix()
+    tiff_path = get_data("circuit.tif").as_posix()
+    syn_path = "syn://syn50555279"
+    good_metadata = {
+        "file_type": "txt",
+        "md5_checksum": "14758f1afd44c09b7992073ccf00b43d",
+    }
+    bad_metadata = {
+        "file_type": "tiff",
+        "md5_checksum": "definitelynottherightmd5checksum",
+    }
+    tiff_metadata = {
+        "file_type": "tiff",
+        "md5_checksum": "c7b08f6decb5e7572efbe6074926a843",
+    }
+    test_files = {
+        "good": File(txt_path, good_metadata),
+        "bad": File(txt_path, bad_metadata),
+        "tiff": File(tiff_path, tiff_metadata),
+        "synapse": File(syn_path, good_metadata),
+    }
+    yield test_files