diff --git a/pyproject.toml b/pyproject.toml index 3003d9bbbe..370f964481 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,7 @@ dependencies = [ llfuse = ["llfuse >= 1.3.8"] pyfuse3 = ["pyfuse3 >= 3.1.1"] nofuse = [] +pydantic = ["pydantic >= 2.8.2"] [project.urls] "Homepage" = "https://borgbackup.org/" diff --git a/src/borg/public/__init__.py b/src/borg/public/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/borg/public/cli_api/__init__.py b/src/borg/public/cli_api/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/borg/public/cli_api/v1.py b/src/borg/public/cli_api/v1.py new file mode 100644 index 0000000000..6ce1be4077 --- /dev/null +++ b/src/borg/public/cli_api/v1.py @@ -0,0 +1,169 @@ +"""Pydantic models that can parse borg 1.x's CLI output. + +The two top-level models are: + +- `BorgLogLine`, which parses any line of borg's logging output, +- all `Borg*Result` classes, which parse the final JSON output of some borg commands. + +The different types of log lines are defined in the other models. +""" + +import json +import logging +import typing +from datetime import datetime +from pathlib import Path + +import pydantic + +_log = logging.getLogger(__name__) + + +class BaseBorgLogLine(pydantic.BaseModel): + def get_level(self) -> int: + """Get the log level for this line as a `logging` level value. + + If this is a log message with a levelname, use it. + Otherwise, progress messages get `DEBUG` level, and other messages get `INFO`. + """ + return logging.DEBUG + + +class ArchiveProgressLogLine(BaseBorgLogLine): + original_size: int + compressed_size: int + deduplicated_size: int + nfiles: int + path: Path + time: float + + +class FinishedArchiveProgress(BaseBorgLogLine): + """JSON object printed on stdout when an archive is finished.""" + + time: float + type: typing.Literal["archive_progress"] + finished: bool + + +class ProgressMessage(BaseBorgLogLine): + operation: int + msgid: typing.Optional[str] + finished: bool + message: typing.Optional[str] + time: float + + +class ProgressPercent(BaseBorgLogLine): + operation: int + msgid: str | None = pydantic.Field(None) + finished: bool + message: str | None = pydantic.Field(None) + current: float | None = pydantic.Field(None) + info: list[str] | None = pydantic.Field(None) + total: float | None = pydantic.Field(None) + time: float + + @pydantic.model_validator(mode="after") + def fields_depending_on_finished(self) -> typing.Self: + if self.finished: + if self.message is not None: + raise ValueError("message must be None if finished is True") + if self.current != self.total: + raise ValueError("current must be equal to total if finished is True") + if self.info is not None: + raise ValueError("info must be None if finished is True") + if self.total is not None: + raise ValueError("total must be None if finished is True") + else: + if self.message is None: + raise ValueError("message must not be None if finished is False") + if self.current is None: + raise ValueError("current must not be None if finished is False") + if self.info is None: + raise ValueError("info must not be None if finished is False") + if self.total is None: + raise ValueError("total must not be None if finished is False") + return self + + +class FileStatus(BaseBorgLogLine): + status: str + path: Path + + +class LogMessage(BaseBorgLogLine): + time: float + levelname: typing.Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] + name: str + message: str + msgid: typing.Optional[str] + + def get_level(self) -> int: + try: + return getattr(logging, self.levelname) + except AttributeError: + _log.warning( + "could not find log level %s, giving the following message WARNING level: %s", + self.levelname, + json.dumps(self), + ) + return logging.WARNING + + +_BorgLogLinePossibleTypes = ( + ArchiveProgressLogLine | FinishedArchiveProgress | ProgressMessage | ProgressPercent | FileStatus | LogMessage +) + + +class BorgLogLine(pydantic.RootModel[_BorgLogLinePossibleTypes]): + """A log line from Borg with the `--log-json` argument.""" + + def get_level(self) -> int: + return self.root.get_level() + + +class _BorgArchive(pydantic.BaseModel): + """Basic archive attributes.""" + + name: str + id: str + start: datetime + + +class _BorgArchiveStatistics(pydantic.BaseModel): + """Statistics of an archive.""" + + original_size: int + compressed_size: int + deduplicated_size: int + nfiles: int + + +class _BorgLimitUsage(pydantic.BaseModel): + """Usage of borg limits by an archive.""" + + max_archive_size: float + + +class _BorgDetailedArchive(_BorgArchive): + """Archive attributes, as printed by `json info` or `json create`.""" + + end: datetime + duration: float + stats: _BorgArchiveStatistics + limits: _BorgLimitUsage + command_line: typing.List[str] + chunker_params: typing.Any | None = None + + +class BorgCreateResult(pydantic.BaseModel): + """JSON object printed at the end of `borg create`.""" + + archive: _BorgDetailedArchive + + +class BorgListResult(pydantic.BaseModel): + """JSON object printed at the end of `borg list`.""" + + archives: typing.List[_BorgArchive]