diff --git a/pasch/modules/__init__.py b/pasch/modules/__init__.py index 1ccd902..dd51a5b 100644 --- a/pasch/modules/__init__.py +++ b/pasch/modules/__init__.py @@ -1,7 +1,9 @@ from .echo import Echo +from .files import Files from .pacman import Pacman __all__: list[str] = [ "Echo", + "Files", "Pacman", ] diff --git a/pasch/modules/files.py b/pasch/modules/files.py new file mode 100644 index 0000000..c36ac40 --- /dev/null +++ b/pasch/modules/files.py @@ -0,0 +1,155 @@ +import hashlib +import json +import random +import string +from pathlib import Path + +from rich.markup import escape + +from pasch.orchestrator import Module, Orchestrator + + +def random_tmp_path(path: Path) -> Path: + prefix = "" if path.name.startswith(".") else "." + suffix = random.sample(string.ascii_letters + string.digits, 6) + name = f"{prefix}{path.name}.{suffix}~pasch" + return path.with_name(name) + + +def atomic_write(path: Path, content: bytes) -> None: + tmp_path = random_tmp_path(path) + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path.write_bytes(content) + tmp_path.rename(path) + + +def hash_data(data: bytes) -> str: + m = hashlib.sha256() + m.update(data) + return f"sha256-{m.hexdigest()}" + + +def hash_file(path: Path) -> str | None: + try: + data = path.read_bytes() + except FileNotFoundError: + return None + return hash_data(data) + + +def path_to_str(path: Path) -> str: + return str(path.resolve()) + + +class FileDb: + def __init__(self, path: Path) -> None: + self._path = path + + def _load(self) -> dict[str, str]: + try: + text = self._path.read_text(encoding="utf-8") + except FileNotFoundError: + return {} + + data = json.loads(text) + if type(data) is not dict: + raise ValueError("file db is not a dict") + for k, v in data.items(): + if type(v) is not str: + raise ValueError(f"file db contains non-str hash at key {k!r}") + return data + + def _save(self, data: dict[str, str]) -> None: + atomic_write(self._path, json.dumps(data).encode("utf-8")) + + def add_hash(self, path: Path, hash: str) -> None: + data = self._load() + data[path_to_str(path)] = hash + self._save(data) + + def remove_hash(self, path: Path) -> None: + data = self._load() + data.pop(path_to_str(path), None) + self._save(data) + + def verify_hash(self, path: Path, cur_hash: str | None) -> str | None: + cur_hash = hash_file(path) + if cur_hash is None: + return + known_hash = self._load().get(path_to_str(path)) + if known_hash is None: + return "File unknown and contents don't match target state." + if known_hash != cur_hash: + return "File contents don't match the last known or target state." + + def paths(self) -> list[str]: + return list(sorted(self._load().keys())) + + +class Files(Module): + def __init__( + self, + orchestrator: Orchestrator, + file_db_name: str = "files.json", + root: Path | None = None, + ) -> None: + super().__init__(orchestrator) + self._files: dict[str, bytes] = {} + self._file_db = FileDb(self.orchestrator.state_dir / file_db_name) + self._root = root or Path.home() + + def _read_path(self, path: Path | str) -> Path: + return self._root / path + + def add(self, path: Path | str, content: bytes) -> None: + path = self._read_path(path) + self._files[path_to_str(path)] = content + + def realize(self) -> None: + for path, content in sorted(self._files.items()): + self._write_file(self._read_path(path), content) + + for path in self._file_db.paths(): + if path not in self._files: + self._remove_file(self._read_path(path)) + + def _write_file(self, path: Path, content: bytes) -> None: + cur_hash = hash_file(path) + target_hash = hash_data(content) + if cur_hash == target_hash: + return + + relative_path = path.relative_to(self._root, walk_up=True) + self.c.print(f"[bold green]+[/] {escape(str(relative_path))}", highlight=False) + + if reason := self._file_db.verify_hash(path, cur_hash): + self.c.print(f"[red]Error:[/] {escape(reason)}", highlight=False) + + # We want to avoid scenarios where we fail to remember a file we've + # written. It is better to remember a file with an incorrect hash than + # to forget it entirely. Thus, we must always update the file db before + # we write a file. + self._file_db.add_hash(path, target_hash) + atomic_write(path, content) + + def _remove_file(self, path: Path) -> None: + relative_path = path.relative_to(self._root, walk_up=True) + self.c.print(f"[bold red]-[/] {escape(str(relative_path))}", highlight=False) + + cur_hash = hash_file(path) + if reason := self._file_db.verify_hash(path, cur_hash): + self.c.print(f"[red]Error:[/] {escape(reason)}", highlight=False) + + try: + path.unlink() + except FileNotFoundError: + pass + for parent in path.parents: + try: + parent.rmdir() + except: + break + + # We want to avoid scenarios where we forget a file without actually + # removing it. Thus, the db must be updated after the removal. + self._file_db.remove_hash(path) diff --git a/pasch/modules/pacman.py b/pasch/modules/pacman.py index 1364c4a..bdfc768 100644 --- a/pasch/modules/pacman.py +++ b/pasch/modules/pacman.py @@ -1,7 +1,6 @@ from dataclasses import dataclass, field from subprocess import CalledProcessError -from rich import print from rich.markup import escape from pasch.cmd import run_capture, run_execute @@ -37,9 +36,9 @@ class Pacman(Module): to_uninstall = installed - target for package in sorted(to_install): - print(f"[bold green]+[/] {escape(package)}") + self.c.print(f"[bold green]+[/] {escape(package)}") for package in sorted(to_uninstall): - print(f"[bold red]-[/] {escape(package)}") + self.c.print(f"[bold red]-[/] {escape(package)}") self._install_packages(to_install) self._uninstall_packages(to_uninstall) diff --git a/pasch/orchestrator.py b/pasch/orchestrator.py index 1a417e0..f3514d4 100644 --- a/pasch/orchestrator.py +++ b/pasch/orchestrator.py @@ -3,21 +3,27 @@ from __future__ import annotations from abc import ABC, abstractmethod from rich import print +from rich.console import Console from rich.markup import escape +from xdg_base_dirs import xdg_state_home class Module(ABC): def __init__(self, orchestrator: Orchestrator) -> None: self.orchestrator = orchestrator self.orchestrator.register(self) + self.c = self.orchestrator.console @abstractmethod def realize(self) -> None: ... class Orchestrator: - def __init__(self, dry_run: bool = False) -> None: + def __init__(self, name: str = "pasch", dry_run: bool = False) -> None: + self.name = name self.dry_run = dry_run + self.state_dir = xdg_state_home() / self.name + self.console = Console(highlight=False) self._frozen: bool = False self._modules: list[Module] = [] diff --git a/pyproject.toml b/pyproject.toml index 8a4b2a8..4613a70 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ version = "0.0.0" description = "Python-based Arch System Config Helper" readme = "README.md" requires-python = ">=3.13" -dependencies = ["rich>=14.1.0"] +dependencies = ["rich>=14.1.0", "xdg-base-dirs>=6.0.2"] [tool.ruff.lint] select = ["RUF", "F"] diff --git a/uv.lock b/uv.lock index b87e4de..77ea825 100644 --- a/uv.lock +++ b/uv.lock @@ -29,10 +29,14 @@ version = "0.0.0" source = { virtual = "." } dependencies = [ { name = "rich" }, + { name = "xdg-base-dirs" }, ] [package.metadata] -requires-dist = [{ name = "rich", specifier = ">=14.1.0" }] +requires-dist = [ + { name = "rich", specifier = ">=14.1.0" }, + { name = "xdg-base-dirs", specifier = ">=6.0.2" }, +] [[package]] name = "pygments" @@ -55,3 +59,12 @@ sdist = { url = "https://files.pythonhosted.org/packages/fe/75/af448d8e52bf1d8fa wheels = [ { url = "https://files.pythonhosted.org/packages/e3/30/3c4d035596d3cf444529e0b2953ad0466f6049528a879d27534700580395/rich-14.1.0-py3-none-any.whl", hash = "sha256:536f5f1785986d6dbdea3c75205c473f970777b4a0d6c6dd1b696aa05a3fa04f", size = 243368, upload-time = "2025-07-25T07:32:56.73Z" }, ] + +[[package]] +name = "xdg-base-dirs" +version = "6.0.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/bf/d0/bbe05a15347538aaf9fa5b51ac3b97075dfb834931fcb77d81fbdb69e8f6/xdg_base_dirs-6.0.2.tar.gz", hash = "sha256:950504e14d27cf3c9cb37744680a43bf0ac42efefc4ef4acf98dc736cab2bced", size = 4085, upload-time = "2024-10-19T14:35:08.114Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fc/03/030b47fd46b60fc87af548e57ff59c2ca84b2a1dadbe721bb0ce33896b2e/xdg_base_dirs-6.0.2-py3-none-any.whl", hash = "sha256:3c01d1b758ed4ace150ac960ac0bd13ce4542b9e2cdf01312dcda5012cfebabe", size = 4747, upload-time = "2024-10-19T14:35:05.931Z" }, +]