Dataclass

BackupHostPath

BackupHostPath(*segments, connection)

Bases: PurePosixPath

Represents a path for a Connection.

Parameters:

Name	Type	Description	Default
segments	str | PathLike[str]	segments of the path	()
connection	Connection	The connection used for this path.	required

Source code in b4_backup/main/dataclass.py

def __init__(self, *segments: str | PathLike[str], connection: "Connection"):
    """
    Args:
        segments: segments of the path
        connection: The connection used for this path.
    """
    super().__init__(*segments)
    self.connection = connection

exists

exists()

Returns:

Type	Description
bool	True if the location exists.

Source code in b4_backup/main/dataclass.py

def exists(self) -> bool:
    """
    Returns:
        True if the location exists.
    """
    try:
        result = self.connection.run_process(["ls", "-d", str(self)])
    except exceptions.FailedProcessError as e:
        if "No such file or directory" in e.stderr:
            return False

        raise

    return result.strip() != ""

is_dir

is_dir()

Checks if a path is a directory.

Source code in b4_backup/main/dataclass.py

def is_dir(self) -> bool:
    """Checks if a path is a directory."""
    result = self.connection.run_process(["ls", "-dl", str(self)])
    return result.strip()[0] == "d"

iterdir

iterdir()

Returns:

Type	Description
list[BackupHostPath]	A list of Paths containing all items in the current directory.

Source code in b4_backup/main/dataclass.py

def iterdir(self) -> list["BackupHostPath"]:
    """
    Returns:
        A list of Paths containing all items in the current directory.
    """
    result = sorted(self.connection.run_process(["ls", str(self)]).strip().split("\n"))

    if result == [""]:
        return []

    return [self / x for x in result]

mkdir

mkdir(parents=False)

Creates a directory.

Parameters:

Name	Type	Description	Default
parents	bool	Also creates parent directories and doesn't fail if path exist.	False

Source code in b4_backup/main/dataclass.py

def mkdir(self, parents: bool = False) -> None:
    """
    Creates a directory.

    Args:
        parents: Also creates parent directories and doesn't fail if path exist.
    """
    self.connection.run_process(["mkdir", str(self)] + ["-p"] * parents)

rename

rename(target)

Renames/Moves the path to the target location.

Parameters:

Name	Type	Description	Default
target	PurePath	The target location to move the object to.	required

Source code in b4_backup/main/dataclass.py

def rename(self, target: PurePath) -> None:
    """
    Renames/Moves the path to the target location.

    Args:
        target: The target location to move the object to.
    """
    self.connection.run_process(["mv", str(self), str(target)])

rmdir

rmdir()

Removes the given empty directory.

Source code in b4_backup/main/dataclass.py

def rmdir(self) -> None:
    """Removes the given empty directory."""
    try:
        self.connection.run_process(["rmdir", str(self)])
    except exceptions.FailedProcessError as e:
        if "No such file or directory" not in e.stderr:
            raise

ChoiceSelector dataclass

ChoiceSelector(data=list())

Describes a set of data, with dynamic choices.

Attributes:

Name	Type	Description
data	list[str]	Contains the actual data

resolve_retention_name

resolve_retention_name(snapshot_names)

Resolves a retention_name selector and returns a list based on the selection.

Returns:

Type	Description
list[str]	List of resolved items

Source code in b4_backup/main/dataclass.py

def resolve_retention_name(self, snapshot_names: Iterable[str]) -> list[str]:
    """
    Resolves a retention_name selector and returns a list based on the selection.

    Returns:
        List of resolved items
    """
    if self.data == ["ALL"]:
        return list({x.split("_", maxsplit=1)[1] for x in snapshot_names})

    return self.data

resolve_target

resolve_target(targets)

Resolves a target selector and returns a list based on the selection.

Returns:

Type	Description
list[str]	List of resolved items

Source code in b4_backup/main/dataclass.py

def resolve_target(self, targets: Iterable[str]) -> list[str]:
    """
    Resolves a target selector and returns a list based on the selection.

    Returns:
        List of resolved items
    """
    expanded_data: set[str] = set()
    for item in self.data:
        if item in targets:
            expanded_data.add(item)
            continue

        for target_name in targets:
            if PurePath(target_name).is_relative_to(item):
                expanded_data.add(target_name)

    return list(expanded_data - {"_default"})

RetentionGroup dataclass

RetentionGroup(
    name,
    target_retention,
    is_source=True,
    obsolete_snapshots=set(),
)

Contains the retention ruleset for a target.

Attributes:

Name	Type	Description
name	str	Name of the retention ruleset
target_retention	dict[str, str]	The retention ruleset for the target itself
is_source	bool	True if this is a source retention ruleset
obsolete_snapshots	set[str]	All snapshots in this set will be condidered obsolete

from_target classmethod

from_target(
    retention_name,
    target,
    is_source=True,
    obsolete_snapshots=None,
)

Create an instance from a target and ruleset name.

Parameters:

Name	Type	Description	Default
retention_name	str	Name of the retention ruleset to select from the target	required
target	BackupTarget	Target to get the ruleset from	required
is_source	bool	Select source ruleset or destination ruleset	True
obsolete_snapshots	set[str] | None	All snapshots in this set will be condidered obsolete	None

Returns:

Type	Description
RetentionGroup	RetentionGroup instance

Source code in b4_backup/main/dataclass.py

@classmethod
def from_target(
    cls,
    retention_name: str,
    target: BackupTarget,
    is_source: bool = True,
    obsolete_snapshots: set[str] | None = None,
) -> "RetentionGroup":
    """
    Create an instance from a target and ruleset name.

    Args:
        retention_name: Name of the retention ruleset to select from the target
        target: Target to get the ruleset from
        is_source: Select source ruleset or destination ruleset
        obsolete_snapshots: All snapshots in this set will be condidered obsolete

    Returns:
        RetentionGroup instance
    """
    target_retentions = target.src_retention if is_source else target.dst_retention
    target_retention = target_retentions.get(retention_name) or target_retentions[DEFAULT]

    return RetentionGroup(
        name=retention_name,
        target_retention=target_retention,
        is_source=is_source,
        obsolete_snapshots=obsolete_snapshots or set(),
    )

Snapshot dataclass

Snapshot(
    name, subvolumes, base_path, _subvolume_delimiter="!"
)

Describes a b4_snapshot.

subvolumes_unescaped property

subvolumes_unescaped

Returns:

Type	Description
None	List all subvolumes without delimiter translation as relative paths.

escape_path classmethod

escape_path(path)

Returns:

Type	Description
BackupHostPath	Escaped variant of subvolume path.

Source code in b4_backup/main/dataclass.py

@classmethod
def escape_path(cls, path: BackupHostPath) -> BackupHostPath:
    """
    Returns:
        Escaped variant of subvolume path.
    """
    return BackupHostPath(
        str(path).replace("/", cls._subvolume_delimiter),
        connection=path.connection,
    )

from_new classmethod

from_new(name, subvolumes, base_path)

Create instance from the backup target location.

Parameters:

Name	Type	Description	Default
name	str	Name of the snapshot	required
subvolumes	list[BackupHostPath]	List of subvolumes without delimiter translation	required
base_path	BackupHostPath	Location of this snapshot	required

Source code in b4_backup/main/dataclass.py

@classmethod
def from_new(
    cls, name: str, subvolumes: list[BackupHostPath], base_path: BackupHostPath
) -> "Snapshot":
    """
    Create instance from the backup target location.

    Args:
        name: Name of the snapshot
        subvolumes: List of subvolumes without delimiter translation
        base_path: Location of this snapshot
    """
    return Snapshot(
        name=name,
        subvolumes=[cls.escape_path(x) for x in subvolumes],
        base_path=base_path,
    )

unescape_path classmethod

unescape_path(path)

Returns:

Type	Description
BackupHostPath	Recreates a path from an escaped variant of subvolume path.

Source code in b4_backup/main/dataclass.py

@classmethod
def unescape_path(cls, path: BackupHostPath) -> BackupHostPath:
    """
    Returns:
        Recreates a path from an escaped variant of subvolume path.
    """
    return BackupHostPath(
        str(path).replace(cls._subvolume_delimiter, "/"),
        connection=path.connection,
    )