Reimplement expand using SnakemakeFormatter instead of Snakemake's expand

src/snakebids/core/datasets.py

@@ -15,11 +15,11 @@
 from pvandyken.deprecated import deprecated
 from typing_extensions import Self, TypedDict
 
+from snakebids.core.expanding import expand as _expand
 from snakebids.core.filtering import filter_list
 from snakebids.exceptions import DuplicateComponentError
 from snakebids.io.console import get_console_size
 from snakebids.io.printing import format_zip_lists, quote_wrap
-from snakebids.snakemake_compat import expand as sn_expand
 from snakebids.types import ZipList
 from snakebids.utils.containers import ImmutableList, MultiSelectDict
 from snakebids.utils.snakemake_templates import MissingEntityError, SnakemakeFormatter
@@ -122,16 +122,11 @@ def expand(
             Keywords not found in the path will be ignored. Keywords take values or
             lists of values to be expanded over the provided paths.
         """
-        return sn_expand(
+        return _expand(
             list(itx.always_iterable(paths)),
-            allow_missing=allow_missing
-            if isinstance(allow_missing, bool)
-            else list(itx.always_iterable(allow_missing)),
-            **{self.entity: list(dict.fromkeys(self._data))},
-            **{
-                wildcard: list(itx.always_iterable(v))
-                for wildcard, v in wildcards.items()
-            },
+            zip_lists={self.entity: list(dict.fromkeys(self._data))},
+            allow_missing=bool(allow_missing),
+            **wildcards,
         )
 
     def filter(
@@ -349,7 +344,7 @@ def expand(
         paths: Iterable[Path | str] | Path | str,
         /,
         allow_missing: bool | str | Iterable[str] = False,
-        **wildcards: str | Iterable[str],
+        **wildcards: Iterable[str | None] | str | None,
     ) -> list[str]:
         """Safely expand over given paths with component wildcards.
 
@@ -376,39 +371,18 @@ def expand(
             Keywords not found in the path will be ignored. Keywords take values or
             lists of values to be expanded over the provided paths.
         """
-
-        def sequencify(item: bool | str | Iterable[str]) -> bool | list[str]:
-            if isinstance(item, bool):
-                return item
-            return list(itx.always_iterable(item))
-
-        allow_missing_seq = sequencify(allow_missing)
-        if self.zip_lists:
-            inner_expand = list(
-                # order preserving deduplication
-                dict.fromkeys(
-                    sn_expand(
-                        list(itx.always_iterable(paths)),
-                        zip,
-                        allow_missing=True if wildcards else allow_missing_seq,
-                        **self.zip_lists,
-                    )
-                )
-            )
-        else:
-            inner_expand = list(itx.always_iterable(paths))
-        if not wildcards:
-            return inner_expand
-
-        return sn_expand(
-            inner_expand,
-            allow_missing=allow_missing_seq,
-            # Turn all the wildcard items into lists because Snakemake doesn't handle
-            # iterables very well
-            **{
-                wildcard: list(itx.always_iterable(v))
-                for wildcard, v in wildcards.items()
-            },
+        path_list = list(itx.always_iterable(paths))
+
+        # When zip_lists is empty and no extra wildcards, return paths as-is
+        # (no formatting: avoids errors on arbitrary path text with {wildcards})
+        if allow_missing and not self.zip_lists and not wildcards:
+            return path_list
+
+        return _expand(
+            path_list,
+            zip_lists=self.zip_lists,
+            allow_missing=bool(allow_missing),
+            **wildcards,
         )
 
     def filter(
@@ -615,7 +589,7 @@ def expand(
         paths: Iterable[Path | str] | Path | str | None = None,
         /,
         allow_missing: bool | str | Iterable[str] = False,
-        **wildcards: str | Iterable[str],
+        **wildcards: Iterable[str | None] | str | None,
     ) -> list[str]:
         """Safely expand over given paths with component wildcards.
 
@@ -645,7 +619,7 @@ def expand(
             Keywords not found in the path will be ignored. Keywords take values or
             lists of values to be expanded over the provided paths.
         """
-        paths = paths or self.path
+        paths = self.path if paths is None else paths
         return super().expand(paths, allow_missing, **wildcards)
 
     @property

src/snakebids/core/expanding.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import itertools as it
+from collections.abc import Iterable
+from typing import Any
+
+import more_itertools as itx
+
+from snakebids.types import ZipListLike
+from snakebids.utils.snakemake_templates import MissingEntityError, SnakemakeFormatter
+
+try:
+    from snakemake.io import AnnotatedString as _AnnotatedString  # type: ignore
+
+except ImportError:
+
+    def _get_flags(path: Any) -> dict[str, Any] | None:
+        """Return None when snakemake is not installed (no AnnotatedString support)."""
+        return None
+
+    def _make_annotated(s: str, flags: dict[str, Any]) -> str:
+        """Return the string unchanged when snakemake is not installed."""
+        return s
+else:
+
+    def _get_flags(path: Any) -> dict[str, Any] | None:
+        """Extract flags from an AnnotatedString path."""
+        if isinstance(path, _AnnotatedString) and path.flags:  # type: ignore
+            return dict(path.flags)  # type: ignore
+        return None
+
+    def _make_annotated(s: str, flags: dict[str, Any]) -> str:
+        """Create an AnnotatedString with the given flags copied from the template."""
+        result = _AnnotatedString(s)
+        result.flags.update(flags)  # type: ignore
+        return result
+
+
+def expand(
+    paths: Iterable[Any],
+    zip_lists: ZipListLike,
+    allow_missing: bool,
+    **wildcards: Any,
+) -> list[str]:
+    """Expand template paths using SnakemakeFormatter.
+
+    For each template path, iterates over rows of the zip-list and optional
+    extra wildcard combinations (product), formatting each path per row.
+    None values in wildcards are converted to "".
+    Output is order-preserving deduplicated.
+    """
+    formatter = SnakemakeFormatter(allow_missing=allow_missing)
+
+    # Normalize extra wildcards: convert None (scalar or in list) → "", ensure lists
+    extra: dict[str, list[str]] = {}
+    for k, vals in wildcards.items():
+        if vals is None:
+            extra[k] = [""]
+        else:
+            extra[k] = ["" if v is None else v for v in itx.always_iterable(vals)]
+
+    rows = list(zip(*zip_lists.values(), strict=True))
+
+    results: list[str] = []
+    for path in paths:
+        path_str = str(path)
+        flags = _get_flags(path)
+        for row, *combo in it.product(rows, *extra.values()):
+            kwargs = {
+                **dict(zip(zip_lists, row, strict=True)),
+                **dict(zip(extra, combo, strict=True)),
+            }
+            try:
+                result = formatter.vformat(path_str, (), kwargs)
+            except MissingEntityError as err:
+                msg = f"no values given for wildcard {err.entity!r}."
+                raise KeyError(msg) from err
+            except KeyError as err:
+                msg = f"no values given for wildcard {err.args[0]!r}."
+                raise KeyError(msg) from err
+
+            if flags is not None:
+                result = _make_annotated(result, flags)
+            results.append(result)
+
+    return list(dict.fromkeys(results))

src/snakebids/snakemake_compat.py

@@ -9,7 +9,6 @@
     from snakemake import get_argument_parser, main
     from snakemake.io import load_configfile
 
-from snakemake.exceptions import WildcardError
 from snakemake.io import expand
 
 # Handle different snakemake versions for regex function
@@ -22,7 +21,6 @@
 
 __all__ = [
     "Snakemake",
-    "WildcardError",
     "configfile",
     "expand",
     "get_argument_parser",

src/snakebids/snakemake_compat.pyi

@@ -8,8 +8,6 @@ from snakemake.common import configfile as configfile  # type: ignore
 
 configfile: ModuleType
 
-class WildcardError(Exception): ...
-
 def load_configfile(configpath: str) -> dict[str, Any]:
     "Load a JSON or YAML configfile as a dict, then checks that it's a dict."