Coverage for collectra/gui/session_cache.py: 95.74%

1"""Best-effort, per-user storage of each pipeline's open GUI folders.""" 

2 

3import hashlib 

4import json 

5import logging 

6import os 

7import tempfile 

8from pathlib import Path 

9 

10from platformdirs import user_cache_dir 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15class FolderSessionCache: 

16 def __init__(self, pipeline_directory: Path): 

17 key = hashlib.sha256( 

18 str(Path(pipeline_directory).resolve()).encode() 

19 ).hexdigest() 

20 self.path = ( 

21 Path(user_cache_dir("collectra", appauthor=False)) 

22 / "gui-sessions" 

23 / f"{key}.json" 

24 ) 

25 

26 self.active_path = self.path.with_name(f"{key}-active.json") 

27 

28 def read_active(self) -> str | None: 

29 try: 

30 active = json.loads(self.active_path.read_text(encoding="utf-8")) 

31 return active if isinstance(active, str) else None 

32 except (OSError, ValueError): 

33 return None 

34 

35 def write_active(self, folder: str) -> None: 

36 self._write_json(self.active_path, folder) 

37 

38 def read(self) -> list[str]: 

39 try: 

40 data = json.loads(self.path.read_text(encoding="utf-8")) 

41 folders = data.get("folders") if isinstance(data, dict) else None 

42 if isinstance(folders, list) and all( 

43 isinstance(path, str) for path in folders 

44 ): 

45 return folders 

46 except (OSError, ValueError): 

47 pass 

48 return [] 

49 

50 def write(self, folders: list[str]) -> None: 

51 self._write_json(self.path, {"folders": folders}) 

52 

53 def _write_json(self, destination: Path, data) -> None: 

54 temporary = None 

55 try: 

56 destination.parent.mkdir(parents=True, exist_ok=True) 

57 with tempfile.NamedTemporaryFile( 

58 mode="w", encoding="utf-8", dir=destination.parent, delete=False 

59 ) as stream: 

60 temporary = Path(stream.name) 

61 json.dump(data, stream) 

62 os.replace(temporary, destination) 

63 except OSError as error: 

64 logger.warning("Could not save GUI folder session: %s", error) 

65 finally: 

66 if temporary is not None: 

67 try: 

68 temporary.unlink(missing_ok=True) 

69 except OSError: 

70 pass