Merge pull request #108 from ArcanaFramework/forward-ref-in-mime

Forward-ref-in-mime and additional logging

Author: tclose

fileformats/core/fileset.py

@@ -1452,15 +1452,22 @@ def copy(
         dest_dir = Path(dest_dir)
         # Logic to determine the laziest mode to use
         mode = self.CopyMode[mode] if isinstance(mode, str) else mode
+        logger.debug("Requested '%s' for the copy operation", mode)
         if len(self.fspaths) == 1:
             # If there is only one path to copy, then collation isn't meaningful
             collation = self.CopyCollation.any
+            logger.debug(
+                "Collation mode is set to 'any' as there is only one path in the fileset"
+            )
         else:
             collation = (
                 self.CopyCollation[collation]
                 if isinstance(collation, str)
                 else collation
             )
+            logger.debug(
+                "Collation mode is set to '%s' for the copy operation", collation
+            )
         # Rule out any copy modes that are not supported given the collation mode
         # and file-system mounts the paths and destination directory reside on
         constraints = []
@@ -1473,7 +1480,9 @@ def copy(
                 f"Destination directory is on CIFS mount ({dest_dir}) "
                 "and we therefore cannot create a symlink"
             )
-            logger.debug(constraint)
+            logger.debug(
+                constraint + ", supported modes restricted to %s", supported_modes
+            )
             constraints.append(constraint)
         not_on_same_mount = [
             p for p in self.fspaths if not FsMountIdentifier.on_same_mount(p, dest_dir)
@@ -1485,7 +1494,9 @@ def copy(
                 f"not on same file-system mount as the destination directory {dest_dir}"
                 "and therefore cannot be hard-linked"
             )
-            logger.debug(constraint)
+            logger.debug(
+                constraint + ", supported modes restricted to %s", supported_modes
+            )
             constraints.append(constraint)
         if (
             new_stem
@@ -1497,9 +1508,19 @@ def copy(
             )
         ):
             supported_modes -= self.CopyMode.leave
+            logger.debug(
+                "Collation mode is set to '%s' or new_stem/prefix/stem_suffix is set, "
+                "therefore we cannot leave the files where they are",
+                collation,
+            )
 
         # Get the intersection of copy modes that are supported and have been requested
         selected_mode = mode & supported_modes
+        logger.debug(
+            "Selected copy mode is '%s' for the copy operation (requested %s)",
+            selected_mode,
+            mode,
+        )
         if not selected_mode:
             msg = (
                 f"Cannot copy {self} using '{mode}' mode as it is not supported by "
@@ -1509,15 +1530,20 @@ def copy(
                 msg += ", and the following constraints:\n" + "\n".join(constraints)
             raise UnsatisfiableCopyModeError(msg)
         if selected_mode & self.CopyMode.leave:
+            logger.debug(
+                "Selected copy mode is set to 'leave', therefore we are not copying anything"
+            )
             return self  # Don't need to do anything
 
         copy_file: ty.Callable[[Path, Path], None]
         copy_dir: ty.Callable[[Path, Path], None]
 
         # Select inner copy/link methods
         if selected_mode & self.CopyMode.symlink:
+            logger.debug('Using symbolic links for the "copy" operation')
             copy_dir = copy_file = os.symlink
         elif selected_mode & self.CopyMode.hardlink:
+            logger.debug('Using hard links for the "copy" operation')
             copy_file = os.link
 
             def hardlink_dir(src: Path, dest: Path) -> None:
@@ -1530,6 +1556,7 @@ def hardlink_dir(src: Path, dest: Path) -> None:
 
             copy_dir = hardlink_dir
         else:
+            logger.debug('Using full copy for the "copy" operation')
             assert selected_mode & self.CopyMode.copy
             copy_dir = shutil.copytree
             copy_file = shutil.copyfile  # type: ignore
@@ -1737,10 +1764,15 @@ def _src_dest_pairs(
             exts = [d[-1] for d in decomposed_fspaths]
             duplicate_exts = [n for n, c in Counter(exts).items() if c > 1]
             if duplicate_exts:
+                reason = ""
+                if new_stem:
+                    reason += f"with a new stem, {new_stem!r}, "
+                if collation == self.CopyCollation.adjacent:
+                    reason += f"with collation mode '{collation}', "
                 raise UnsatisfiableCopyModeError(
-                    f"Cannot copy/move {self} with collation mode "
-                    f'"{collation}", as there are duplicate extensions, {duplicate_exts}, '
-                    f"in file paths: " + "\n".join(str(p) for p in self.fspaths)
+                    f"Cannot copy/move {self} {reason} as there are duplicate "
+                    f"extensions, {duplicate_exts}, in file paths: "
+                    + "\n".join(str(p) for p in self.fspaths)
                 )
             # Set default for new_stem if not provided and collating file-set to be adjacent
             if new_stem is None:

fileformats/core/identification.py

@@ -1,5 +1,6 @@
 import operator
 from pathlib import Path
+import inspect
 import typing as ty
 import re
 from fileformats.core.exceptions import (
@@ -128,8 +129,15 @@ def to_mime(
         namespace scheme instead of putting all non-standard types into the 'application'
         registry if not
     """
+    # Handle forward references
+    if isinstance(datatype, ty.ForwardRef):
+        datatype = datatype.__forward_arg__
     origin = ty.get_origin(datatype)
-    if origin is None and not issubclass(datatype, fileformats.core.DataType):
+    if (
+        origin is None
+        and inspect.isclass(datatype)
+        and not issubclass(datatype, fileformats.core.DataType)
+    ):
         raise TypeError(
             f"Cannot convert {datatype} to mime-type as it is not a file-set class"
         )
@@ -146,9 +154,6 @@ def to_mime(
         return item_mime
     if origin is ty.Union:
         return ",".join(to_mime(t, official=official) for t in ty.get_args(datatype))
-    # Handle case
-    if isinstance(datatype, ty.ForwardRef):
-        datatype = datatype.__forward_arg__
     if (
         isinstance(datatype, str)
         and datatype.startswith("fileformats.")

fileformats/core/tests/test_utils.py

@@ -101,11 +101,11 @@ def test_copy_symlink_with_unrequired_fallback(fsobject: FsObject, dest_dir: Pat
 
 
 def test_copy_hardlink(fsobject: FsObject, dest_dir: Path):
-    cpy = fsobject.copy(dest_dir, mode=File.CopyMode.symlink)
+    cpy = fsobject.copy(dest_dir, mode=File.CopyMode.hardlink_or_copy)
     assert all(p.parent == dest_dir for p in cpy.fspaths)
     assert set(p.name for p in cpy.fspaths) == set(p.name for p in fsobject.fspaths)
     assert all(
-        os.path.samefile(c, o)
+        c != o and os.path.samefile(c, o)
         for c, o in zip(sorted(cpy.fspaths), sorted(fsobject.fspaths))
         if o.is_file()
     )

fileformats/field/__init__.py

@@ -190,7 +190,7 @@ class Array(
     ] = (Singular,)
     item_type: ty.Optional[ty.Type[Singular[ItemType, ty.Any]]] = None
 
-    primitive = tuple
+    primitive = tuple  # type: ignore[assignment]
 
     value: ty.Tuple[ItemType, ...]