diff --git a/README.md b/README.md index ba30ac2..a90e52a 100644 --- a/README.md +++ b/README.md @@ -12,17 +12,14 @@ Early alpha. Works on my machines. Python 3.14 is required. You can get it using [uv](https://github.com/astral-sh/uv). -Also you will need one of backup backends. -[Restic](https://github.com/restic/restic) and [Rustic](https://github.com/rustic-rs/rustic) are supported. +Also you will need [Rustic](https://github.com/rustic-rs/rustic) as a backup backend. v0.11.4 is fully compatible. ```bash -apt update && apt install -y restic -# or download rustic binary from https://github.com/rustic-rs/rustic/releases/tag/v0.11.3 +pacman -S rustic +# or get the binary binary from https://github.com/rustic-rs/rustic/releases/tag/v0.11.4 -# get using uv +# and install lohup using uv uv tool install git+https://github.com/kam1sh/lohup -# or by pip -pip install ``` ## Usage @@ -41,8 +38,8 @@ lohup snapshots --repo cloud ## Core features * Configuration in TOML -* Restic/Rustic as a backup driver -* Backup hooks, such as create/remove btrfs filesystem snapshot +* Local database for snapshot information +* Backup hooks, such as create/remove btrfs snapshot before backup * Environment variables ### Features in TODO @@ -53,6 +50,7 @@ lohup snapshots --repo cloud * Metrics and OpenTelemetry support * Binary releases, win+linux amd64 along with Python wheel * Include files in profiles with `paths-from` +* Pluggable engines ## Development @@ -66,4 +64,4 @@ uv run pytest --tb=short uv run pyinstaller -F lohup-cli.spec ``` -[1] in my native language loh (лох) means "dummy" or "loser". So, yeah, backups are for losers. +[1] in Russian language loh (лох) means "dummy" or "loser". So, yeah, backups are for losers. diff --git a/src/lohup/app.py b/src/lohup/app.py index 40c560e..9b46fca 100644 --- a/src/lohup/app.py +++ b/src/lohup/app.py @@ -1,44 +1,41 @@ -from lohup.engine import Snapshot -from lohup.storage.storage import Storage -from lohup.config import LocalRepository, S3Repository +import hashlib import subprocess as procs +from collections import defaultdict from lohup import config as tomlconf +from lohup.config import LocalRepository, S3Repository +from lohup.engine import Snapshot, SnapshotInfo from lohup.logger import BasicLogger, LoggerProto, LogLevel -from lohup.restic import Restic from lohup.rustic import Rustic +from lohup.storage import refresher +from lohup.storage.file import SnapshotFile +from lohup.storage.snapshot import SnapshotRecord +from lohup.storage.storage import Storage class Lohup: - subsystem: str | None _config: tomlconf.TomlConfig | None + _storage: Storage | None - def __init__(self, config_path: str | None, logger: LoggerProto | None = None, cache: bool = False): + def __init__(self, config_path: str | None, logger: LoggerProto | None = None): self._config_path = config_path or "lohup.toml" self._config = None - self.subsystem = None - self._cache = cache + self._storage = None self.log = logger or BasicLogger(level=LogLevel.INFO) def load(self): self._config = tomlconf.TomlConfig.from_file(self._config_path) - self.subsystem = self.config.settings.subsystem - if self.subsystem not in ("restic", "rustic"): - raise KeyError(f"Invalid subsystem: {self.subsystem}") - if self._cache: - pth = self.config.settings.cache_dir - if not pth.exists(): - pth.mkdir(parents=True) - pth.joinpath("CACHEDIR.TAG").write_text( - "Signature: 8a477f597d28d172789f06886806bc55" - ) - self._cache = Storage.from_file(pth / "db.sqlite3") + pth = self.config.settings.data_dir + if not pth.exists(): + pth.mkdir(parents=True) + name = f"{self._checksum()}.db" + self._storage = Storage.from_file(pth / name) - def invoke_direct(self, repo: str | None, args: tuple[str, ...]): + def invoke_direct(self, repo: str | None, args: list[str]): spec = self._repo_by_name(repo) engine = self._engine_for(spec) with engine: - engine.run(args) + engine.call(args) def _exechooks(self, hooks: list): for h in hooks: @@ -77,19 +74,12 @@ class Lohup: raise KeyError(f"No repo attached to profile: {profile.name!r}") return default - def _engine_for(self, repo: tomlconf.Repository) -> Restic | Rustic: - match self.subsystem: - case "rustic": - return Rustic( - repo, - log=self.log, - conf_dir=self.config.settings.build_dir, - storage=self.storage, - ) - case "restic": - return Restic(repo, log=self.log) - case x: - raise KeyError(f"Unknown subsystem: {x}") + def _engine_for(self, repo: tomlconf.Repository) -> Rustic: + return Rustic( + repo, + log=self.log, + conf_dir=self.config.settings.build_dir, + ) def backup(self, profile: str): spec = self._profile_for(profile) @@ -115,48 +105,116 @@ class Lohup: self._exechooks(self.config.hooks.after_all) self.log.info("Finished!") - def snapshots(self, repo: str | None, profile: str | None = None): + def snapshots( + self, repo: str | None, profile: str | None = None + ) -> dict[str, list[Snapshot]]: spec = self._repo_by_name(repo) - engine = self._engine_for(spec) - with engine: - result = engine.snapshots(profile=profile) - if cache := self.storage: - total = [] - existing = {x.id for x in total} - for arr in result.values(): - total.extend(arr) - with cache.session() as session: - print(len(Snapshot.findall(session))) - response = Snapshot.update_all(session, repo=spec.name, items=total) - session.commit() - if not response["untouched"]: - return result - to_delete = existing.difference(response["untouched"]) - if not to_delete: - return result - print(response) - with cache.session() as session: - Snapshot.delete_all(session, repo=spec.name, ids=list(to_delete), profile=profile) + if self.storage.is_empty(): + raise ValueError( + "Local database is empty. Update it using 'fetch' subcommand." + ) + with self.storage.session() as session: + records = SnapshotRecord.all_by_repository(session, spec.name) + if profile is not None: + records = filter(lambda r: r.profile == profile, records) + items = [Snapshot.from_record(x) for x in records] + result = defaultdict(list) + for item in items: + result[item.profile].append(item) + for name in result: + result[name].sort(key=lambda x: x.time) return result - def profile_info(self, repo: str | None, profile: str, offset: int): + def profile_info( + self, repo: str | None, profile: str, offset: int, depth: int + ) -> Snapshot: spec = self._repo_by_name(repo) - engine = self._engine_for(spec) - with engine: - all_snapshots = engine.snapshots(profile=profile) - if not isinstance(all_snapshots, dict): - raise ValueError("internal error - unknown snapshots format") - snapshots = all_snapshots[profile] + with self.storage.session() as session: + snapshots = SnapshotRecord.all_by_profile( + session, repo=spec.name, profile=profile + ) + if offset >= len(snapshots): + raise IndexError( + f"Requested offset {offset} exceeds snapshots count ({len(snapshots)})" + ) snapshot = snapshots[offset] - info = engine.info(snapshot.id) - return info + items = SnapshotFile.find_all( + session, repo=spec, snapshot=snapshot, depth=depth + ) + info = SnapshotInfo.from_db(items) + result = Snapshot.from_record(snapshot) + if info is None: + raise KeyError("Internal error - snapshot has no files.") + result.info = info + return result - def snapshot_info(self, repo: str, ref: str): + def snapshot_info(self, repo: str, ref: str, depth: int) -> Snapshot: spec = self._repo_by_name(repo) - engine = self._engine_for(spec) - with engine: - info = engine.info(ref) - return info + with self.storage.session() as session: + snapshot = SnapshotRecord.find_by_key(session, repo, key=ref) + if snapshot is None: + raise KeyError("No such snapshot.") + files = SnapshotFile.find_all( + session, repo=spec, snapshot=snapshot, depth=depth + ) + info = SnapshotInfo.from_db(files) + result = Snapshot.from_record(snapshot) + if info is None: + raise KeyError("Internal error - snapshot has no files.") + result.info = info + return result + + def refresh(self, clean: bool, force: bool = False): + if clean: + self.storage.recreate() + first = list(self.config.repos.values())[0] + rustic = self._engine_for(first) + ref = refresher.Refresher(storage=self.storage, config=self.config, force=force) + repos = ref.update_repos(self.config.repos) + # repo name -> json from rustic + infos = {} + for name in repos.added: + rustic.repo = self.config.repos[name] + with rustic: + data = rustic.repoinfo() + if data is None: + repos.added.remove(name) + repos.message(f"Repository {name} does not exist, skipping") + continue + infos[name] = data + ref.add_repos(infos) + pulled = [] + for repo in self.config.repos.values(): + rustic.repo = repo + with rustic: + profiles = rustic.snapshots() + for snapshots in profiles.values(): + pulled.extend(snapshots) + snapshots_result = ref.update_snapshots(pulled) + to_check = set(snapshots_result.added) + to_check.update(snapshots_result.updated) + by_repo = defaultdict(list) + for key in to_check: + snapshot = snapshots_result.items[key] + by_repo[snapshot.repo].append(snapshot) + # key id -> record + snapshots: dict[str, SnapshotRecord] = {} + with self.storage.session() as session: + for repo in by_repo: + for row in SnapshotRecord.all_by_repository(session, repo): + snapshots[row.key] = row + # snapshot id -> snapshot info + for key in to_check: + item = snapshots_result.items[key] + snapshot = snapshots[key] + rustic.repo = self.config.repos[item.repo] + with rustic: + self.log.info( + f"Fetching file tree of snapshot {key} (repo={item.repo})" + ) + info = rustic.info(key) + ref.add_files(info=info, snapshot_id=snapshot.id) + return repos, snapshots_result def _invoke_profile(self, restic, profile: tomlconf.Profile): with restic as engine: @@ -186,9 +244,12 @@ class Lohup: return self._config @property - def storage(self) -> Storage | None: - if self._cache is None: + def storage(self) -> Storage: + if not self._storage: raise ValueError("Storage is not available.") - if isinstance(self._cache, bool): - return None - return self._cache + return self._storage + + def _checksum(self): + key = str(self.config.path.absolute()) + hashsum = hashlib.sha3_256(key.encode(), usedforsecurity=False) + return hashsum.hexdigest()[:12] diff --git a/src/lohup/cli.py b/src/lohup/cli.py index 7a7d0de..28342f7 100644 --- a/src/lohup/cli.py +++ b/src/lohup/cli.py @@ -1,5 +1,5 @@ import re -from typing import no_type_check +from typing import no_type_check, cast import click import humanize @@ -11,11 +11,10 @@ from lohup.config import ConfigError @click.group() @click.option("--config", envvar="LOHUP_CONFIG", default="lohup.toml") -@click.option("--cache", type=bool, default=True) @click.pass_context -def cli(ctx, config, cache): +def cli(ctx, config): log = logger.CliLogger(level=logger.LogLevel.DEBUG) - ctx.obj = Lohup(config_path=config, logger=log, cache=cache) + ctx.obj = Lohup(config_path=config, logger=log) try: ctx.obj.load() except ConfigError as e: @@ -31,7 +30,7 @@ def restic(obj: Lohup, repo: str | None, args: tuple[str, ...]): """ Pass command to restic """ - obj.invoke_direct(repo, args) + obj.invoke_direct(repo, list(args)) @cli.command() @@ -66,24 +65,9 @@ def snapshots(obj: Lohup, repo: str | None, profile: str | None): items = raw_items.copy() items.sort(key=lambda x: x.time, reverse=True) for snap in items: - click.echo(f"\tSnapshot {snap.id}:") - started = humanize.naturaltime(snap.time) - text = click.style(started, fg="blue") - took = humanize.naturaldelta(snap.duration) - click.echo(f"\t\tStarted: {text} (took {took})") - ratio = snap.changes_ratio * 100 - ratio = click.style(f"{ratio:.1f}%", fg="blue") - click.echo( - f"\t\tFiles changed (since previous): {snap.total_changes} ({ratio})" - ) - uncompressed = humanize.naturalsize(snap.size, binary=True) - compressed = humanize.naturalsize(snap.size_compressed, binary=True) - total_bytes = snap.bytes_read - ratio = snap.size / total_bytes if total_bytes else 0 - ratio = click.style(f"{ratio:.1f}%", fg="blue") - click.echo(f"\t\tDiff size: {compressed=!s}, {uncompressed=!s} ({ratio})") - click.echo(f"\t\tHost: {snap.hostname}") - click.echo() + lines = _write_snapshot(snap) + click.echo(f"\tSnapshot {snap.id}:\n\t\t", nl=False) + click.echo("\n\t\t".join(lines)) @cli.command() @@ -110,14 +94,50 @@ def info(obj: Lohup, repo: str | None, depth: int, ref: str): offset = 0 if (value := match.group(2)) is not None: offset = int(value.removeprefix("^")) - info = obj.profile_info(repo, profile=profile, offset=offset) + snapshot = obj.profile_info(repo, profile=profile, offset=offset, depth=depth) elif repo: - info = obj.snapshot_info(repo=repo, ref=ref) + snapshot = obj.snapshot_info(repo=repo, ref=ref, depth=depth) else: raise click.ClickException("Requested info by ID without repository name.") + info = cast(engine.SnapshotInfo, snapshot.info) info.build_tree(depth=depth) + lines = _write_snapshot(snapshot) + click.echo(f"Snapshot {snapshot.id}") + click.echo("\n".join(lines)) for line in info.show(depth=depth): click.echo(line) +@cli.command() +@click.option("--clean", is_flag=True, help="Refresh database from scratch") +@click.option("--force", is_flag=True, help="Ignore items TTL") +@click.pass_obj +def update(obj: Lohup, clean: bool, force: bool): + repos, snapshots = obj.refresh(clean=clean, force=force) + click.echo("Repository updates:") + click.echo(repos.text()) + click.echo("Snapshot updates:") + click.echo(snapshots.text()) + + +def _write_snapshot(snap: engine.Snapshot): + lines = [] + started = click.style(humanize.naturaltime(snap.time), fg="blue") + took = humanize.naturaldelta(snap.duration) + lines.append(f"Started: {started} (took {took})") + ratio = snap.changes_ratio * 100 + ratio = click.style(f"{ratio:.1f}%", fg="blue") + lines.append( + f"Files changed (since previous): {snap.total_changes} ({ratio})" + ) + uncompressed = humanize.naturalsize(snap.size, binary=True) + compressed = humanize.naturalsize(snap.size_compressed, binary=True) + total = humanize.naturalsize(snap.bytes_read) + ratio = snap.size / snap.bytes_read if snap.bytes_read else 0 + ratio = click.style(f"{ratio:.1f}%", fg="blue") + lines.append(f"Size: {total}") + lines.append(f"Diff size: {compressed=!s}, {uncompressed=!s} ({ratio})") + lines.append(f"Host: {snap.hostname}") + return lines + if __name__ == "__main__": cli() diff --git a/src/lohup/config.py b/src/lohup/config.py index eb949d9..9d1deb1 100644 --- a/src/lohup/config.py +++ b/src/lohup/config.py @@ -34,8 +34,7 @@ class Settings: backup_base_dir: Path globalvars: dict[str, str] build_dir: Path - subsystem: str - cache_dir: Path + data_dir: Path @staticmethod def load(conf: dict, validator: DeepValidator): @@ -46,7 +45,7 @@ class Settings: f"backup-base-dir: got ({pth!r}): variables are not allowed" ) pth = Path(pth) - if not pth.is_absolute: + if not pth.is_absolute(): validator.error( f"backup-base-dir: got ({pth!r}): only absolute paths are supported" ) @@ -57,14 +56,13 @@ class Settings: validator.error(f"tmp-dir: got {tmpdir!r}: variables are not allowed") tmp_path = Path(tmpdir) elif platform.system() == "Windows": - tmp_path = Path.home().joinpath("AppData", "Local", "Temp", "lohup") - cache_path = Path.home().joinpath(".cache", "lohup") + tmp_path = Path.home().joinpath("AppData", "Local", "lohup") + data_dir = Path.home().joinpath(".local", "share", "lohup") settings = Settings( backup_base_dir=basepath, globalvars=conf.get("globals") or {}, build_dir=tmp_path, - subsystem=conf.get("subsystem-name", "restic"), - cache_dir=cache_path + data_dir=data_dir ) for key, value in settings.globalvars.items(): with validator.inner(f"variable {key!r}") as inner: @@ -260,6 +258,7 @@ class TomlConfig: hooks: HookSet templater: ValidatedTemplater profiles: dict[str, Profile] + path: Path @staticmethod def from_file(name: str): @@ -349,5 +348,6 @@ class TomlConfig: hooks=hooks, templater=templater, profiles=profiles, + path=path ) return toml diff --git a/src/lohup/engine.py b/src/lohup/engine.py index f1c0375..37f488e 100644 --- a/src/lohup/engine.py +++ b/src/lohup/engine.py @@ -1,37 +1,44 @@ -from lohup.storage.snapshot import SnapshotRecord -from sqlalchemy.orm import Session -import humanize.filesize -from typing_extensions import cast +import json +import typing as ty from collections import defaultdict -from lohup.storage.file import SnapshotFile -from sqlalchemy import select, delete -from lohup.storage.storage import Storage -from pathlib import PosixPath from dataclasses import dataclass, field from datetime import datetime -import typing as ty +from pathlib import PosixPath, Path + +import humanize.filesize +from sqlalchemy import delete, select +from sqlalchemy.orm import Session +from typing_extensions import cast + +from lohup.storage.file import SnapshotFile +from lohup.storage.snapshot import SnapshotRecord @dataclass class Snapshot: id: str + repo: str profile: str tags: list[str] hostname: str time: datetime end: datetime - raw: dict[str, ty.Any] + raw: dict[str, ty.Any] = field(repr=False) parent: str | None _engine: str + info: SnapshotInfo | None = field(default=None) @staticmethod - def from_rustic(item: dict[str, ty.Any], group: dict[str, ty.Any]) -> Snapshot: + def from_rustic( + item: dict[str, ty.Any], group: dict[str, ty.Any], repo: str + ) -> Snapshot: profile = "" for tag in group["tags"]: if tag.startswith("lohup.profile="): profile = tag.removeprefix("lohup.profile=") return Snapshot( id=item["id"][:8], + repo=repo, profile=profile, tags=group["tags"], hostname=group["hostname"], @@ -42,44 +49,35 @@ class Snapshot: _engine="rustic", ) - @staticmethod - def from_restic(item: dict[str, ty.Any]) -> Snapshot: - profile = "" - for tag in item["tags"]: - if tag.startswith("lohup.profile="): - profile = tag.removeprefix("lohup.profile=") - return Snapshot( - id=item["short_id"], - profile=profile, - tags=item["tags"], - hostname=item["hostname"], - time=datetime.fromisoformat(item["time"]), - end=datetime.fromisoformat(item["summary"]["backup_end"]), - raw=item, - parent=item.get("parent"), - _engine="restic", - ) - @staticmethod def findall(session: Session): out = [] stmt = select(SnapshotRecord) for row in session.scalars(stmt): - out.append(Snapshot( - id=row.key, - profile=row.profile, - tags=[], - hostname="", - time=row.created, - end=row.created, - raw={}, - parent=row.parent, - _engine="" - )) + out.append(Snapshot.from_record(row)) return out @staticmethod - def update_all(session: Session, repo: str, items: list["Snapshot"]) -> dict[str, list[str]]: + def from_record(row: SnapshotRecord) -> Snapshot: + info = json.loads(row.raw_info) + return Snapshot( + id=row.key, + repo=row.repo.name, + profile=row.profile, + tags=[], + hostname=info["hostname"], + time=row.created, + end=row.created, + raw=info, + parent=row.parent, + _engine="", + # record=row, + ) + + @staticmethod + def update_all( + session: Session, repo: str, items: list["Snapshot"] + ) -> dict[str, list[str]]: stmt = select(SnapshotRecord.key).where(SnapshotRecord.repo == repo) stored = set(session.scalars(stmt)) result = { @@ -89,13 +87,15 @@ class Snapshot: for snapshot in items: if snapshot.id not in stored: result["added"].append(snapshot.id) - session.add(SnapshotRecord( - repo=repo, - key=snapshot.id, - profile=snapshot.profile, - parent=snapshot.parent, - created=snapshot.time, - )) + session.add( + SnapshotRecord( + repo=repo, + key=snapshot.id, + profile=snapshot.profile, + parent=snapshot.parent, + created=snapshot.time, + ) + ) else: result["untouched"].append(snapshot.id) return result @@ -147,7 +147,6 @@ class Snapshot: def duration(self): return self.end - self.time - @dataclass(slots=True) class FileItem: path: str @@ -181,6 +180,7 @@ class Directory: def flags(self): return "d" + T = ty.TypeVar("T") @@ -202,15 +202,19 @@ class SnapshotInfo: root: Directory | None = field(default=None) @staticmethod - def load_db( - storage: Storage, repo: str, snapshot: str - ) -> "SnapshotInfo" | None: + def from_db(items: list[SnapshotFile]) -> "SnapshotInfo" | None: info = SnapshotInfo() - with storage.session() as session: - query = select(SnapshotFile).filter_by(repo=repo, snapshot=snapshot) - result = session.scalars(query) - for row in result: - parent = PosixPath(row.path).parent + for row in items: + parent = PosixPath(row.path).parent + if row.kind.startswith("d"): + item = Directory( + path=row.path, + parent=str(parent), + name=row.name, + size=row.size, + date=row.mtime, + ) + else: item = FileItem( path=row.path, parent=str(parent), @@ -219,59 +223,68 @@ class SnapshotInfo: size=row.size, date=row.mtime, ) - info.tree[item.path] = item - if not info.tree: - return None + info.tree[item.path] = item + if not info.tree: + return None return info - def save(self, storage: Storage, repo: str, snapshot: str): - with storage.session() as session: - exists = session.scalars( - select(SnapshotFile).filter_by(repo=repo, snapshot=snapshot).limit(1) - ) - if exists.first(): - return - for file in self.tree.values(): - session.add( - SnapshotFile( - repo=repo, - snapshot=snapshot, - kind=file.flags, - path=file.path, - name=file.name, - size=file.size, - mtime=file.date, - ) + def save(self, session: Session, snapshot_id: int): + exists = session.scalars( + select(SnapshotFile).filter_by(snapshot_id=snapshot_id).limit(1) + ) + if exists.first(): + return + for file in self.tree.values(): + session.add( + SnapshotFile( + snapshot_id=snapshot_id, + kind=file.flags, + path=file.path, + name=file.name, + size=file.size, + mtime=file.date, + depth=file.depth, ) - session.commit() + ) - def _itertree(self, depth: int) -> ty.Generator[FileItem | Directory]: + def _itertree(self, depth: int | None) -> ty.Generator[FileItem | Directory]: for pth in self.tree: value = self.tree[pth] - if value.depth > depth: + if depth and value.depth > depth: continue yield value - def build_tree(self, depth: int): + def build_tree(self, depth: int | None): root: Directory | None = None + nearest: FileItem | None = None for file in self._itertree(depth): - file = cast(FileItem, file) pathobj = PosixPath(file.path) - if not file.flags.startswith("d"): + if isinstance(file, FileItem): + if nearest is None or not pathobj.is_relative_to(nearest.path): + nearest = file continue - item = Directory( - path=file.path, parent=file.parent, name=file.name, date=file.date - ) - self.tree[file.path] = item if root is None or not pathobj.is_relative_to(root.path): - root = item + root = file if root is None: - raise ValueError("Failed to resolve root directory") + if nearest is None: + raise ValueError("No root or any file to fetch.") + parent = Path(nearest.parent) + root = Directory( + path=nearest.parent, + parent=str(parent.parent), + name=parent.name, + date=nearest.date, + ) + self.tree.setdefault(root.path, root) for file in self._itertree(depth): if isinstance(file, FileItem): self.tree[file.parent].children.append(file) # ty: ignore continue - file.size = sum(x.size for x in self.tree.values() if x.path.startswith(f"{file.path}/")) + if file.size != 0: + continue + file.size = sum( + x.size for x in self.tree.values() if x.path.startswith(f"{file.path}/") + ) for file in self._itertree(depth): if parent := self.tree.get(file.parent): parent.children.append(file) # ty: ignore @@ -308,7 +321,7 @@ class SnapshotInfo: file = current.value prefix = " " * (file.depth - 1) line = f"{file.name}/" if isinstance(file, Directory) else file.name - size = "0" + size = "0 Bytes" if file.size: size = humanize.filesize.naturalsize(file.size) lines.append(f"{size:>10} {prefix}{line}") diff --git a/src/lohup/restic.py b/src/lohup/restic.py deleted file mode 100644 index b419299..0000000 --- a/src/lohup/restic.py +++ /dev/null @@ -1,98 +0,0 @@ -import json -import os -import subprocess as procs -from collections import defaultdict -from dataclasses import dataclass, field -from pathlib import Path - -from lohup import config, engine -from lohup.logger import BasicLogger, CliLogger - - -@dataclass -class Restic: - repo: config.Repository - log: CliLogger | BasicLogger - binary: str = field(default="restic") - - def environ(self): - env = os.environ.copy() - env["RESTIC_PASSWORD_FILE"] = self.repo.repo_key_file.value - match self.repo: - case config.S3Repository(): - if value := self.repo.region: - env["AWS_DEFAULT_REGION"] = value - if masked := self.repo.access_key_file: - env["AWS_ACCESS_KEY_ID"] = Path(masked.value).read_text().strip() - if masked := self.repo.secret_key_file: - env["AWS_SECRET_ACCESS_KEY"] = ( - Path(masked.value).read_text().strip() - ) - path = self.repo.path.lstrip("/") - url = f"{self.repo.endpoint}/{self.repo.bucket}/{path}" - env["RESTIC_REPOSITORY"] = f"s3:{url}" - case config.LocalRepository(): - env["RESTIC_REPOSITORY"] = self.repo.path - return env - - def run(self, args): - cmd, env = self._prepare(args) - procs.check_call(cmd, env=env) - - def backup(self, profile: config.Profile): - args = ["backup", "--tag", profile.name] - args.extend(profile.cli_args) - match profile: - case config.PathsProfile(): - for pth in profile.exclude_paths: - args.extend(["-e", pth]) - args.extend(profile.paths) - self.run(args) - case config.CommandProfile(): - args.append("--stdin") - match profile.command: - case str(x): - self.pipe_stdout(args, src_cmd=x.split()) - case list(x): - self.pipe_stdout(args, src_cmd=x) - case _: - raise NotImplementedError(profile.command) - - def pipe_stdout(self, args: list[str], src_cmd: list[str]): - cmd = [self.binary] + args - env = self.environ() - with procs.Popen(cmd, stdin=procs.PIPE, env=env) as restic: - procs.check_call(src_cmd, stdout=restic.stdin) - - def snapshots(self, format="text", profile: str | None = None): - cmd, env = self._prepare(["snapshots", "--compact"]) - if format == "json": - cmd.append("--json") - if profile: - cmd.extend(["--tag", f"lohup.profile={profile}"]) - result = procs.check_output(cmd, env=env, encoding="utf-8") - if format == "json": - content = json.loads(result) - snapshots = defaultdict(list) - for x in content: - snap = engine.Snapshot.from_restic(x) - snapshots[snap.name].append(snap) - return snapshots - return result - - def info(self, ref: str) -> engine.SnapshotInfo: - raise NotImplementedError - - def _prepare(self, args: list[str]): - if self.repo is None: - raise ValueError("repo not set") - cmd = [self.binary] - cmd.extend(args) - env = self.environ() - return cmd, env - - def __enter__(self): - return self - - def __exit__(self, type, value, traceback): - return diff --git a/src/lohup/rustic.py b/src/lohup/rustic.py index c7789a3..0684159 100644 --- a/src/lohup/rustic.py +++ b/src/lohup/rustic.py @@ -1,13 +1,12 @@ -from lohup.storage.storage import Storage -from typing import cast, IO -import traceback -from datetime import datetime -import re import json +import re import subprocess as procs +import traceback from collections import defaultdict from dataclasses import dataclass, field +from datetime import datetime from pathlib import Path, PosixPath +from typing import IO, Any, cast import tomlkit @@ -20,7 +19,6 @@ class Rustic: repo: config.Repository log: LoggerProto conf_dir: Path - storage: Storage | None binary: str = field(default="rustic") @property @@ -56,7 +54,14 @@ class Rustic: out = [self.binary, "-P", str(self.conf_dir / "rustic")] return out - def run(self, args): + def run(self, args: list[str], combined_output=True) -> procs.CompletedProcess[str]: + cmd = self._cmdline() + cmd.extend(args) + stderr = procs.STDOUT if combined_output else procs.PIPE + proc = procs.run(cmd, stdout=procs.PIPE, stderr=stderr, text=True) + return proc + + def call(self, args: list[str]): cmd = self._cmdline() cmd.extend(args) procs.check_call(cmd) @@ -75,7 +80,7 @@ class Rustic: for pth in profile.exclude_paths: args.extend(["--glob", f"!{pth}"]) args.extend(profile.paths) - self.run(args) + self.call(args) case config.CommandProfile(): args.append("-") match profile.command: @@ -104,17 +109,12 @@ class Rustic: for group in content: key = group["group_key"] for x in group["snapshots"]: - snap = engine.Snapshot.from_rustic(x, group=key) + snap = engine.Snapshot.from_rustic(x, group=key, repo=self.repo.name) grouped[snap.name].append(snap) snapshots.append(snap) return grouped - def info(self, snapshot: str): - if self.storage: - if value := engine.SnapshotInfo.load_db( - self.storage, repo=self.repo.name, snapshot=snapshot - ): - return value + def info(self, snapshot: str) -> engine.SnapshotInfo: result = engine.SnapshotInfo() def read_(match): @@ -126,7 +126,7 @@ class Rustic: mtime = datetime.strptime(mtime_raw, "%d %b %Y %H:%M") path = match.group("path") pth = PosixPath(path) - result.tree[path] = engine.FileItem( + item = engine.FileItem( path=path, parent=str(pth.parent), name=pth.name, @@ -134,6 +134,14 @@ class Rustic: size=int(size), date=mtime, ) + if flags.startswith("d"): + item = engine.Directory( + path=path, + parent=item.parent, + name=pth.name, + date=mtime, + ) + result.tree[path] = item cmd = self._cmdline() cmd.extend(["ls", "-l", snapshot]) @@ -147,7 +155,7 @@ class Rustic: for line in stream: match = pattern.match(line.strip()) if not match: - print(f"unmatched line: {line!s}") + self.log.warning(f"unmatched line: {line!s}") continue try: read_(match) @@ -155,10 +163,19 @@ class Rustic: traceback.print_exc() if not result.tree: raise KeyError("No such snapshot") - if self.storage: - result.save(self.storage, repo=self.repo.name, snapshot=snapshot) + result.build_tree(depth=None) return result + def repoinfo(self) -> dict[str, Any] | None: + proc = self.run(["repoinfo", "--json"], combined_output=False) + if proc.returncode == 0: + return json.loads(proc.stdout) + if "No repository config file found" in proc.stderr: + return None + exc = ValueError("Process returned error") + exc.add_note(f"Output: {proc.stdout}{proc.stderr}") + raise exc + def __enter__(self): self.write_config() return self diff --git a/src/lohup/storage/file.py b/src/lohup/storage/file.py index 2a9b6df..94575fb 100644 --- a/src/lohup/storage/file.py +++ b/src/lohup/storage/file.py @@ -1,18 +1,59 @@ from datetime import datetime -from sqlalchemy import String, Text, BigInteger, DateTime -from sqlalchemy.orm import Mapped, mapped_column -from typing import final +from typing import TYPE_CHECKING, final + +from sqlalchemy import ( + BigInteger, + DateTime, + ForeignKey, + Integer, + String, + Text, + select, +) +from sqlalchemy.orm import Mapped, Session, mapped_column, relationship + +from lohup.config import Repository from lohup.storage.base import Base +if TYPE_CHECKING: + from lohup.storage.snapshot import SnapshotRecord + @final class SnapshotFile(Base): __tablename__ = "files" id: Mapped[int] = mapped_column(primary_key=True) - repo: Mapped[str] = mapped_column(String(64)) - snapshot: Mapped[str] = mapped_column(String(64)) + snapshot_id: Mapped[int] = mapped_column(ForeignKey("snapshots.id")) + snapshot: Mapped["SnapshotRecord"] = relationship() + # file|dir kind: Mapped[str] = mapped_column(String(10)) path: Mapped[str] = mapped_column(Text()) + # how deep file is in the snapshot tree + depth: Mapped[int] = mapped_column(Integer()) name: Mapped[str] = mapped_column(Text()) + # sum of files for directories size: Mapped[int] = mapped_column(BigInteger()) mtime: Mapped[datetime] = mapped_column(DateTime()) + + @staticmethod + def find_all( + session: Session, + repo: Repository, + snapshot: "SnapshotRecord", + depth: int | None = None, + ) -> list["SnapshotFile"]: + from lohup.storage.repository import RepositoryRecord + from lohup.storage.snapshot import SnapshotRecord + + stmt = ( + select(SnapshotFile) + .join(SnapshotFile.snapshot) + .join(SnapshotRecord.repo) + .where( + SnapshotRecord.id == snapshot.id, + RepositoryRecord.name == repo.name + ) + ) + if depth is not None: + stmt = stmt.where(SnapshotFile.depth <= depth) + return list(session.scalars(stmt)) diff --git a/src/lohup/storage/refresher.py b/src/lohup/storage/refresher.py new file mode 100644 index 0000000..f1b6d22 --- /dev/null +++ b/src/lohup/storage/refresher.py @@ -0,0 +1,148 @@ +import json +from lohup.storage.snapshot import SnapshotRecord +from lohup.engine import Snapshot, SnapshotInfo +import typing +import typing as ty +from dataclasses import dataclass, field +from datetime import datetime, timedelta + +from sqlalchemy import select, delete + +from lohup import config +from lohup.storage.repository import RepositoryRecord +from lohup.storage.storage import Storage + +T = typing.TypeVar("T") + + +@dataclass +class UpdateResult[T]: + items: dict[str, T] = field(default_factory=dict) + added: list[str] = field(default_factory=list) + unchanged: list[str] = field(default_factory=list) + updated: list[str] = field(default_factory=list) + removed: list[str] = field(default_factory=list) + _messages: list[str] = field(default_factory=list) + + def message(self, line: str): + self._messages.append(line) + + def text(self) -> str: + lines = [] + if self.added: + lines.append(f"{len(self.added)} added") + if self.updated: + lines.append(f"{len(self.updated)} updated") + if self.removed: + lines.append(f"{len(self.removed)} removed") + if self.unchanged: + lines.append(f"{len(self.unchanged)} untouched") + lines.extend(self._messages) + return "\n".join(lines) + + +@dataclass +class Refresher: + storage: Storage + config: config.TomlConfig + force: bool + + def update_repos(self, repos: dict[str, config.Repository]) -> UpdateResult[str]: + result = UpdateResult[str]() + existing = [] + with self.storage.session() as session: + stmt = select(RepositoryRecord) + rows = session.scalars(stmt) + existing.extend(rows) + existing_names = {x.name for x in existing} + to_remove = [] + for name in repos: + if name not in existing_names: + result.added.append(name) + for row in existing: + delta = datetime.now() - row.updated + if row.name not in repos: + to_remove.append(row.id) + result.removed.append(row.name) + elif self.force or delta > timedelta(hours=8): + result.updated.append(row.name) + else: + result.unchanged.append(row.name) + if to_remove: + stmt = delete(RepositoryRecord).where(RepositoryRecord.id.in_(to_remove)) + with self.storage.session() as session: + session.execute(stmt) + return result + + def add_repos(self, items: dict[str, ty.Any]): + pending = [] + for name, info in items.items(): + data = {x["tpe"]: x for x in info["files"]["repo"]} + size = sum(x["size"] for x in data.values()) + pending.append( + RepositoryRecord( + name=name, size=size, updated=datetime.now() + ) + ) + with self.storage.session() as session: + session.add_all(pending) + session.commit() + + def update_snapshots(self, pulled: list[Snapshot]) -> UpdateResult[Snapshot]: + reponames = {r.repo for r in pulled} + result = UpdateResult[Snapshot]() + result.items = {x.id: x for x in pulled} + records = [] + with self.storage.session() as session: + stmt = ( + select(SnapshotRecord) + .join(SnapshotRecord.repo) + .where(RepositoryRecord.name.in_(reponames)) + ) + rows = session.scalars(stmt) + records.extend(rows) + for row in records: + if row.key not in result.items: + result.items[row.key] = Snapshot.from_record(row) + record_keys = {x.key for x in records} + existing_ids = {x.id for x in pulled} + for snapshot in pulled: + if snapshot.id not in record_keys: + result.added.append(snapshot.id) + rm_ids = [] + for row in records: + if row.key not in existing_ids: + result.removed.append(row.key) + rm_ids.append(row.id) + else: + result.unchanged.append(row.key) + if result.removed: + stmt = delete(SnapshotRecord).where(SnapshotRecord.id.in_(rm_ids)) + with self.storage.session() as session: + session.execute(stmt) + pending = [x for x in result.items.values() if x.id in result.added] + if pending: + self.add_snapshots(pending) + return result + + def add_snapshots(self, items: list[Snapshot]): + stmt = select(RepositoryRecord) + with self.storage.session() as session: + rows = session.scalars(stmt) + repos = {x.name: x for x in rows} + for item in items: + record = SnapshotRecord( + repo_id=repos[item.repo].id, + key=item.id, + profile=item.profile, + parent=item.parent, + created=item.time, + raw_info=json.dumps(item.raw), + ) + session.add(record) + session.commit() + + def add_files(self, snapshot_id: int, info: SnapshotInfo): + with self.storage.session() as session: + info.save(session, snapshot_id) + session.commit() diff --git a/src/lohup/storage/repository.py b/src/lohup/storage/repository.py new file mode 100644 index 0000000..8e9b0d6 --- /dev/null +++ b/src/lohup/storage/repository.py @@ -0,0 +1,20 @@ +from datetime import datetime +from sqlalchemy import String, DateTime, Integer +from sqlalchemy.orm import Mapped, mapped_column, relationship +from lohup.storage.base import Base +from typing import final, TYPE_CHECKING + +if TYPE_CHECKING: + from lohup.storage.snapshot import SnapshotRecord + + +@final +class RepositoryRecord(Base): + __tablename__ = "repositories" + id: Mapped[int] = mapped_column(primary_key=True) + name: Mapped[str] = mapped_column(String(128)) + # total repository size + size: Mapped[int] = mapped_column(Integer()) + updated: Mapped[datetime] = mapped_column(DateTime()) + + snapshots: Mapped[list["SnapshotRecord"]] = relationship() diff --git a/src/lohup/storage/snapshot.py b/src/lohup/storage/snapshot.py index 8fa95ee..dbca55c 100644 --- a/src/lohup/storage/snapshot.py +++ b/src/lohup/storage/snapshot.py @@ -1,17 +1,50 @@ from datetime import datetime -from sqlalchemy import String, DateTime -from sqlalchemy.orm import Mapped, mapped_column from typing import final + +from sqlalchemy import DateTime, ForeignKey, String, select +from sqlalchemy.orm import Mapped, Session, mapped_column, relationship + from lohup.storage.base import Base +from lohup.storage.repository import RepositoryRecord + @final class SnapshotRecord(Base): __tablename__ = "snapshots" id: Mapped[int] = mapped_column(primary_key=True) - repo: Mapped[str] = mapped_column(String(64)) + repo_id: Mapped[int] = mapped_column(ForeignKey("repositories.id")) + repo: Mapped[RepositoryRecord] = relationship(back_populates="snapshots") key: Mapped[str] = mapped_column(String(64)) profile: Mapped[str] = mapped_column(String(32)) parent: Mapped[str] = mapped_column(String(64), nullable=True) + raw_info: Mapped[str] = mapped_column(String()) created: Mapped[datetime] = mapped_column(DateTime()) + @staticmethod + def all_by_repository(session: Session, repo: str) -> list["SnapshotRecord"]: + stmt = ( + select(SnapshotRecord) + .join(SnapshotRecord.repo) + .where(RepositoryRecord.name == repo) + .order_by(SnapshotRecord.created) + ) + return list(session.scalars(stmt)) + @staticmethod + def all_by_profile(session: Session, repo: str, profile: str) -> list["SnapshotRecord"]: + stmt = ( + select(SnapshotRecord) + .join(SnapshotRecord.repo) + .where(RepositoryRecord.name == repo, SnapshotRecord.profile == profile) + .order_by(SnapshotRecord.created) + ) + return list(session.scalars(stmt)) + + @staticmethod + def find_by_key(session: Session, repo: str, key: str) -> "SnapshotRecord" | None: + stmt = ( + select(SnapshotRecord) + .join(SnapshotRecord.repo) + .where(RepositoryRecord.name == repo, SnapshotRecord.key == key) + ) + return session.scalar(stmt) diff --git a/src/lohup/storage/storage.py b/src/lohup/storage/storage.py index ec8388f..6bafb2a 100644 --- a/src/lohup/storage/storage.py +++ b/src/lohup/storage/storage.py @@ -1,10 +1,11 @@ +from lohup.storage.repository import RepositoryRecord from lohup.storage.snapshot import SnapshotRecord import typing as ty from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path -from sqlalchemy import Engine, create_engine +from sqlalchemy import Engine, create_engine, select, func from sqlalchemy.orm import Session from lohup.storage.base import Base @@ -28,11 +29,22 @@ class Storage: with Session(self.engine) as sess: yield sess + def recreate(self): + Base.metadata.drop_all(self.engine) + self.create() + def create(self): Base.metadata.create_all( self.engine, tables=[ # ty: ignore[invalid-argument-type] - SnapshotFile.__table__, + RepositoryRecord.__table__, SnapshotRecord.__table__, + SnapshotFile.__table__, ], ) + + def is_empty(self): + with self.session() as session: + stmt = select(func.count()).select_from(RepositoryRecord) + result = session.scalar(stmt) + return not result