B4 backup

B4Backup dataclass

B4Backup(timezone=BaseConfig.timezone)

Main controller class for the backups. Does the backups and stuff.

Parameters:

Name	Type	Description	Default
timezone	str	Timezone to use	timezone

backup

backup(src_host, dst_host, snapshot_name)

Performs a backup for a single target.

dst_host can be none. In this case nothing will be sent and only a snapshot + clean up on source side is performed.

Parameters:

Name	Type	Description	Default
src_host	SourceBackupTargetHost	An active source host instance	required
dst_host	DestinationBackupTargetHost | None	An active destination host instance	required
snapshot_name	str	The name of the new snapshot	required

Source code in b4_backup/main/b4_backup.py

def backup(
    self,
    src_host: SourceBackupTargetHost,
    dst_host: DestinationBackupTargetHost | None,
    snapshot_name: str,
) -> None:
    """
    Performs a backup for a single target.

    dst_host can be none. In this case nothing will be sent and only a snapshot + clean up on source side is performed.

    Args:
        src_host: An active source host instance
        dst_host: An active destination host instance
        snapshot_name: The name of the new snapshot
    """
    log.info("Snapshot name: %s", snapshot_name)

    src_host.create_snapshot(snapshot_name)

    if dst_host:
        src_host.send_snapshot(dst_host, snapshot_name)

    retention_name = ChoiceSelector([self._extract_retention_name(snapshot_name)])
    self.clean(
        src_host=src_host,
        dst_host=dst_host,
        retention_names=retention_name,
    )

clean

clean(
    src_host,
    dst_host=None,
    retention_names=ChoiceSelector(["ALL"]),
)

Apply a retention ruleset on the selected targets.

Parameters:

Name	Type	Description	Default
src_host	SourceBackupTargetHost	An active source host instance	required
dst_host	DestinationBackupTargetHost | None	An active destination host instance	None
retention_names	ChoiceSelector	Name suffix of this backup (retention ruleset)	ChoiceSelector(['ALL'])

Source code in b4_backup/main/b4_backup.py

def clean(
    self,
    src_host: SourceBackupTargetHost,
    dst_host: DestinationBackupTargetHost | None = None,
    retention_names: ChoiceSelector = ChoiceSelector(["ALL"]),
) -> None:
    """
    Apply a retention ruleset on the selected targets.

    Args:
        src_host: An active source host instance
        dst_host: An active destination host instance
        retention_names: Name suffix of this backup (retention ruleset)
    """
    self._clean_target(src_host, dst_host, retention_names)
    self._clean_replace(src_host)
    self._clean_empty_dirs(src_host, dst_host)

delete

delete(host, snapshot_name)

Delete a specific snapshot from a specific target/host.

Parameters:

Name	Type	Description	Default
host	BackupTargetHost	the selected target host	required
snapshot_name	str	The name of the snapshot to delete	required

Source code in b4_backup/main/b4_backup.py

def delete(
    self,
    host: BackupTargetHost,
    snapshot_name: str,
) -> None:
    """
    Delete a specific snapshot from a specific target/host.

    Args:
        host: the selected target host
        snapshot_name: The name of the snapshot to delete
    """
    snapshots = host.snapshots()

    if snapshot_name not in snapshots:
        log.warning("Snapshot %s does not exist on %s", snapshot_name, host.type)
        return

    host.delete_snapshot(snapshots[snapshot_name])

delete_all

delete_all(host, retention_names=ChoiceSelector(['ALL']))

Delete all snapshots from a specific target/host/retention item.

Parameters:

Name	Type	Description	Default
host	BackupTargetHost	the selected target host	required
retention_names	ChoiceSelector	The retention names the snapshots have to contain	ChoiceSelector(['ALL'])

Source code in b4_backup/main/b4_backup.py

def delete_all(
    self,
    host: BackupTargetHost,
    retention_names: ChoiceSelector = ChoiceSelector(["ALL"]),
) -> None:
    """
    Delete all snapshots from a specific target/host/retention item.

    Args:
        host: the selected target host
        retention_names: The retention names the snapshots have to contain
    """
    resolved_retention_names = set(retention_names.resolve_retention_name(host.snapshots()))

    for snapshot_name, snapshot in host.snapshots().items():
        if self._extract_retention_name(snapshot_name) not in resolved_retention_names:
            continue

        host.delete_snapshot(snapshot)

generate_snapshot_name

generate_snapshot_name(name=None)

Generate a name for a new snapshot.

Parameters:

Name	Type	Description	Default
name	str | None	Retention rule name	None

Returns:

Type	Description
str	Name for a snapshot

Source code in b4_backup/main/b4_backup.py

def generate_snapshot_name(self, name: str | None = None) -> str:
    """
    Generate a name for a new snapshot.

    Args:
        name: Retention rule name

    Returns:
        Name for a snapshot
    """
    snapshot_name = arrow.utcnow().to(self.timezone).format(self._timestamp_fmt)

    if name:
        snapshot_name += f"_{name}"

    return snapshot_name

restore

restore(src_host, dst_host, snapshot_name, strategy)

Restore a snapshot to one or more targets.

Parameters:

Name	Type	Description	Default
src_host	SourceBackupTargetHost	An active source host instance	required
dst_host	DestinationBackupTargetHost | None	An active destination host instance	required
snapshot_name	str	Name of the snapshot you want to restore	required
strategy	TargetRestoreStrategy	Restore strategy or procedure to apply	required

Source code in b4_backup/main/b4_backup.py

def restore(
    self,
    src_host: SourceBackupTargetHost,
    dst_host: DestinationBackupTargetHost | None,
    snapshot_name: str,
    strategy: TargetRestoreStrategy,
) -> None:
    """
    Restore a snapshot to one or more targets.

    Args:
        src_host: An active source host instance
        dst_host: An active destination host instance
        snapshot_name: Name of the snapshot you want to restore
        strategy: Restore strategy or procedure to apply
    """
    if snapshot_name == "REPLACE":
        if strategy != TargetRestoreStrategy.REPLACE:
            raise exceptions.SnapshotNotFoundError(
                "REPLACE can only be restored using REPLACE restore strategy"
            )

        log.info("Reverting last REPLACE restore")
        self._rollback_replace(src_host)

    elif strategy == TargetRestoreStrategy.REPLACE:
        log.info("Using REPLACE restore strategy")
        self._restore_replace(src_host, dst_host, snapshot_name)

    else:
        log.info("Using SAFE restore strategy")
        self._restore_safe(src_host, dst_host, snapshot_name)

sync

sync(src_host, dst_host)

Send unsended snapshots to the destination and clean them.

Parameters:

Name	Type	Description	Default
src_host	SourceBackupTargetHost	An active source host instance	required
dst_host	DestinationBackupTargetHost	An active destination host instance	required

Source code in b4_backup/main/b4_backup.py

def sync(
    self,
    src_host: SourceBackupTargetHost,
    dst_host: DestinationBackupTargetHost,
) -> None:
    """
    Send unsended snapshots to the destination and clean them.

    Args:
        src_host: An active source host instance
        dst_host: An active destination host instance
    """
    self.clean(src_host, dst_host)

    src_snapshots = src_host.snapshots()
    dst_snapshots = dst_host.snapshots()

    for snapshot_name in src_snapshots.keys() - dst_snapshots.keys():
        src_host.send_snapshot(dst_host, snapshot_name)

    self.clean(src_host, dst_host)