AbstractVersionedDataset

def _clear_version_cache(self) -> None:
    """Clear both load and save version caches."""
    self._cached_load_version = None
    self._cached_save_version = None

def _fetch_latest_load_version(self) -> str:
    """Fetch the most recent existing version from the given path.
    Results are cached to avoid repeated filesystem operations.
    """
    # Return cached version if available
    if self._cached_load_version is not None:
        return self._cached_load_version

    # When load version is unpinned, fetch the most recent existing
    # version from the given path.
    version_paths = self._get_versions()

    most_recent = next((path for path in version_paths), None)
    if not most_recent:
        message = f"Did not find any versions for {self}"
        raise VersionNotFoundError(message)

    # Cache and return the result
    self._cached_load_version = PurePath(most_recent).parent.name
    return self._cached_load_version

def _fetch_latest_save_version(self) -> str:
    """Generate and cache the current save version"""
    # Return cached version if available
    if self._cached_save_version is not None:
        return self._cached_save_version

    # Generate new timestamp and cache it
    self._cached_save_version = generate_timestamp()
    return self._cached_save_version

def _get_load_path(self) -> PurePosixPath:
    if not self._version:
        # When versioning is disabled, load from original filepath
        return self._filepath

    load_version = self.resolve_load_version()
    return self._get_versioned_path(load_version)  # type: ignore[arg-type]

def _get_save_path(self) -> PurePosixPath:
    if not self._version:
        # When versioning is disabled, return original filepath
        return self._filepath

    save_version = self.resolve_save_version()
    versioned_path = self._get_versioned_path(save_version)  # type: ignore[arg-type]

    if self._exists_function(str(versioned_path)):
        raise DatasetError(
            f"Save path '{_redact_url_credentials(str(versioned_path))}' for "
            f"{self!s} must not exist if versioning is enabled."
        )

    return versioned_path

def _get_version_from_path(self, path: str) -> str:
    """Extract version string from a versioned dataset path.

    Args:
        path: Full path to a versioned dataset file.

    Returns:
        Version string extracted from the path.
    """
    return PurePath(path).parent.name

def _get_versioned_path(self, version: str) -> PurePosixPath:
    if _is_unsafe_version(version):
        raise DatasetError(
            f"Version string '{version}' is not allowed. "
            "Version strings must be a single non-empty path component with no "
            "path separators ('/' or '\\') and must not be '.' or '..'."
        )
    return self._filepath / version / self._filepath.name

def _get_versions(self) -> list[str]:
    """Get all existing versions for this dataset, sorted by most recent first."""
    pattern = str(self._get_versioned_path("*"))
    try:
        version_paths = sorted(self._glob_function(pattern), reverse=True)
    except Exception as exc:
        message = (
            f"Did not find any versions for {self}. This could be "
            f"due to insufficient permission. Exception: {exc}"
        )
        raise VersionNotFoundError(message) from exc

    return [path for path in version_paths if self._exists_function(path)]

def _release(self) -> None:
    super()._release()
    self._clear_version_cache()

@classmethod
def _save_wrapper(
    cls, save_func: Callable[[Self, _DI], None]
) -> Callable[[Self, _DI], None]:
    """Decorate `save_func` with logging and error handling code."""

    @wraps(save_func)
    def save(self: Self, data: _DI) -> None:
        self._clear_version_cache()
        save_version = (
            self.resolve_save_version()
        )  # Make sure last save version is set
        try:
            super()._save_wrapper(save_func)(self, data)
        except (FileNotFoundError, NotADirectoryError) as err:
            # FileNotFoundError raised in Win, NotADirectoryError raised in Unix
            _default_version = "YYYY-MM-DDThh.mm.ss.sssZ"
            filepath_name = _redact_url_credentials(self._filepath.name)
            filepath_parent = _redact_url_credentials(
                self._filepath.parent.as_posix()
            )
            filepath_posix = _redact_url_credentials(self._filepath.as_posix())
            raise DatasetError(
                f"Cannot save versioned dataset '{filepath_name}' to "
                f"'{filepath_parent}' because a file with the same "
                f"name already exists in the directory. This is likely because "
                f"versioning was enabled on a dataset already saved previously. Either "
                f"remove '{filepath_name}' from the directory or manually "
                f"convert it into a versioned dataset by placing it in a versioned "
                f"directory (e.g. with default versioning format "
                f"'{filepath_posix}/{_default_version}/{filepath_name}"
                f"')."
            ) from err

        load_version = self.resolve_load_version()
        if load_version != save_version:
            warnings.warn(
                _CONSISTENCY_WARNING.format(save_version, load_version, str(self))
            )
            self._clear_version_cache()

    return save

def exists(self) -> bool:
    """Checks whether a dataset's output already exists by calling
    the provided _exists() method.

    Returns:
        Flag indicating whether the output already exists.

    Raises:
        DatasetError: when underlying exists method raises error.

    """
    self._logger.debug("Checking whether target of %s exists", str(self))
    try:
        return self._exists()
    except VersionNotFoundError:
        return False
    except Exception as exc:  # SKIP_IF_NO_SPARK
        message = f"Failed during exists check for dataset {self!s}.\n{exc!s}"
        raise DatasetError(message) from exc

def list_versions(self, full_path: bool = True) -> list[str]:
    """List all available versions of this dataset.

    This method allows you to retrieve all existing versions of a versioned
    dataset. It's useful for tracking dataset history, auditing changes, or
    implementing custom version selection logic.

    Args:
        full_path: If True, returns the full path to each version.
            If False, returns only the version strings (timestamps).

    Returns:
        A list of version paths (if full_path=True) or version strings
        (if full_path=False), sorted in reverse chronological order
        (most recent first).

    Raises:
        VersionNotFoundError: If the dataset has no versions or if there
            are permission issues accessing the version directory.

    Example:
        ::

            >>> dataset = MyVersionedDataset(
            ...     filepath="data/model.pkl",
            ...     version=Version(None, None)
            ... )
            >>> # After saving multiple versions...
            >>> versions = dataset.list_versions(full_path=False)
            >>> print(versions)
            ['2024-01-15T10.30.00.000Z', '2024-01-14T09.15.00.000Z']
            >>> # Get full paths
            >>> paths = dataset.list_versions(full_path=True)
            >>> print(paths[0])
            'data/model.pkl/2024-01-15T10.30.00.000Z/model.pkl'
    """
    version_paths = self._get_versions()

    if full_path:
        return version_paths

    return [self._get_version_from_path(path) for path in version_paths]

def resolve_load_version(self) -> str | None:
    """Compute the version the dataset should be loaded with."""
    if not self._version:
        return None
    if self._version.load:
        return self._version.load  # type: ignore[no-any-return]
    return self._fetch_latest_load_version()

def resolve_save_version(self) -> str | None:
    """Compute the version the dataset should be saved with."""
    if not self._version:
        return None
    if self._version.save:
        return self._version.save  # type: ignore[no-any-return]
    return self._fetch_latest_save_version()