Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add pydantic models for borg 1.x's CLI (#8338) #8343

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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/"
Expand Down
11 changes: 9 additions & 2 deletions src/borg/helpers/progress.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import logging
import json
import time
import typing

from ..logger import create_logger

Expand All @@ -24,9 +25,15 @@ def __init__(self, msgid=None):
self.id = self.operation_id()
self.msgid = msgid

def make_json(self, *, finished=False, **kwargs):
def make_json(self, *, finished=False, override_time: typing.Optional[float] = None, **kwargs):
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let me know if you think this new argument is bad. The function is harder to test when it always takes time.time().

kwargs.update(
dict(operation=self.id, msgid=self.msgid, type=self.JSON_TYPE, finished=finished, time=time.time())
dict(
operation=self.id,
msgid=self.msgid,
type=self.JSON_TYPE,
finished=finished,
time=override_time or time.time(),
)
)
return json.dumps(kwargs)

Expand Down
Empty file added src/borg/public/__init__.py
Empty file.
Empty file.
169 changes: 169 additions & 0 deletions src/borg/public/cli_api/v1.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,169 @@
"""Pydantic models that can parse borg 1.x's JSON 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]
20 changes: 20 additions & 0 deletions src/borg/testsuite/public/cli_api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
import borg.public.cli_api.v1 as v1
from borg.helpers.progress import ProgressIndicatorBase


def test_parse_progress_percent_unfinished():
percent = ProgressIndicatorBase()
override_time = 4567.23
json_output = percent.make_json(finished=False, current=10, override_time=override_time)
assert v1.ProgressPercent.model_validate_json(json_output) == v1.ProgressPercent(
operation=1, msgid=None, finished=False, message=None, current=10, info=None, total=None, time=4567.23
)


def test_parse_progress_percent_finished():
percent = ProgressIndicatorBase()
override_time = 4567.23
json_output = percent.make_json(finished=True, override_time=override_time)
assert v1.ProgressPercent.model_validate_json(json_output) == v1.ProgressPercent(
operation=1, msgid=None, finished=True, message=None, current=None, info=None, total=None, time=override_time
)
Loading