Module trase.tools.aws.s3_snapshots

Snapshots of S3 "folders" (key prefixes).

A snapshot records the current version ID of every object underneath a key prefix. It does not copy any data: it relies on the bucket having versioning enabled, so that restoring a snapshot is a matter of copying old versions back over the top of the current ones.

Snapshots are stored as JSON files in a per-user data directory (see snapshot_directory()), one file per snapshot, so that they survive between sessions and can be inspected by hand.

The typical workflow is

snapshot = take_snapshot("brazil/soy/sei_pcs/v2.7.0/", name="before-rerun")
...                                    # run something that overwrites those objects
restore_snapshot(load_snapshot("before-rerun"))

See trase.tools.cli.s3 for the trase s3 snapshot|snapshots|restore commands wrapping these functions.

Functions

def default_snapshot_name() ‑> str

The name a snapshot gets when it is not given one: the time it was taken

def list_current_objects(prefix: str, bucket: str = None, client=None) ‑> Dict[str, ObjectVersion]

The current version of every object under prefix, as a dictionary of S3 key to ObjectVersion.

Objects whose latest version is a delete marker - that is, objects which have been deleted but whose old versions are still around - are not included.

def list_snapshots(directory: pathlib.Path | None = None) ‑> List[Snapshot]

All snapshots which have been taken on this machine, most recently taken first.

Files which cannot be parsed as a snapshot are ignored, so that one bad file does not stop the others from being listed.

def load_snapshot(name: str, directory: pathlib.Path | None = None) ‑> Snapshot

Read the snapshot called name

def new_objects(snapshot: Snapshot,
current: Dict[str, ObjectVersion]) ‑> List[str]

Keys which exist under the snapshot's prefix but were not in the snapshot: objects created (or recreated) since it was taken, as listed by list_current_objects().

def normalise_prefix(prefix: str) ‑> str

Interpret a prefix as a folder: brazil/soy becomes brazil/soy/, so that we do not pick up the neighbouring brazil/soybeans/. An empty prefix, meaning the whole bucket, is left alone.

def read_snapshot(path: pathlib.Path) ‑> Snapshot

Read a snapshot from a file

def restore_snapshot(snapshot: Snapshot,
client=None,
delete_new_objects: bool = True,
current: Dict[str, ObjectVersion] | None = None,
printer=<function <lambda>>) ‑> RestoreSummary

Put every object back the way it was when the snapshot was taken.

Objects which have been overwritten are rolled back by copying their old version over the top of the current one; objects which have been deleted are resurrected the same way. Objects whose contents already match the snapshot - either because they are still at the version it recorded, or because they have already been restored to it - are left alone, so that restoring a snapshot twice does not create pointless versions.

Restoring rewrites an object as a fresh copy of an old version, which keeps its contents and metadata but resets its last modified date and storage class.

Args

snapshot

the snapshot to restore

delete_new_objects

also delete objects created since the snapshot was taken. Note that these deletions are themselves versioned, and so can be undone

current

what is in S3 now, from list_current_objects(). Pass this if you have already listed the prefix - to show the user what is about to be deleted, say - so that exactly those objects are the ones acted on

printer

a function used to report progress, for example print

Returns: a summary of what was restored, left alone, deleted and failed

def slugify_snapshot_name(name: str) ‑> str

The name a snapshot is stored under: a slug of the name asked for, so that names double up as file names. Before rerun (2023) becomes before-rerun-2023.

Slugs are also how snapshots are looked up again, so trase s3 restore finds a snapshot whether it is given the name as typed or as slugified.

def snapshot_directory(mkdirs: bool = False) ‑> pathlib.Path

Directory in which snapshots are stored: an OS-appropriate per-user data directory, for example ~/.local/share/trase/s3-snapshots on Linux.

def snapshot_path(name: str, directory: pathlib.Path | None = None) ‑> pathlib.Path

Path of the file in which the snapshot name is (or would be) stored

def take_snapshot(prefix: str,
name: str | None = None,
bucket: str = None,
client=None,
directory: pathlib.Path | None = None,
overwrite: bool = False,
printer=<function <lambda>>) ‑> Snapshot

Record the current version ID of every object under prefix and save it as a snapshot, which can later be rolled back to with restore_snapshot().

Args

prefix

the S3 key prefix ("folder") to snapshot, recursively

name

name of the snapshot, which is slugified. Defaults to the time it was taken

bucket

S3 bucket, defaulting to the configured Trase bucket

directory

where to store the snapshot, defaulting to snapshot_directory()

overwrite

replace an existing snapshot of the same name

printer

a function used to report progress, for example print

Returns: the snapshot which was taken

def write_snapshot(snapshot: Snapshot,
directory: pathlib.Path | None = None,
overwrite: bool = False) ‑> pathlib.Path

Write a snapshot to a file, returning the path it was written to

Classes

class ObjectVersion (version_id: str, etag: str)

One version of one S3 object

Ancestors

• builtins.tuple

Instance variables

var etag : str

Alias for field number 1

var version_id : str

Alias for field number 0

class RestoreSummary (restored: List[str] = <factory>,
unchanged: List[str] = <factory>,
deleted: List[str] = <factory>,
failed: Dict[str, str] = <factory>)

What restore_snapshot() did

Instance variables

var deleted : List[str]

var failed : Dict[str, str]

var restored : List[str]

var unchanged : List[str]

class Snapshot (name: str,
bucket: str,
prefix: str,
taken: datetime.datetime,
objects: Dict[str, ObjectVersion])

The version ID of every object under a key prefix at a moment in time

Static methods

def from_dict(dictionary: dict) ‑> Snapshot

Instance variables

var bucket : str

var name : str

var objects : Dict[str, ObjectVersion]

var prefix : str

prop s3_uri : str

var taken : datetime.datetime

Methods

def to_dict(self) ‑> dict

class SnapshotError (*args, **kwargs)

Raised when a snapshot cannot be taken, read or restored

Ancestors

• builtins.Exception
• builtins.BaseException