tested out optional content types for TypedSet and TypedDirectory

Author: tclose

fileformats/core/collection.py

@@ -3,7 +3,8 @@
 from abc import ABCMeta, abstractproperty
 from fileformats.core import FileSet, validated_property, mtime_cached_property
 from fileformats.core.decorators import classproperty
-from fileformats.core.exceptions import FormatDefinitionError, FormatMismatchError
+from fileformats.core.exceptions import FormatMismatchError
+from fileformats.core.utils import get_optional_type
 
 
 class TypedCollection(FileSet, metaclass=ABCMeta):
@@ -50,16 +51,7 @@ def _validate_required_content_types(self) -> None:
     def potential_content_types(cls) -> ty.Tuple[ty.Type[FileSet], ...]:
         content_types: ty.List[ty.Type[FileSet]] = []
         for content_type in cls.content_types:  # type: ignore[assignment]
-            if ty.get_origin(content_type) is ty.Union:
-                args = ty.get_args(content_type)
-                if not len(args) == 2 and None not in args:
-                    raise FormatDefinitionError(
-                        "Only Optional types are allowed in content_type definitions, "
-                        f"not {content_type}"
-                    )
-                content_types.append(args[0] if args[0] is not None else args[1])
-            else:
-                content_types.append(content_type)  # type: ignore[arg-type]
+            content_types.append(get_optional_type(content_type))  # type: ignore[arg-type]
         return tuple(content_types)
 
     @classproperty

fileformats/core/mixin.py

@@ -4,7 +4,7 @@
 import logging
 from .datatype import DataType
 import fileformats.core
-from .utils import describe_task, matching_source
+from .utils import describe_task, matching_source, get_optional_type
 from .decorators import validated_property, classproperty
 from .identification import to_mime_format_name
 from .converter_helpers import SubtypeVar, ConverterSpec
@@ -292,6 +292,7 @@ def my_func(file: MyFormatWithClassifiers[Integer]):
     # Default values for class attrs
     multiple_classifiers = True
     allowed_classifiers: ty.Optional[ty.Tuple[ty.Type[Classifier], ...]] = None
+    allow_optional_classifiers = False
     exclusive_classifiers: ty.Tuple[ty.Type[Classifier], ...] = ()
     ordered_classifiers = False
     generically_classifiable = False
@@ -320,7 +321,9 @@ def wildcard_classifiers(
     ) -> ty.FrozenSet[ty.Type[SubtypeVar]]:
         if classifiers is None:
             classifiers = cls.classifiers if cls.is_classified else ()
-        return frozenset(t for t in classifiers if issubclass(t, SubtypeVar))
+        return frozenset(
+            t for t in classifiers if issubclass(get_optional_type(t), SubtypeVar)  # type: ignore[misc]
+        )
 
     @classmethod
     def non_wildcard_classifiers(
@@ -329,7 +332,9 @@ def non_wildcard_classifiers(
         if classifiers is None:
             classifiers = cls.classifiers if cls.is_classified else ()
         assert classifiers is not None
-        return frozenset(q for q in classifiers if not issubclass(q, SubtypeVar))
+        return frozenset(
+            q for q in classifiers if not issubclass(get_optional_type(q), SubtypeVar)
+        )
 
     @classmethod
     def __class_getitem__(
@@ -341,11 +346,15 @@ def __class_getitem__(
             classifiers_tuple = tuple(classifiers)
         else:
             classifiers_tuple = (classifiers,)
+        classifiers_to_check = tuple(
+            get_optional_type(c, cls.allow_optional_classifiers)
+            for c in classifiers_tuple
+        )
 
         if cls.allowed_classifiers:
             not_allowed = [
                 q
-                for q in classifiers_tuple
+                for q in classifiers_to_check
                 if not any(issubclass(q, t) for t in cls.allowed_classifiers)
             ]
             if not_allowed:
@@ -357,15 +366,17 @@ def __class_getitem__(
         if cls.multiple_classifiers:
             if not cls.ordered_classifiers:
                 # Check for duplicate classifiers in the multiple list
-                if len(classifiers_tuple) > 1:
+                if len(classifiers_to_check) > 1:
                     # Sort the classifiers into categories and ensure that there aren't more
                     # than one type for each category. Otherwise, if the classifier doesn't
                     # belong to a category, check to see that there aren't multiple sub-classes
                     # in the classifier set
                     repetitions: ty.Dict[
                         ty.Type[Classifier], ty.List[ty.Type[Classifier]]
-                    ] = {c: [] for c in cls.exclusive_classifiers + classifiers_tuple}
-                    for classifier in classifiers_tuple:
+                    ] = {
+                        c: [] for c in cls.exclusive_classifiers + classifiers_to_check
+                    }
+                    for classifier in classifiers_to_check:
                         for exc_classifier in repetitions:
                             if issubclass(classifier, exc_classifier):
                                 repetitions[exc_classifier].append(classifier)
@@ -381,7 +392,10 @@ def __class_getitem__(
                             )
                         )
                 classifiers_tuple = tuple(
-                    sorted(set(classifiers_tuple), key=lambda x: x.__name__)
+                    sorted(
+                        set(classifiers_tuple),
+                        key=lambda x: get_optional_type(x).__name__,
+                    )
                 )
         else:
             if len(classifiers_tuple) > 1:

fileformats/core/utils.py

@@ -11,6 +11,7 @@
 from contextlib import contextmanager
 from .typing import FspathsInputType
 import fileformats.core
+from fileformats.core.exceptions import FormatDefinitionError
 
 if ty.TYPE_CHECKING:
     import pydra.engine.core
@@ -228,3 +229,39 @@ def import_extras_module(klass: ty.Type["fileformats.core.DataType"]) -> ExtrasM
     else:
         extras_imported = True
     return ExtrasModule(extras_imported, extras_pkg, extras_pypi)
+
+
+TypeType = ty.TypeVar("TypeType", bound=ty.Type[ty.Any])
+
+
+def get_optional_type(
+    type_: ty.Union[TypeType, ty.Type[ty.Optional[TypeType]]], allowed: bool = True
+) -> TypeType:
+    """Checks if a type is an Optional type
+
+    Parameters
+    ----------
+    type_ : ty.Type
+        the type to check
+    allowed : bool
+        whether Optional types are allowed or not
+
+    Returns
+    -------
+    bool
+        whether the type is an Optional type or not
+    """
+    if ty.get_origin(type_) is None:
+        return type_  # type: ignore[return-value]
+    if not allowed:
+        raise FormatDefinitionError(
+            f"Optional types are not allowed in content_type definitions ({type_}) "
+            "in this context"
+        )
+    args = ty.get_args(type_)
+    if len(args) != 2 and None in ty.get_args(type_):
+        raise FormatDefinitionError(
+            "Only Optional types are allowed in content_type definitions, "
+            f"not {type_}"
+        )
+    return args[0] if args[0] is not None else args[1]  # type: ignore[no-any-return]

fileformats/generic/directory.py

@@ -110,4 +110,5 @@ class DirectoryOf(WithClassifiers, TypedDirectory):  # type: ignore[misc]
     # WithClassifiers-required class attrs
     classifiers_attr_name = "content_types"
     allowed_classifiers = (FileSet,)
+    allow_optional_classifiers = True
     generically_classifiable = True

fileformats/generic/set.py

@@ -1,8 +1,10 @@
 import typing as ty
+import itertools
 from pathlib import Path
-from fileformats.core import FileSet
+from fileformats.core import FileSet, validated_property
 from fileformats.core.mixin import WithClassifiers
 from fileformats.core.collection import TypedCollection
+from fileformats.core.exceptions import FormatMismatchError
 
 
 class TypedSet(TypedCollection):
@@ -25,9 +27,21 @@ def __repr__(self) -> str:
             paths_repr += ", ..."
         return f"{self.type_name}({paths_repr})"
 
+    @validated_property
+    def _all_paths_used(self) -> None:
+        all_contents_paths = set(itertools.chain(*(c.fspaths for c in self.contents)))
+        missing = self.fspaths - all_contents_paths
+        if missing:
+            contents_str = "\n".join(repr(c) for c in self.contents)
+            raise FormatMismatchError(
+                f"Paths {[str(p) for p in missing]} are not used by any of the "
+                f"contents of {self.type_name}:\n{contents_str}"
+            )
+
 
 class SetOf(WithClassifiers, TypedSet):  # type: ignore[misc]
     # WithClassifiers-required class attrs
     classifiers_attr_name = "content_types"
     allowed_classifiers = (FileSet,)
+    allow_optional_classifiers = True
     generically_classifiable = True

fileformats/generic/tests/test_directory.py

@@ -1,5 +1,8 @@
+import typing as ty
+import pytest
 from fileformats.generic import Directory, DirectoryOf, SetOf
-from fileformats.testing import MyFormatGz
+from fileformats.testing import MyFormatGz, YourFormat, EncodedText
+from fileformats.core.exceptions import FormatMismatchError
 
 
 def test_sample_directory():
@@ -16,3 +19,43 @@ def test_sample_set_of():
     sample = SetOf[MyFormatGz].sample()
     assert isinstance(sample, SetOf)
     assert all(isinstance(c, MyFormatGz) for c in sample.contents)
+
+
+def test_directory_optional_contents(tmp_path):
+    my_format = MyFormatGz.sample(dest_dir=tmp_path)
+    sample_dir = DirectoryOf[MyFormatGz](tmp_path)
+    EncodedText.sample(dest_dir=tmp_path)
+    assert sample_dir.contents == [my_format]
+
+    with pytest.raises(
+        FormatMismatchError, match="Did not find the required content types"
+    ):
+        DirectoryOf[MyFormatGz, YourFormat](sample_dir)
+
+    optional_dir = DirectoryOf[MyFormatGz, ty.Optional[YourFormat]](sample_dir)
+    assert optional_dir.contents == [my_format]
+    your_format = YourFormat.sample(dest_dir=tmp_path)
+    optional_dir = DirectoryOf[MyFormatGz, ty.Optional[YourFormat]](sample_dir)
+    assert optional_dir.contents == [my_format, your_format]
+    required_dir = DirectoryOf[MyFormatGz, YourFormat](sample_dir)
+    assert required_dir.contents == [my_format, your_format]
+
+
+def test_set_optional_contents():
+    my_format = MyFormatGz.sample()
+    your_format = YourFormat.sample()
+
+    sample_set = SetOf[MyFormatGz, YourFormat](my_format, your_format)
+    assert sample_set.contents == [my_format, your_format]
+    with pytest.raises(
+        FormatMismatchError, match="are not used by any of the contents of "
+    ):
+        SetOf[MyFormatGz](my_format, your_format)
+    with pytest.raises(
+        FormatMismatchError, match="Did not find the required content types"
+    ):
+        SetOf[MyFormatGz, YourFormat](my_format)
+    sample_set = SetOf[MyFormatGz, ty.Optional[YourFormat]](my_format)
+    assert sample_set.contents == [my_format]
+    sample_set = SetOf[MyFormatGz, ty.Optional[YourFormat]](my_format, your_format)
+    assert sample_set.contents == [my_format, your_format]