# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Reusable recent-file history and storage abstractions.
.. versionadded:: v2.4
:Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
:Description: Maintain bounded recent-file history independently of any graphical interface.
"""
from __future__ import annotations
import json
from collections.abc import Callable, Iterable, Sequence
from pathlib import Path
from typing import Protocol
[docs]
class RecentFileStorage(Protocol):
"""Define the replaceable persistence interface used by recent-file histories.
.. versionadded:: v2.4
"""
[docs]
def read(self) -> Sequence[str]:
"""Return stored paths without interpreting or validating them."""
[docs]
def write(self, paths: Sequence[str]) -> None:
"""Persist paths in the order supplied by the history service."""
[docs]
class JsonRecentFileStorage:
"""Persist recent-file history as a UTF-8 JSON array at a configured path.
.. versionadded:: v2.4
"""
def __init__(self, path: Path | str) -> None:
self._path = Path(path)
[docs]
def read(self) -> Sequence[str]:
"""Read the stored path list, treating unavailable or invalid data as empty."""
try:
data = json.loads(self._path.read_text(encoding='utf-8'))
except (OSError, UnicodeError, json.JSONDecodeError):
return []
if not isinstance(data, list) or not all(isinstance(item, str) for item in data):
return []
return data
[docs]
def write(self, paths: Sequence[str]) -> None:
"""Write the path list, creating its parent directory when needed."""
self._path.parent.mkdir(parents=True, exist_ok=True)
self._path.write_text(json.dumps(list(paths), ensure_ascii=False), encoding='utf-8')
[docs]
class RecentFileService:
"""Maintain configurable, normalized, duplicate-free recent-file history.
Storage and path-existence checks are injectable so the service can be shared
by independent MAFw applications and tested without constructing GUI widgets.
.. versionadded:: v2.4
"""
def __init__(
self,
storage: RecentFileStorage,
*,
max_entries: int = 10,
exists: Callable[[Path], bool] = Path.is_file,
) -> None:
if max_entries < 1:
raise ValueError('max_entries must be at least one.')
self._storage = storage
self._max_entries = max_entries
self._exists = exists
self._paths: list[Path] = []
self._restore()
[docs]
def files(self) -> list[Path]:
"""Return existing recent files, pruning stale paths from storage."""
self._prune_missing()
# Keep callers from changing service-owned history without its ordering and persistence rules.
return list(self._paths)
[docs]
def add(self, path: Path | str) -> list[Path]:
"""Promote a file to the top of history and return the updated entries."""
normalized = self._normalize(path)
self._paths = [normalized, *(item for item in self._paths if item != normalized)][: self._max_entries]
self._persist()
# Return a snapshot so later caller edits cannot silently alter the service state.
return list(self._paths)
[docs]
def _restore(self) -> None:
"""Load persisted entries and normalize, de-duplicate, limit, and prune them."""
try:
stored = self._storage.read()
except Exception:
stored = []
restored: list[Path] = []
for raw_path in stored:
try:
path = self._normalize(raw_path)
except (OSError, RuntimeError, ValueError):
continue
if path not in restored and self._is_file(path):
restored.append(path)
if len(restored) >= self._max_entries:
break
self._paths = restored
self._persist()
[docs]
def _normalize(self, path: Path | str) -> Path:
"""Resolve a path to a stable absolute representation without requiring it to exist."""
return Path(path).expanduser().resolve(strict=False)
[docs]
def _is_file(self, path: Path) -> bool:
"""Check file availability while treating filesystem errors as stale entries."""
try:
return self._exists(path)
except OSError:
return False
[docs]
def _prune_missing(self) -> None:
"""Remove paths that disappeared after they were added to history."""
existing = [path for path in self._paths if self._is_file(path)]
if existing != self._paths:
self._paths = existing
self._persist()
[docs]
def _persist(self) -> None:
"""Persist history without allowing storage failures to break file operations."""
try:
self._storage.write([str(path) for path in self._paths])
except Exception:
# File operations remain successful when the history location is unavailable.
pass
[docs]
class InMemoryRecentFileStorage:
"""Keep recent-file data in memory for applications without persistent storage.
.. versionadded:: v2.4
"""
def __init__(self, paths: Iterable[str] = ()) -> None:
self._paths = list(paths)
[docs]
def read(self) -> Sequence[str]:
"""Return the currently stored paths."""
return list(self._paths)
[docs]
def write(self, paths: Sequence[str]) -> None:
"""Replace the stored paths."""
self._paths = list(paths)