Atlas - install_db.py
Home / npkg-testing / tools / npkg Lines: 1 | Size: 3633 bytes [Download] [Show on GitHub] [Search similar files] [Raw] [Raw (proxy)][FILE BEGIN]1import json 2from pathlib import Path 3from typing import Any, Dict, List 4 5from .paths import list_manifest_paths, manifest_root, package_manifest_path 6from .types import Package 7 8 9def write_install_manifest(pkg: Package, install_root: Path, members: List[str]) -> None: 10 db_dir = manifest_root(install_root) 11 db_dir.mkdir(parents=True, exist_ok=True) 12 manifest = { 13 "package": pkg.name, 14 "version": pkg.version, 15 "install_root": str(install_root), 16 "paths": members, 17 } 18 manifest_path = package_manifest_path(pkg.name, install_root) 19 with manifest_path.open("w", encoding="utf-8") as handle: 20 json.dump(manifest, handle, indent=2) 21 22 23def read_install_manifest(pkg_name: str, install_root: Path) -> Dict[str, Any]: 24 manifest_path = package_manifest_path(pkg_name, install_root) 25 if not manifest_path.exists(): 26 raise RuntimeError(f"package '{pkg_name}' is not installed in {install_root}") 27 with manifest_path.open("r", encoding="utf-8") as handle: 28 data = json.load(handle) 29 if not isinstance(data, dict): 30 raise RuntimeError(f"invalid manifest for package '{pkg_name}'") 31 return data 32 33 34def uninstall_from_manifest(pkg_name: str, install_root: Path) -> None: 35 data = read_install_manifest(pkg_name, install_root) 36 raw_paths = data.get("paths", []) 37 if not isinstance(raw_paths, list): 38 raise RuntimeError(f"invalid manifest path list for package '{pkg_name}'") 39 40 unique_paths = [] 41 seen = set() 42 for item in raw_paths: 43 if not isinstance(item, str): 44 continue 45 normalized = item.strip().lstrip("/") 46 if not normalized or normalized in seen: 47 continue 48 seen.add(normalized) 49 unique_paths.append(normalized) 50 51 for rel_path in sorted(unique_paths, key=lambda value: (value.count("/"), len(value)), reverse=True): 52 target = install_root / rel_path 53 if target.is_symlink() or target.is_file(): 54 target.unlink(missing_ok=True) 55 elif target.is_dir(): 56 try: 57 target.rmdir() 58 except OSError: 59 pass 60 61 for rel_path in sorted(unique_paths, key=lambda value: value.count("/"), reverse=True): 62 current = (install_root / rel_path).parent 63 while current != install_root and current.exists(): 64 try: 65 current.rmdir() 66 except OSError: 67 break 68 current = current.parent 69 70 manifest_path = package_manifest_path(pkg_name, install_root) 71 manifest_path.unlink(missing_ok=True) 72 73 db_dir = manifest_root(install_root) 74 try: 75 db_dir.rmdir() 76 except OSError: 77 pass 78 79 80def load_installed_rows(install_root: Path) -> List[Dict[str, Any]]: 81 rows: List[Dict[str, Any]] = [] 82 for manifest_path in list_manifest_paths(install_root): 83 try: 84 with manifest_path.open("r", encoding="utf-8") as handle: 85 data = json.load(handle) 86 except (OSError, json.JSONDecodeError) as error: 87 rows.append( 88 { 89 "manifest": manifest_path, 90 "error": str(error), 91 } 92 ) 93 continue 94 95 rows.append( 96 { 97 "manifest": manifest_path, 98 "package": str(data.get("package") or manifest_path.stem), 99 "version": str(data.get("version") or "unknown"), 100 "path_count": len(data.get("paths", [])) if isinstance(data.get("paths", []), list) else 0, 101 "error": None, 102 } 103 ) 104 105 return rows 106[FILE END](C) 2025 0x4248 (C) 2025 4248 Media and 4248 Systems, All part of 0x4248 See LICENCE files for more information. Not all files are by 0x4248 always check Licencing.