Separate handling of File URL and local path

Author: Bruno Grande

setup.cfg

@@ -149,7 +149,6 @@ extensions =
     pre_commit
 
 [autoflake]
-check=true
 in-place=true
 remove-all-unused-imports=true
 remove-unused-variables=true

src/dcqc/file.py

@@ -5,7 +5,8 @@
 from collections.abc import Collection, Mapping
 from copy import deepcopy
 from dataclasses import asdict, dataclass
-from pathlib import Path, PurePosixPath
+from pathlib import Path
+from tempfile import mkdtemp
 from typing import Any, Optional
 
 from fs.base import FS
@@ -58,9 +59,20 @@ def get_file_type(cls, file_type: str) -> FileType:
 
 @dataclass
 class File(SerializableMixin):
+    """Construct a File object.
+
+    Args:
+        url (str): URL indicating the location of the file.
+        metadata (Mapping[str, Any]): File metadata.
+        relative_to (Optional[Path]): Used to update any
+            local URLs if they are relative to a directory
+            other than the current work directory (default).
+    """
+
     url: str
     metadata: dict[str, Any]
     type: str
+    local_path: Optional[str]
 
     LOCAL_REGEX = re.compile(r"((file|osfs)://)?/?[^:]+")
 
@@ -69,40 +81,72 @@ def __init__(
         url: str,
         metadata: Mapping[str, Any],
         relative_to: Optional[Path] = None,
+        local_path: Optional[str] = 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.url = self._relativize_url(url, relative_to)
         self.metadata = dict(metadata)
         self.type = self._pop_file_type()
-        self.file_name = self._get_file_name()
+
         self._fs: Optional[FS]
         self._fs = None
+        self._fs_path: Optional[str]
+        self._fs_path = None
+        self._name: Optional[str]
+        self._name = None
 
-    @property
-    def fs(self) -> FS:
-        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
+        self.local_path = local_path or self._init_local_path()
+
+    def _relativize_url(self, url: str, relative_to: Optional[Path]) -> str:
+        """Update local URLs if relative to a directory other than CWD."""
+        relative_to = relative_to or Path.cwd()
+        if self.is_url_local(url):
+            scheme, separator, resource = url.rpartition("://")
+            path = Path(resource)
+            if not path.is_absolute():
+                resource = os.path.relpath(relative_to / resource)
+            url = f"{scheme}{separator}{resource}"
+        return url
 
     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) -> str:
-        path = PurePosixPath(self.url)
-        return path.name
+    def _init_local_path(self) -> Optional[str]:
+        if self.is_url_local():
+            local_path = self.fs.getsyspath(self.fs_path)
+        else:
+            local_path = None
+        return local_path
+
+    def _initialize_fs(self) -> tuple[FS, str]:
+        """Retrieve and store parent FS and basename."""
+        fs, fs_path = open_parent_fs(self.url)
+        self._fs_path = fs_path
+        self._fs = fs
+        return fs, fs_path
+
+    @property
+    def fs(self) -> FS:
+        fs = self._fs
+        if fs is None:
+            fs, _ = self._initialize_fs()
+        return fs
+
+    @property
+    def fs_path(self) -> str:
+        fs_path = self._fs_path
+        if fs_path is None:
+            _, fs_path = self._initialize_fs()
+        return fs_path
+
+    @property
+    def name(self) -> str:
+        name = self._name
+        if name is None:
+            info = self.fs.getinfo(self.fs_path)
+            name = info.name
+        return name
 
     def get_file_type(self) -> FileType:
         return FileType.get_file_type(self.type)
@@ -115,22 +159,20 @@ def get_metadata(self, key: str) -> Any:
             raise ValueError(message)
         return self.metadata[key]
 
-    def is_local(self, url: Optional[str] = None) -> bool:
+    def is_url_local(self, url: Optional[str] = None) -> bool:
         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 is_file_local(self, url: Optional[str] = None) -> bool:
+        return self.local_path is not None
+
+    def get_local_path(self) -> Path:
+        if self.local_path is None:
+            message = "Local path is unavailable. Use stage() to create a local copy."
+            raise ValueError(message)
+        return Path(self.local_path)
 
-    def stage(self, destination: Optional[str] = None, overwrite: bool = False) -> str:
+    def stage(self, destination: Optional[str] = None, overwrite: bool = False) -> Path:
         """Download remote files and copy local files.
 
         A destination is required for remote files.
@@ -143,28 +185,24 @@ def stage(self, destination: Optional[str] = None, overwrite: bool = False) -> s
                 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.
+            Path: The path of the local copy.
         """
         if not destination:
-            if self.is_local():
-                return self.url
+            if self.local_path is not None:
+                return self.get_local_path()
             else:
-                message = f"Destination is required for remote files ({self.url})."
-                raise ValueError(message)
+                destination = mkdtemp()
 
         # 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_path = destination_path / self.name
             destination = destination_path.as_posix()
 
         if not destination_path.parent.exists():
@@ -175,15 +213,14 @@ def stage(self, destination: Optional[str] = None, overwrite: bool = False) -> s
             message = f"Destination ({destination}) already exists. Enable overwrite."
             raise FileExistsError(message)
 
-        if self.is_local():
+        if self.is_url_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)
+            with destination_path.open("wb") as dest_file:
+                self.fs.download(self.fs_path, dest_file)
 
-        self.url = destination
-        return self.url
+        return destination_path
 
     def to_dict(self) -> SerializedObject:
         return asdict(self)

src/dcqc/parsers.py

@@ -13,6 +13,7 @@ class CsvParser:
 
     def __init__(self, path: Path):
         self.path = path
+        self.csv_directory = self.path.parent
 
     def list_rows(self) -> Iterator[tuple[int, dict]]:
         with self.path.open(newline="") as file:
@@ -21,14 +22,17 @@ def list_rows(self) -> Iterator[tuple[int, dict]]:
                 yield index, row
 
     def _row_to_file(self, row: dict[str, str]) -> File:
-        csv_directory = self.path.parent
         url = row.pop("url")
-        file = File(url, row, relative_to=csv_directory)
+        file = File(url, row, relative_to=self.csv_directory)
         return file
 
     def create_files(self) -> Iterator[File]:
-        for _, row in self.list_rows():
+        for index, row in self.list_rows():
             file = self._row_to_file(row)
+            if not file.is_file_local():
+                destination = self.csv_directory / "staged_files" / f"index_{index}"
+                destination.mkdir(parents=True, exist_ok=True)
+                file.stage(destination.as_posix(), overwrite=True)
             yield file
 
     def create_targets(self) -> Iterator[Target]:

src/dcqc/tests/tests.py

@@ -37,7 +37,7 @@ def compute_status(self) -> TestStatus:
     def _compute_md5_checksum(self, file: File) -> str:
         local_path = file.get_local_path()
         hash_md5 = hashlib.md5()
-        with open(local_path, "rb") as f:
+        with local_path.open("rb") as f:
             for chunk in iter(lambda: f.read(4096), b""):
                 hash_md5.update(chunk)
         actual_md5 = hash_md5.hexdigest()
@@ -49,7 +49,7 @@ class LibTiffInfoTest(ExternalTestMixin, TestABC):
 
     def generate_process(self) -> Process:
         file = self._get_single_target_file()
-        path = file.get_local_path()
+        path = file.get_local_path().as_posix()
         command_args = ["tiffinfo", path]
         process = Process(
             container="autamus/libtiff:4.4.0",
@@ -63,7 +63,7 @@ class BioFormatsInfoTest(ExternalTestMixin, TestABC):
 
     def generate_process(self) -> Process:
         file = self._get_single_target_file()
-        path = file.get_local_path()
+        path = file.get_local_path().as_posix()
         command_args = [
             'export PATH="$PATH:/opt/bftools"',
             ";",
@@ -85,7 +85,7 @@ class OmeXmlSchemaTest(ExternalTestMixin, TestABC):
 
     def generate_process(self) -> Process:
         file = self._get_single_target_file()
-        path = file.get_local_path()
+        path = file.get_local_path().as_posix()
         command_args = [
             'export PATH="$PATH:/opt/bftools"',
             ";",