Python API reference¶
The package exposes three names: BackupConfig, EZBak, and ezbak. Build a
BackupConfig, pass it to EZBak, and call the backup methods.
BackupConfig¶
BackupConfig is the typed configuration model. It validates on construction. It
raises pydantic.ValidationError when a required option is missing, or when a
value is malformed. Every field is in the
configuration reference.
from pathlib import Path
from ezbak import BackupConfig
config = BackupConfig(
name="my-backup",
source_paths=[Path("/data")],
storage_paths=[Path("/backups")],
keep_last=10,
)
A BackupConfig needs a name and at least one storage location
(storage_paths, aws_s3_bucket_name, or both). It does not read the
environment. Only the CLI and the container do that.
EZBak¶
EZBak is the one public class. Construct it with a BackupConfig.
from ezbak import EZBak, BackupConfig
backups = EZBak(BackupConfig(name="my-backup", source_paths=["/data"], storage_paths=["/backups"]))
ezbak() shortcut¶
ezbak(**kwargs) builds the BackupConfig for you. These two lines are
equivalent:
backups = ezbak(name="my-backup", source_paths=["/data"], storage_paths=["/backups"])
backups = EZBak(BackupConfig(name="my-backup", source_paths=["/data"], storage_paths=["/backups"]))
When you want an explicit, reusable configuration object, prefer
EZBak(BackupConfig(...)). Use ezbak(**kwargs) in quick scripts.
Methods¶
| Method | Returns | Purpose |
|---|---|---|
create_backup() |
list[Backup] |
Archive the sources and write to every storage location. |
list_backups() |
list[Backup] |
Every backup, oldest to newest. |
prune_backups(dry_run=False) |
list[Backup] |
Delete the backups the keep rules no longer keep. |
restore_backup(restore_path=None, *, clean_before_restore=False, backup=None) |
RestoreOutcome |
Restore a backup into a directory. |
get_latest_backup() |
Backup \| None |
The newest backup, or None when there are none. |
get_backup_as_of(point_in_time) |
Backup \| None |
The newest backup at or before a point in time. |
backups.create_backup()
print([backup.name for backup in backups.list_backups()])
backups.prune_backups()
backups.restore_backup(restore_path="/restore")
prune_backups(dry_run=True) returns the backups the policy no longer keeps, and
deletes none of them. A real prune returns the backups it confirmed deleted.
Two cases are not errors: no backup to restore, and a target ezbak declined to
overwrite. For those, restore_backup() returns a RestoreOutcome member and
raises nothing. It still raises RestoreFailedError on a real download or extract
failure, so a failed restore never looks like a success.
Breaking change: restore_backup() no longer returns a bool
restore_backup() returned True on a successful restore, and False when
there was no backup to restore. It now returns a RestoreOutcome member.
Update the code that reads the return value as a boolean:
RestoreOutcome¶
restore_backup() returns one of three RestoreOutcome members, so a caller can
tell an actual restore apart from a no-op:
| Member | Meaning |
|---|---|
RestoreOutcome.RESTORED |
ezbak extracted a backup into the target. |
RestoreOutcome.NO_BACKUP |
No backup matched the restore criteria. |
RestoreOutcome.SKIPPED_POPULATED |
skip_restore_if_populated is set and the target already held data, so ezbak left it alone. |
Import RestoreOutcome from ezbak.constants:
from ezbak.constants import RestoreOutcome
outcome = backups.restore_backup(restore_path="/restore")
match outcome:
case RestoreOutcome.RESTORED:
print("Restored")
case RestoreOutcome.NO_BACKUP:
print("Nothing to restore")
case RestoreOutcome.SKIPPED_POPULATED:
print("Target already had data; left it alone")
See Restore backups for skip_restore_if_populated, and
Fresh deploys for the pre-start restore that
both outcomes support.
Point-in-time restore¶
get_backup_as_of(point_in_time) returns the newest backup at or before the end
of the period you name. Pass its result to restore_backup(backup=...).
backup = backups.get_backup_as_of("20241201")
if backup:
backups.restore_backup(restore_path="/restore", backup=backup)
An explicit backup argument takes priority over a configured restore_date,
which in turn takes priority over the latest backup.
unreadable_locations¶
list_backups() never raises. It returns whatever backups it found, even when a
configured storage location cannot be read. Read the unreadable_locations
property alongside it to know whether that result is the whole picture.
backups.list_backups()
if backups.unreadable_locations:
print(f"Inventory incomplete: could not read {', '.join(backups.unreadable_locations)}")
A non-empty list names the storage locations ezbak cannot use or enumerate. The
cause is a bad S3 credential, an unreachable bucket, or a local path ezbak cannot
read. A backup absent from list_backups() can therefore still exist in one of those
locations. create_backup() and restore_backup() treat the same condition as a
hard failure instead of an incomplete result. See
BackupFailedError and
RestoreFailedError below.
The property indexes the storage locations when they are not indexed yet. The first access therefore performs network I/O against S3, instead of returning a cached attribute.
An empty inventory is cached until the next backup run
ezbak caches an index that finds zero backups like any other index.
list_backups() therefore keeps returning that result until
create_backup() or prune_backups() invalidates it. A long-lived process
that watches for archives another writer creates has to set
rebuild_storage_locations = True to force a fresh scan.
ezbak builds its own boto3 session
S3 access goes through a boto3.Session that ezbak constructs itself, so
ezbak does not use a session installed with boto3.setup_default_session().
Pass credentials through BackupConfig, or leave them unset to use the
ambient credential chain.
Exceptions¶
Every exception the library raises subclasses EZBakError, so one
except EZBakError catches any failure.
classDiagram
EZBakError <|-- ConfigurationError
EZBakError <|-- StorageInitError
EZBakError <|-- StorageWriteError
EZBakError <|-- StorageReadError
EZBakError <|-- StorageDeleteError
EZBakError <|-- BackendNotFoundError
EZBakError <|-- BackupFailedError
EZBakError <|-- RestoreFailedError
| Exception | Raised when |
|---|---|
ConfigurationError |
A path or another precondition is invalid: no sources, a source that does not exist, an unusable restore path. |
StorageInitError |
A storage location cannot be initialized: bad credentials, an unreachable bucket. |
StorageWriteError |
A backend cannot write an archive. |
StorageReadError |
A backend cannot read an archive back for a restore. |
StorageDeleteError |
A backend cannot delete an archive during a prune. |
BackendNotFoundError |
Internal invariant failure: no backend handles a storage type. |
BackupFailedError |
One or more storage locations cannot be written. |
RestoreFailedError |
An archive cannot be downloaded, read, or extracted. |
Import them from ezbak.exceptions:
BackupFailedError¶
create_backup() raises BackupFailedError when a configured storage location
cannot be used. It still writes to every location that works, so a partial
failure keeps the copies that succeeded.
from ezbak.exceptions import BackupFailedError
try:
backups.create_backup()
except BackupFailedError as error:
print(f"Failed storage locations: {error.failed_storage_locations}")
print(f"Backups that succeeded: {[b.name for b in error.created_backups]}")
The error carries two attributes:
failed_storage_locations: the locations that failed.created_backups: theBackupobjects written before the failure.
RestoreFailedError¶
restore_backup() raises RestoreFailedError when it cannot download, read, or
extract the archive. This matters most with clean_before_restore, which empties
the target before the extract. Without the raised error, a silent failure leaves
an empty directory and no signal.
from ezbak.exceptions import RestoreFailedError
try:
backups.restore_backup(restore_path="/restore")
except RestoreFailedError as error:
print(f"Restore failed: {error}")
See Failure behavior for how the library, the CLI, and the container each surface these errors.