From 9f9763c02e3dff2097284e73224f14627f7bb5bf Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 5 Aug 2026 07:20:26 +0200 Subject: [PATCH] Remove the speculative FileSystemSource FileSystemSource sketched a provenance-sidecar loading convention (data files next to fingerprint-keyed YAML provenance) but never gained a concrete subclass: nothing implements deserialize_data, and from_path, storage_type, get_content_df, and the module-level sidecar scanner have zero callers in src, tests, or docs. The practical load-data-from-disk path is the Spool node. Keep the small Source ABC the engine consumes (isinstance checks in pipe.py and executor.py) minus its dead optional from_path hook; drop the FileSystemSource re-exports from derzug and derzug.workflow. --- CHANGELOG.md | 1 + src/derzug/__init__.py | 1 - src/derzug/workflow/__init__.py | 2 - src/derzug/workflow/source.py | 154 +------------------------------- 4 files changed, 2 insertions(+), 156 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0deabb6..3eb4249 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ DerZug is pre-alpha; these renames are a clean break with no compatibility shims. Workflows (`.ows` files) saved with an earlier version that reference the affected widgets will not load and must be rebuilt. +- The speculative `derzug.FileSystemSource` (and its provenance-sidecar loading helpers) was removed; it had no concrete subclass and no callers. The small `Source` ABC the workflow engine consumes remains. - The `derzug.orange` module (the `Setting` subclass) moved to `derzug.settings`. Update `from derzug.orange import Setting` to `from derzug.settings import Setting`. diff --git a/src/derzug/__init__.py b/src/derzug/__init__.py index d5cc47f..67e719d 100644 --- a/src/derzug/__init__.py +++ b/src/derzug/__init__.py @@ -5,7 +5,6 @@ def __getattr__(name: str): if name in { "CompiledWorkflow", "compile_workflow", - "FileSystemSource", "Pipe", "PipeBuilder", "Provenance", diff --git a/src/derzug/workflow/__init__.py b/src/derzug/workflow/__init__.py index b3b26e4..c036010 100644 --- a/src/derzug/workflow/__init__.py +++ b/src/derzug/workflow/__init__.py @@ -11,7 +11,6 @@ __all__ = ( "CompiledWorkflow", "compile_workflow", - "FileSystemSource", "Pipe", "PipeBuilder", "Provenance", @@ -25,7 +24,6 @@ _EXPORTS = { "CompiledWorkflow": (".compiler", "CompiledWorkflow"), "compile_workflow": (".compiler", "compile_workflow"), - "FileSystemSource": (".source", "FileSystemSource"), "Pipe": (".pipe", "Pipe"), "PipeBuilder": (".graph", "PipeBuilder"), "Provenance": (".provenance", "Provenance"), diff --git a/src/derzug/workflow/source.py b/src/derzug/workflow/source.py index c6920d7..18a46e1 100644 --- a/src/derzug/workflow/source.py +++ b/src/derzug/workflow/source.py @@ -7,12 +7,7 @@ import warnings from abc import ABC, abstractmethod from collections.abc import Iterator -from functools import lru_cache -from pathlib import Path -from typing import ClassVar, Literal, Self, TypeVar - -import pandas as pd -from pydantic import Field +from typing import TypeVar from .model import WorkflowModel from .provenance import Provenance @@ -20,24 +15,6 @@ DataType = TypeVar("DataType") -def get_provmap_and_fingerprints_from_path(path, fmt_str=".yaml"): - """ - Get the provenance map and managed fingerprints from a source path. - """ - destination_path = Path(path) - assert destination_path.is_dir(), "A directory is required." - if fmt_str and not fmt_str.startswith("."): - fmt_str = f".{fmt_str}" - provenance_map = {} - fingerprints = [] - for prov_path in destination_path.glob(f"*{fmt_str}"): - fingerprint = prov_path.name[: -len(fmt_str)] - fingerprint = fingerprint.rstrip(".") - provenance_map[fingerprint] = Provenance.load(prov_path) - fingerprints.append(fingerprint) - return provenance_map, fingerprints - - class Source[DataType](WorkflowModel, ABC): """ A Source of pipe inputs. @@ -78,132 +55,3 @@ def __getitem__(self, item) -> DataType: def __iter__(self) -> Iterator[DataType]: """Iterate over the source.""" raise NotImplementedError("Not implemented") - - # Optional methods. - @classmethod - def from_path(cls, path, provenance=None, **kwargs) -> Self: - """Load a source from a path.""" - raise NotImplementedError("Not implemented") - - -class FileSystemSource[DataType](Source[DataType], ABC): - """ - Base class for sources that read from the filesystem. - """ - - path: Path - provenance: tuple[Provenance, ...] = Field(default_factory=tuple) - data_extension: ClassVar[str] = "" - # Indicates if the sink data is stored in a single file next to the - # provenance or in a directory. - storage_type: Literal["file", "directory"] = "directory" - - @classmethod - def from_path(cls, path, provenance=None, fingerprint=None, **kwargs): - """ - Load a source from a provenance or data path. - """ - path = Path(path) - if provenance is not None: - normalized = cls._normalize_provenance(provenance) - return cls(path=path, provenance=normalized) - # This should support two modes; if you pass a directory with data - # files, or if you pass the Sink directory plus the fingerprint. - if fingerprint is not None: - prov_map, fps = get_provmap_and_fingerprints_from_path(path) - if isinstance(fingerprint, int): - fingerprint = fps[fingerprint] - provenance = prov_map.get(fingerprint) - path = path / fingerprint - else: - fingerprint = path.stem - prov_map, _ = get_provmap_and_fingerprints_from_path(path.parent) - provenance = prov_map.get(fingerprint) - # In this case the provenance is not on the same level as the - # data file, we need to look inside the data file. - if provenance is None: - prov_map, _ = get_provmap_and_fingerprints_from_path(path) - provenance = prov_map.get(fingerprint) - normalized = cls._normalize_provenance(provenance) - return cls(path=path, provenance=normalized) - - @classmethod - def _normalize_provenance( - cls, provenance: Provenance | tuple[Provenance, ...] | None - ) -> tuple[Provenance, ...]: - """Normalize provenance inputs into a tuple.""" - if provenance is None: - return () - if hasattr(provenance, "to_source_provenance"): - return provenance.to_source_provenance() - if isinstance(provenance, tuple): - return provenance - return (provenance,) - - @lru_cache - def get_content_df(self) -> pd.DataFrame: - """Get a dataframe containing the contents of this source.""" - out = [] - if self.data_extension: - pattern = f"*.{self.data_extension}" - else: - pattern = "*" - root = Path(self.path) - for path in root.rglob(pattern): - stat = path.stat() - data_root = root - if root.is_dir(): - try: - rel = path.relative_to(root) - except ValueError: - rel = None - if rel is not None and rel.parts: - first_path = root / rel.parts[0] - data_root = first_path if first_path.is_dir() else root - sub = { - "st_size": stat.st_size, - "st_mtime": stat.st_mtime, - "st_ctime": stat.st_ctime, - } - sub["path"] = str(path) - sub["data_root"] = str(data_root) - out.append(sub) - df = pd.DataFrame(out) - if not df.empty: - df = df.sort_values(["st_ctime", "st_mtime", "path"], kind="mergesort") - return df - - def get_single_data(self) -> DataType: - """ - Get the first data from a filesystem source. - - If multiple directory roots exist, return the first directory's data. - """ - df = self.get_content_df() - if df.empty: - msg = f"Source: {self} contains no data!" - raise ValueError(msg) - root_col = "data_root" if "data_root" in df.columns else "path" - root_values = df[root_col] - if len(root_values.unique()) > 1: - msg = f"Source: {self} contains more than one data!" - warnings.warn(msg) - first_root = Path(root_values.iloc[0]) - if first_root.is_dir(): - return self.deserialize_data(first_root) - return self.deserialize_data(Path(df["path"].iloc[0])) - - def __len__(self) -> int: - return len(self.get_content_df()) - - def __iter__(self) -> Iterator[DataType]: - for path in self.get_content_df()["path"].values: - yield self.deserialize_data(Path(path)) - - def __getitem__(self, item) -> DataType: - path = self.get_content_df()["path"].values[item] - return self.deserialize_data(Path(path)) - - @abstractmethod - def deserialize_data(self, path: Path) -> DataType: - """Deserialize the contents of a path."""