snapm package

Module contents

Snapm top-level package.

class snapm.FsTabReader(path='/etc/fstab')

Bases: object

A class to read and query data from an fstab file.

This class reads an fstab-like file and provides methods to iterate over its entries and look up specific entries based on their properties.

class FsTabEntry(what, where, fstype, options, freq, passno)

Bases: tuple

__repr__()

Return a nicely formatted representation string

freq

Alias for field number 4

fstype

Alias for field number 2

options

Alias for field number 3

passno

Alias for field number 5

what

Alias for field number 0

where

Alias for field number 1

__init__(path='/etc/fstab')

Initializes the FsTabReader object by reading and parsing the file.

Parameters:

path (str) – The path to the fstab file. Defaults to ‘/etc/fstab’.

Raises:

• SnapmNotFoundError – If the specified fstab file does not exist.

• SnapmSystemError – If there is an error reading the fstab file.

__repr__()

Return a machine readable string representation of this FsTabReader.

lookup(key, value)

Finds and generates all entries matching a specific key-value pair.

Parameters:

• key (str) – The field to search by. Must be one of ‘what’, ‘where’, ‘fstype’, ‘options’, ‘freq’, or ‘passno’.

• value (str|int) – The value to match for the given key.

Yields:

A 6-tuple for each matching fstab entry.

Raises:

KeyError – If the provided key is not a valid fstab field name.

class snapm.MountInfoReader(path: str = '/proc/self/mountinfo')

Bases: object

A class to read and query the mount table of the calling process.

Mounts are indexed by device number rather than by the device name recorded at mount time: a logical volume renamed after it was mounted keeps its old name in the mount table for as long as it stays mounted, and matching on the name then reports it as unmounted (#1030).

/proc/self/mountinfo is read rather than /proc/self/mounts because it reports the device number of each mount directly, so finding the mounts of a device costs one stat() of that device instead of one stat() for every entry in the table.

The device number the kernel reports for a mount is that of the mounted file system, which is the st_rdev of the backing device only for a file system backed by a single block device. A file system with an anonymous superblock (btrfs, overlay) is given a device number with a major of zero that matches no block device at all, so those mounts are additionally indexed by the device named in the mount table: see _index_entry(). That name is the one recorded at mount time, so for an anonymous superblock a device renamed after it was mounted is still reported as unmounted.

class MountInfoEntry(mount_id, parent_id, devno, root, where, options, fstype, what, super_options)

Bases: tuple

__repr__()

Return a nicely formatted representation string

devno

Alias for field number 2

fstype

Alias for field number 6

mount_id

Alias for field number 0

options

Alias for field number 5

parent_id

Alias for field number 1

root

Alias for field number 3

super_options

Alias for field number 8

what

Alias for field number 7

where

Alias for field number 4

__init__(path: str = '/proc/self/mountinfo')

Initialise a MountInfoReader by reading and parsing path.

Parameters:

path (str) – The path to the mountinfo file to read. Defaults to ‘/proc/self/mountinfo’.

Raises:

SnapmSystemError – If the mountinfo file cannot be read.

__repr__()

Return a machine readable string representation of this MountInfoReader.

device_mounted(devpath: str) → bool

Test whether the block device at devpath is mounted.

Parameters:

devpath (str) – The path to the block device to test.

Returns:

True if devpath is mounted, or False otherwise.

Return type:

bool

devno_mounted_at(devno: int, path: str) → bool

Test whether the device with device number devno is the device mounted at path.

The device number the kernel reports for a mount is not the st_rdev of the device it was mounted from when the file system has an anonymous superblock, so the two cannot be compared directly: this goes through the index, which resolves a device to its mounts either way. Call this rather than comparing entry.devno with a device number of your own (#1030).

Parameters:

• devno (int) – The device number of a block device to test, in the form returned by os.stat().st_rdev.

• path (str) – The mount point to test.

Returns:

True if devno is mounted at path, or False otherwise.

Return type:

bool

entries: List[MountInfoEntry]

lookup_device(devpath: str) → List[MountInfoEntry]

Return the mounts of the block device at devpath.

Parameters:

devpath (str) – The path to the block device to look up.

Returns:

The mounts of devpath, in mount table order, or an empty list if it is not mounted, does not exist, or is not a block device.

Return type:

List[MountInfoReader.MountInfoEntry]

lookup_devno(devno: int) → List[MountInfoEntry]

Return the mounts of the device with device number devno.

Mounts of a file system with an anonymous superblock are returned for the device they were mounted by: the index carries them under that device as well as under the device number the kernel reports.

Parameters:

devno (int) – The device number of a block device to look up, in the form returned by os.stat().st_rdev.

Returns:

The mounts of devno, in mount table order, or an empty list if the device is not mounted.

Return type:

List[MountInfoReader.MountInfoEntry]

lookup_mount_point(path: str) → MountInfoEntry | None

Return the mount at path, or None if nothing is mounted there.

A path may appear more than once in the mount table when one mount is stacked over another: the last entry is the one in effect.

Parameters:

path (str) – The mount point to look up.

Returns:

The MountInfoEntry mounted at path, or None.

Return type:

Optional[MountInfoReader.MountInfoEntry]

class snapm.ProgressAwareHandler(stream: TextIO | None = None, **kwargs)

Bases: StreamHandler

A logging handler that coordinates with active Progress instances.

After emitting a log record, notifies any Progress instances writing to the same stream so they can avoid erasing the log message.

__init__(stream: TextIO | None = None, **kwargs)

Initialize the handler.

If stream is not specified, sys.stderr is used.

emit(record)

Emit a log record with progress coordination.

Notifies active progress indicators before emitting the record so they can update their output to avoid being overwritten.

Parameters:

record (logging.LogRecord) – The log record to emit.

Return type:

None

class snapm.Selection(name=None, uuid=None, basename=None, index=None, timestamp=None, nr_snapshots=None, mount_points=None, origin=None, mount_point=None, snapshot_name=None, snapshot_uuid=None, sched_name: None | str = None, sched_sources: None | str = None, sched_default_size_policy: None | str = None, sched_autoindex: None | bool = None, sched_gc_type: None | str = None, sched_gc_params: None | str = None, sched_enabled: None | bool = None, sched_running: None | bool = None, sched_calendarspec: None | str = None)

Bases: object

Selection criteria for snapshot sets and snapshots.

__init__(name=None, uuid=None, basename=None, index=None, timestamp=None, nr_snapshots=None, mount_points=None, origin=None, mount_point=None, snapshot_name=None, snapshot_uuid=None, sched_name: None | str = None, sched_sources: None | str = None, sched_default_size_policy: None | str = None, sched_autoindex: None | bool = None, sched_gc_type: None | str = None, sched_gc_params: None | str = None, sched_enabled: None | bool = None, sched_running: None | bool = None, sched_calendarspec: None | str = None)

__repr__()

Format this Selection object as a machine readable string.

The returned string may be passed to the Selection initialiser to duplicate the original Selection.

Returns:

A machine readable string representation of this Selection object

Return type:

string

__str__()

Format this Selection object as a human readable string.

Returns:

A human readable string representation of this Selection object

Return type:

string

all_attrs = ['name', 'uuid', 'basename', 'index', 'timestamp', 'nr_snapshots', 'mount_points', 'origin', 'mount_point', 'snapshot_name', 'snapshot_uuid', 'sched_name', 'sched_sources', 'sched_default_size_policy', 'sched_autoindex', 'sched_gc_type', 'sched_gc_params', 'sched_enabled', 'sched_running', 'sched_calendarspec']

basename = None

check_valid_selection(snapshot_set=False, snapshot=False, schedule=False)

Check a Selection for valid criteria.

Check this Selection object to ensure it contains only criteria that are valid for the specified object type(s).

Returns None if the object passes the check, or raise ValueError if invalid criteria exist.

Parameters:

• snapshot_set – Selection may include SnapshotSet data

• snapshot – Selection may include Snapshot data

Returns:

None on success

Return type:

NoneType

Raises:

ValueError if excluded criteria are present

classmethod from_cmd_args(cmd_args)

Initialise Selection from command line arguments.

Construct a new Selection object from the command line arguments in cmd_args. Each set selection attribute from cmd_args is copied into the Selection. The resulting object may be passed to either the SnapshotSet or Snapshot search functions. (find_snapshot_sets, find_snapshots, as well as the snapm.command calls that accept a selection argument.

Parameters:

cmd_args – The command line selection arguments.

Returns:

A new Selection instance

Return type:

Selection

index = None

is_null()

Test this Selection object for null selection criteria.

Return True if this Selection object matches all objects, or False otherwise.

Returns:

True if this Selection is null

Return type:

bool

is_single()

Test this Selection object for single item selection criteria.

Returns True if this Selection object matches a single object or False otherwise.

A Selection object matches a single object if either the name or uuid fields is specified.

Returns:

True if this selection is single or False otherwise.

Return type:

bool

mount_point = None

mount_points = None

name = None

nr_snapshots = None

origin = None

sched_autoindex: None | bool = None

sched_calendarspec: None | str = None

sched_default_size_policy: None | str = None

sched_enabled: None | bool = None

sched_gc_params: None | str = None

sched_gc_type: None | str = None

sched_name: None | str = None

sched_running: None | bool = None

sched_sources: None | str = None

schedule_attrs = ['sched_name', 'sched_sources', 'sched_default_size_policy', 'sched_autoindex', 'sched_gc_type', 'sched_gc_params', 'sched_enabled', 'sched_running', 'sched_calendarspec']

snapshot_attrs = ['origin', 'mount_point', 'snapshot_name', 'snapshot_uuid']

snapshot_name = None

snapshot_set_attrs = ['name', 'uuid', 'basename', 'index', 'timestamp', 'nr_snapshots', 'mount_points']

snapshot_uuid = None

timestamp = None

uuid = None

class snapm.SizePolicy(source, mount, free_space, fs_used, dev_size, policy)

Bases: object

Class representing a configured instance of a size policy.

__init__(source, mount, free_space, fs_used, dev_size, policy)

Initialise a new SizePolicy object using the supplied parameters.

Parameters:

• source – The source for this SizePolicy: a mount point or block device path.

• mount – The mount point path if mounted.

• free_space – The free space available to provision the snapshot in bytes.

• fs_used – The current file system occupancy in bytes.

• dev_size – The origin device size in bytes.

• policy – A size policy string.

Returns:

A SizePolicy object configured for the specified size.

property size

Return the backing store size for a snapshot based on the policy configuration.

class snapm.SizePolicyType(*values)

Bases: Enum

Enum class representing the possible snapshot size policies.

FIXED = 'FIXED'

PERCENT_FREE = 'FREE'

PERCENT_SIZE = 'SIZE'

PERCENT_USED = 'USED'

class snapm.SnapStatus(*values)

Bases: Enum

Enum class representing snapshot status: Active, Inactive, Invalid or Reverting.

ACTIVE = 1

INACTIVE = 2

INVALID = 3

REVERTING = 4

__str__()

Return a string representation of this SnapStatus object.

Returns:

“Active”, “Inactive”, “Invalid”, or “Reverting”.

exception snapm.SnapmArgumentError

Bases: SnapmError

An invalid argument was passed to a snapshot manager API call.

exception snapm.SnapmBusyError

Bases: SnapmError

A resource needed by the current command is already in use: for e.g. a snapshot merge is in progress for a previous snapshot set.

exception snapm.SnapmCalloutError

Bases: SnapmError

An error calling out to an external program.

exception snapm.SnapmError

Bases: Exception

Base class for snapshot manager errors.

exception snapm.SnapmExistsError

Bases: SnapmError

The named snapshot set already exists.

exception snapm.SnapmInvalidIdentifierError

Bases: SnapmError

An invalid identifier was given.

exception snapm.SnapmLimitError

Bases: SnapmError

A configured resource limit would be exceeded.

exception snapm.SnapmMountError(what: str, where: str, status: int, stderr: str)

Bases: SnapmError

An error performing a mount operation.

__init__(what: str, where: str, status: int, stderr: str)

Initialise a new SnapmMountError exception.

Parameters:

• what – The source for the failed mount operation.

• where – The intended mount point of the operation.

• status – The exit status of the mount(8) program.

• stderr – The error message from mount(8).

exception snapm.SnapmNoProviderError

Bases: SnapmError

No snapshot provider plugin was found.

exception snapm.SnapmNoSpaceError

Bases: SnapmError

Insufficient space is available for the requested operation.

exception snapm.SnapmNotFoundError

Bases: SnapmError

The requested object does not exist.

exception snapm.SnapmParseError

Bases: SnapmError

An error parsing user input.

exception snapm.SnapmPathError

Bases: SnapmError

An invalid path was supplied, for example attempting to snapshot something that is not a mount point or block device.

exception snapm.SnapmPluginError

Bases: SnapmError

An error performing an action via a plugin.

exception snapm.SnapmRecursionError

Bases: SnapmError

A snapshot set source corresponds to a snapshot of another snapshot set.

exception snapm.SnapmSizePolicyError

Bases: SnapmError

An invalid size policy was specified.

exception snapm.SnapmStateError

Bases: SnapmError

The status of an object does not allow an operation to proceed.

exception snapm.SnapmSystemError

Bases: SnapmError

An error when calling the operating system.

exception snapm.SnapmSystemdError

Bases: SnapmError

An error manipulating systemd units.

exception snapm.SnapmSystemdPendingError

Bases: SnapmSystemdError

A systemd job was accepted but its completion could not be confirmed, either because the wait for the expected unit state timed out, or because the unit state could not be obtained.

The job remains queued and may complete at any time: callers that need to undo the requested state change must do so even though the wait failed.

exception snapm.SnapmTimerError

Bases: SnapmError

An error manipulating systemd timers.

exception snapm.SnapmUmountError(where: str, status: int, stderr: str)

Bases: SnapmError

An error performing an unmount operation.

__init__(where: str, status: int, stderr: str)

Initialise a new SnapmUmountError exception.

Parameters:

• where – The mount point for the failed umount operation.

• status – The exit status of the umount(8) program.

• stderr – The error message from umount(8).

class snapm.Snapshot(name, snapset_name, origin, timestamp, mount_point, provider)

Bases: ABC

Abstract base class for individual snapshots. Each snapshot plugin should subclass Snapshot to provide a specific implementation.

__init__(name, snapset_name, origin, timestamp, mount_point, provider)

Initialise a new Snapshot object.

Parameters:

• name – The name of the snapshot.

• snapset_name – The name of the SnapshotSet this snapshot is a part of.

• origin – The origin volume of the snapshot.

• timestamp – The creation timestamp of the snapshot set.

• mount_point – The mount point path this snapshot refers to.

• provider – The plugin providing this snapshot.

__str__()

Return a string representation of this Snapshot object.

Returns:

A multi-line string describing this snapshot.

activate()

Activate this snapshot.

as_dict()

Return a representation of this Snapshot as a dictionary.

abstract property autoactivate

The autoactivation status of this snapshot. Returns True if the snapshot is automatically activated or False otherwise.

check_resize(size_policy)

Check whether this snapshot can be resized or not by calling the provider plugin with the updated size_policy.

check_revert()

Check whether this snapshot can be reverted or not by calling the provider plugin to verify the state of the snapshot.

deactivate()

Deactivate this snapshot.

delete()

Delete this snapshot.

abstract property devpath

The device path for this snapshot.

abstract property free

The space available to this snapshot in bytes. For snapshots with fixed-size backstores this reflects the space remaining in the current backstore. For snapshots that dynamically allocate space to the snapshot it indicates the pooled space available.

property index

The index of this snapshot, if set, or the special value snapm.SNAPSHOT_INDEX_NONE otherwise.

abstractmethod invalidate_cache()

Invalidate any cached data describing this snapshot.

is_origin_mounted(mountinfo: MountInfoReader) → bool

Test whether the origin volume for this Snapshot is mounted at the snapshot’s mount point according to mountinfo.

Parameters:

mountinfo (MountInfoReader) – The mount table to test against.

Returns:

True if this snapshot’s origin is currently mounted or False otherwise.

Return type:

bool

is_snapshot_mounted(mountinfo: MountInfoReader) → bool

Test whether the snapshot volume for this Snapshot is mounted anywhere according to mountinfo.

Parameters:

mountinfo (MountInfoReader) – The mount table to test against.

Returns:

True if this snapshot is currently mounted or False otherwise.

Return type:

bool

json(pretty=False)

Return a string representation of this Snapshot in JSON notation.

property mount_point

The mount point of this snapshot.

property mounted

Test whether either the snapshot or origin volume for this Snapshot is currently mounted and in use.

Returns:

True if this snapshot or its origin is currently mounted or False otherwise.

property name

The name of this snapshot.

abstract property origin

The origin of this snapshot.

property origin_mounted

Test whether the origin volume for this Snapshot is currently mounted and in use.

Returns:

True if this snapshot’s origin is currently mounted or False otherwise.

abstract property origin_options

File system options needed to specify the origin of this snapshot.

rename(new_name)

Rename a snapshot within a snapshot set.

Parameters:

new_name – The new name for the snapshot.

resize(size_policy)

Attempt to resize this snapshot according to the updated size_policy.

revert()

Request to revert a snapshot and revert the content of the origin volume to its state at the time of the snapshot.

This may be deferred until the next device activation or mount operation for the respective volume.

abstract property size

The size of this snapshot in bytes. For snapshots with fixed-size backstores this reflects the size of the backstore. For snapshots that dynamically allocate space to the snapshot it reflects the device or volume size.

property snapset_basename

The basename of the snapshot set this snapshot belongs to, minus any index.

property snapset_name

The name of the snapshot set this snapshot belongs to.

property snapshot_mounted

Test whether the snapshot volume for this Snapshot is currently mounted and in use.

Returns:

True if this snapshot is currently mounted or False otherwise.

property snapshot_set

The SnapshotSet this snapshot belongs to.

property source

The block device or mount point from which this Snapshot was created.

abstract property status

The status of this snapshot. Returns a SnapStatus enum value representing the current state of the snapshot.

property time

The human readable timestamp of this snapshot.

property timestamp

The numerical timestamp of this snapshot.

to_dict()

Return a representation of this Snapshot as a dictionary.

property uuid

The UUID of this snapshot.

class snapm.SnapshotSet(name, timestamp, snapshots)

Bases: object

Representation of a set of snapshots taken at the same point in time and managed as a group.

__init__(name, timestamp, snapshots)

Initialise a new SnapshotSet object.

Parameters:

• name – The name of this SnapshotSet.

• timestamp – The creation timestamp of this SnapshotSet.

• snapshots – A list of Snapshot objects to be included in this set.

__str__()

Return a string representation of this SnapshotSet.

Returns:

A multi-line string describing this snapshot set.

activate()

Attempt to activate all members of this SnapshotSet.

Raises:

SnapmPluginError if a plugin fails to perform the requested operation.

as_dict(members=False)

Backwards compatibility

property autoactivate

The overall autoactivation status of this snapshot set. Returns True if all snapshots within the set have autoactivate enabled or False otherwise.

property basename

The basename of this snapshot set, minus any index.

property categories: List[str]

Return the time categories to which this SnapshotSet belongs as a list of strings.

Returns:

A list of human readable time category strings.

Return type:

List[str]

property datetime

The timestamp of this snapshot set as a datetime.datetime object.

deactivate()

Attempt to deactivate all members of this SnapshotSet.

Raises:

SnapmPluginError if a plugin fails to perform the requested operation.

delete()

Delete the on-disk representation of this SnapshotSet.

Raises:

SnapmBusyError if the snapshot set is in use. SnapmPluginError if an error occurs deleting snapshot set members.

property devices

The list of block devices in this snapshot set.

property index

The index of this snapshot set, if set, or the special value snapm.SNAPSET_INDEX_NONE otherwise.

json(members=False, pretty=False)

Return a string representation of this SnapshotSet in JSON notation.

property mount_points

The list of mount points in this snapshot set.

property mounted

Test whether the either the origin or snapshot volumes for this SnapshotSet are currently mounted and in use.

Returns:

True if any of the snapshots belonging to this SnapshotSet are currently mounted, or False otherwise.

property name

The name of this snapshot set.

property nr_snapshots

The number of snapshots in this snapshot set.

property origin_mounted

Test whether the origin volumes for this SnapshotSet are currently mounted and in use.

Returns:

True if any of the snapshots belonging to this SnapshotSet are currently mounted, or False otherwise.

rename(new_name)

Rename this SnapshotSet to new_name.

Raises:

SnapmBusyError if the snapshot set is mounted, or SnapmError if a call to rename any member of the snapshot set fails.

resize(sources, size_policies)

Attempt to resize the SnapshotSet according to the list of sources and corresponding size_policy strings.

Raises:

SnapmNoSpaceError if there is insufficient space available for the requested operation.

revert()

Initiate a revert operation on this SnapshotSet.

Raises:

SnapmPluginError if a plugin fails to perform the requested operation.

set_categories(categories: List[str])

Set the time categories to which this SnapshotSet belongs as a list of strings. This should only be called by the Manager class after running snapshot set recategorization: it is not intended to be called from client code.

Parameters:

categories (List[str]) – The list of categories to set for this SnapshotSet.

snapshot_by_mount_point(mount_point)

Return the snapshot corresponding to mount_point.

Parameters:

mount_point – The mount point path to search for.

Returns:

A Snapshot object for the given mount point.

Raises:

SnapmNotFoundError if the specified mount point is not present in this SnapshotSet.

snapshot_by_source(source)

Return the snapshot corresponding to source.

Parameters:

source – The block device or mount point path to search for.

Returns:

A Snapshot object for the given source path.

Raises:

SnapmNotFoundError if the specified source path is not present in this SnapshotSet.

property snapshot_mounted

Test whether the snapshot volumes for this SnapshotSet are currently mounted and in use.

Returns:

True if any of the snapshots belonging to this SnapshotSet are currently mounted, or False otherwise.

property snapshots

The list of snapshots in this snapshot set.

property sources

The list of source mount points and block devices in this snapshot set.

property status

The overall status of this snapshot set. Returns SnapStatus.ACTIVE if all members of the set are valid and active, SnapStatus.INACTIVE if any members are inactive, or SnapStatus.INVALID if any member of the set is invalid.

property time

The human readable timestamp of this snapshot set.

property timestamp

The numerical timestamp of this snapshot set.

to_dict(members=False)

Return a representation of this SnapshotSet as a dictionary.

property uuid

The UUID of this snapshot set.

class snapm.SubsystemFilter(name='')

Bases: Filter

Filters DEBUG records based on a set of enabled subsystem names. Non-DEBUG records or DEBUG records without a ‘subsystem’ attribute are always passed through.

__init__(name='')

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

filter(record)

Filter log records based on subsystem and debug level.

Always passes non-DEBUG records. DEBUG records are filtered based on whether their subsystem is in the enabled subsystems set.

Parameters:

record (logging.LogRecord) – The log record to filter.

Returns:

True if the record should be logged, False otherwise.

Return type:

bool

set_debug_subsystems(subsystems)

Sets the collection of subsystems to allow.

class snapm.UtabReader(path: str = '/run/mount/utab')

Bases: object

A class to read and query the userspace mount table.

The kernel never sees a mount option in the “x-” namespace, so mount(8) keeps those options in a table of its own, and umount(8) removes the entry for a mount point as it unmounts it. An entry therefore belongs to one mount: it appears when that mount is made and goes when it is unmounted, where the kernel’s own device numbers and mount IDs describe a file system and are both reused by a later mount of the same file system at the same path.

The table is maintained by libmount and its format is private to it, so only the fields this needs are read and the rest of each entry is passed over. A table that is not there is an empty one: nothing calls this to find out what is mounted, only to ask whether a mount it has already found carries an option, and on a system where no mount carries one there is no table at all.

A table that is there and cannot be read is a different answer, and readable tells the two apart. Without it a transient read error would read as “this mount carries no option”, which is the answer that makes a caller act.

class UtabEntry(what, where, options)

Bases: tuple

__repr__()

Return a nicely formatted representation string

options

Alias for field number 2

what

Alias for field number 0

where

Alias for field number 1

__init__(path: str = '/run/mount/utab')

Initialise a UtabReader by reading and parsing path.

Parameters:

path (str) – The path to the userspace mount table to read. Defaults to ‘/run/mount/utab’.

__repr__()

Return a machine readable string representation of this UtabReader.

entries: List[UtabEntry]

has_option(path: str, option: str) → bool

Test whether the mount at path carries the mount option option.

Parameters:

• path (str) – The mount point to test.

• option (str) – The option to test for, including its value.

Returns:

True if the mount at path carries option, or False otherwise.

Return type:

bool

lookup_mount_point(path: str) → UtabEntry | None

Return the userspace mount table entry for path, or None if there is none.

The last entry for a path wins, as it does in MountInfoReader.lookup_mount_point(). The two are asked about the same path and their answers are compared, so they have to agree about which mount a path names when one is stacked over another.

Parameters:

path (str) – The mount point to look up.

Returns:

The UtabEntry for path, or None.

Return type:

Optional[UtabReader.UtabEntry]

readable

Whether the table was read. A table that is not there is readable and empty; one that exists and could not be read is neither, and what it would have said is unknown.

snapm.bool_to_yes_no(value)

Convert boolean-like to yes/no string.

Return the string “yes” if value evaluates to True, or “no” otherwise.

Parameters:

value – A boolean value to interpret.

Returns:

A string value that is “yes” if value is True, or “no” otherwise.

snapm.build_snapset_mount_list(snapset: SnapshotSet, fstab: FsTabReader)

Build a list of command line mount unit definitions for the snapshot set snapset. Mount points that are not part of the snapset are substituted from /etc/fstab.

Parameters:

• snapset (SnapshotSet) – The snapshot set to build a mount list for.

• fstab (FsTabReader) – An FsTabReader instance to use.

Returns:

A list of 4-tuples, each containing (device_path, mount_point, filesystem_type, mount_options) for auxiliary mounts. The root filesystem (“/”) and swap entries are excluded from the list.

Return type:

List[Tuple[str, str, str, str]]

snapm.device_number(devpath: str) → int | None

Return the device number of the block device at devpath.

Parameters:

devpath (str) – The path to the block device to stat.

Returns:

The device number of devpath, or None if it does not exist or is not a block device.

Return type:

Optional[int]

snapm.find_snapset_root(snapset, fstab: FsTabReader, origin: bool = False)

Find the device that backs the root filesystem for snapshot set snapset.

If the snapset does not include the root volume look the device up via the fstab.

Parameters:

• snapset (SnapshotSet) – The SnapshotSet to check.

• fstab (FsTabReader) – An FsTabReader instance to use.

• origin (bool) – Always return the origin device, even if a snapshot exists for the root mount point.

Returns:

The device path for the root filesystem.

Return type:

str

Raises:

SnapmNotFoundError – If no root device can be found in the snapshot set or fstab.

snapm.get_current_rss() → int

Return the running process’s current VmRSS value in bytes.

Returns:

VmRSS read from /proc/self/status as an integer byte value, or 0 meaning “VmRSS unavailable”.

Type:

int

snapm.get_debug_mask()

Return the current debug mask for the snapm package.

Returns:

The current debug mask value

Return type:

int

snapm.get_device_fstype(devpath: str) → str

Determine the file system type for the device at devpath.

Parameters:

devpath – The path to the device.

Returns:

The file system type, ‘xfs’, ‘ext4’, ‘swap’, etc.

Return type:

str

Raises:

• SnapmNotFoundError – If the device is not found.

• SnapmSystemError – If the ‘blkid’ command exits with a non-zero status due to reasons other than the identifier not being found (e.g., permission issues).

• SnapmCalloutError – For any other unexpected errors during command execution or parsing.

snapm.get_device_path(by_type: str, identifier: str) → str | None

Translates a filesystem UUID or label to its corresponding device path using the blkid command.

Parameters:

• by_type – The type of identifier to search for.

• identifier – The UUID or label of the filesystem.

Returns:

The device path if found, otherwise None.

Return type:

Optional[str]

Raises:

• ValueError – If ‘identifier’ is empty or ‘by_type’ is not ‘uuid’ or ‘label’.

• SnapmNotFoundError – If the ‘blkid’ command is not found on the system.

• SnapmSystemError – If the ‘blkid’ command exits with a non-zero status due to reasons other than the identifier not being found (e.g., permission issues).

• SnapmCalloutError – For any other unexpected errors during command execution or parsing.

snapm.get_total_memory() → int

Return the total physical memory available to the system, excluding swap, in bytes, or 0 meaning “MemTotal” unavailable.

Returns:

Total physical RAM in bytes.

Return type:

int

snapm.is_size_policy(policy)

Test whether a string is a valid size policy

Parameters:

policy – A possible policy string to test.

Returns:

True if policy is a valid size policy string or False otherwise.

snapm.notify_log_output(stream: TextIO)

Notify progress instances that log output is about to occur on stream.

Called by ProgressAwareHandler before emitting a record.

Parameters:

stream (TextIO) – The stream that will receive output.

snapm.parse_size_with_units(value)

Parse a size string with optional unit suffix and return a value in bytes,

Parameters:

size – The size string to parse.

Returns:

an integer size in bytes.

Raises:

ValueError if the string could not be parsed as a valid size value.

snapm.register_progress(progress: ProgressBase | ThrobberBase)

Register a progress instance for log coordination.

snapm.resolve_device_spec(device_spec: str, raise_on_error: bool = False) → str | None

Resolve a device specification to a device path.

Handles device specifications in the form of: - UUID=<uuid> - filesystem UUID - LABEL=<label> - filesystem label - PARTUUID=<partuuid> - partition UUID - PARTLABEL=<partlabel> - partition label - /dev/… - device path (returned as-is)

Uses get_device_path() for UUID and LABEL resolution, and symlinks in /dev/disk/by-partuuid/ and /dev/disk/by-partlabel/ for partition identifiers.

Parameters:

• device_spec (str) – The device specification to resolve.

• raise_on_error (bool) – If True, raise SnapmNotFoundError when device is not found. If False, return None when device is not found.

Returns:

The resolved device path, or None if not found and raise_on_error is False.

Return type:

Optional[str]

Raises:

SnapmNotFoundError – If the device is not found and raise_on_error is True.

snapm.select_snapshot(select, snapshot)

Test SnapshotSet against Selection criteria.

Test the supplied Snapshot against the selection criteria in select and return True if it passes, or False otherwise.

Parameters:

• select – The selection criteria

• snapshot – The Snapshot to test

Return type:

bool

Returns:

True if Snapshot passes selection or False otherwise.

snapm.select_snapshot_set(select, snapshot_set)

Test SnapshotSet against Selection criteria.

Test the supplied SnapshotSet against the selection criteria in select and return True if it passes, or False otherwise.

Parameters:

• select – The selection criteria

• snapshot_set – The SnapshotSet to test

Return type:

bool

Returns:

True if SnapshotSet passes selection or False otherwise.

snapm.set_debug_mask(mask)

Set the debug mask for the snapm package.

Parameters:

mask – the logical OR of the SNAPM_DEBUG_* values to log.

Return type:

None

snapm.size_fmt(value)

Format a size in bytes as a human readable string.

Parameters:

value – The integer value to format.

Returns:

A human readable string reflecting value.

snapm.unescape_mounts(escaped: str) → str

Unescape octal escapes in values read from /proc/*mounts

The kernel escapes a fixed set of characters when writing mount table fields. From fs/proc_namespace.c:

static inline void mangle(struct seq_file *m, const char *s)
{
        seq_escape(m, s, " \t\n\\#");
}

Giving five sequences that may appear in a field:

• \040 represents a space

• \011 represents a tab

• \012 represents a newline

• \043 represents a hash

• \134 represents a backslash

No other character is escaped, so every remaining byte - including multi-byte UTF-8 sequences - passes through unmodified.

Note: some systemd units (like systemd-cryptsetup credential mounts) have literal escape sequence characters in their directory names (e.g. \x2d). The kernel escapes only the leading backslash, so these appear as \134x2d in /proc/mounts and decode back to a literal \x2d here.

Parameters:

escaped (str) – The string to unescape.

Returns:

The unescaped string with octal values replaced by literal character values.

Return type:

str

snapm.unregister_progress(progress: ProgressBase | ThrobberBase)

Unregister a progress instance.

Submodules

snapm.manager module

Top level interface to the snapshot manager.

class snapm.manager.CalendarSpec(calendarspec: str)

Bases: object

Class representing systemd CalendarSpec expressions.

__init__(calendarspec: str)

Validate and parse a systemd calendarspec expression into a CalendarSpec object.

Parameters:

calendarspec – A string containing an calendarspec expression.

Raises:

ValueError if calendarspec is not a valid calendarspec expression.

__repr__()

Return a string representation of this CalendarSpec instance in the form of a call to the CalendarSpec initializer.

__str__()

Return a string representation of this CalendarSpec instance.

_cond_refresh()

Refresh this CalendarSpec object’s time-dependent properties if the current next_elapse value is in the past.

_refresh()

Refresh this CalendarSpec object’s time-dependent properties.

property from_now

Return a string representation of the time remaining until this CalendarSpec next elapses.

Returns:

The time remaining until the next elapse as a string.

Return type:

str

property in_utc

Return the next elapse time for this CalendarSpec object as an instance of datetime.datetime in UTC.

Returns:

The next elapse time as a UTC datetime object.

Return type:

datetime.datetime

property next_elapse

Return the next elapse time for this CalendarSpec object as an instance of datetime.datetime.

Returns:

The next elapse time as a datetime object.

Return type:

datetime.datetime

property occurs

True if this CalendarSpec object’s calendar expression will occur in the future, or False otherwise.

property original

The original form of this CalendarSpec as a string.

class snapm.manager.GcPolicy(policy_name: str, policy_type: GcPolicyType, params: dict)

Bases: object

An instance of a garbage collection policy.

__init__(policy_name: str, policy_type: GcPolicyType, params: dict)

Initialise a new GcPolicy.

Parameters:

• policy_type (GcPolicyType) – The policy type.

• params (dict) – A dictionary of parameters suitable for policy_type.

__repr__()

Return a machine readable string representation of this GcPolicy.

Returns:

A machine readable string.

Return type:

str

__str__()

Return a human readable string representation of this GcPolicy.

Returns:

A human readable string.

Return type:

str

disable()

Disable this GcPolicy and its corresponding timer.

enable()

Enable this GcPolicy and its corresponding timer.

property enabled

Return True if this GcPolicy and its corresponding timer are enabled, and False otherwise.

evaluate(sets: List[SnapshotSet])

Evaluate the list of SnapshotSet objects in sets against this GcPolicy and return a list of SnapshotSet objects that should be garbage collected.

Parameters:

sets (list[SnapshotSet].) – The list of SnapshotSet objects to evaluate, sorted in order of increasing creation date.

Returns:

A list of SnapshotSet objects to garbage collect.

Return type:

list[SnapshotSet]

classmethod from_cmd_args(cmd_args)

Initialise garbage collection policy from command line arguments.

Construct a new GcPolicy object from the command line arguments in cmd_args. The GcPolicyType is determined by the policy_type argument, and the type-specific policy parameters are initialised from the remaining arguments.

Parameters:

cmd_args – The command line selection arguments.

Returns:

A new GcPolicy instance

Return type:

GcPolicy

classmethod from_dict(data)

Instantiate a GcPolicy object from values in data.

Parameters:

data (dict) – A dictionary describing a GcPolicy instance.

Returns:

An instance of GcPolicy reflecting the values in data.

Return type:

GcPolicy

classmethod from_json(value)

Instantiate a GcPolicy object from a JSON string in value.

Parameters:

value – A JSON string describing a GcPolicyParams instance.

Returns:

An instance of a GcPolicyParams subclass reflecting the values in the JSON string value.

Return type:

GcPolicyParams

property has_params

Return True if this GcPolicy object has defined policy parameters or False otherwise.

Returns:

True if parameters are set else False.

Return type:

bool

json(pretty=False)

Return a JSON representation of this GcPolicy.

Returns:

This GcPolicy as a JSON string.

Return type:

str

property name

This GcPolicy instance’s name.

property params

This GcPolicy instance’s GcPolicyParams value.

property running

Return True if this GcPolicy and its corresponding timer are enabled, and False otherwise.

start()

Start this the timer for this GcPolicy.

stop()

Stop the timer for this GcPolicy.

to_dict()

Return a dictionary representation of this GcPolicy.

Returns:

This GcPolicy as a dictionary.

Return type:

dict

property type

This GcPolicy instance’s GcPolicyType value.

class snapm.manager.GcPolicyType(*values)

Bases: Enum

Garbage collection policy types enum.

AGE = 'Age'

ALL = 'All'

COUNT = 'Count'

TIMELINE = 'Timeline'

class snapm.manager.Manager(debug: bool = False)

Bases: object

Snapshot Manager high level interface.

__init__(debug: bool = False)

Initialise a new Manager instance.

Parameters:

debug (bool) – Raise exceptions in plugin operations for debugging.

_build_service_list(add: List[str], remove: List[str]) → List[str]

Build the effective service list for a single snapshot set operation.

The services configured in _SERVICES_D_PATH are loaded and the units named in add and remove are then applied to the result.

A unit named in add that does not exist is an error: the caller asked for a unit that cannot be acted upon. A unit that exists but that is not running is ignored with a warning, as for a configured service: a unit that was not running when the snapshot set was created must not be started once it is complete.

A unit named in remove that is not in the service list is ignored with a warning. The list of configured services varies from system to system and a request to exclude a service that this system does not run is not an error.

Parameters:

• add (List[str]) – A list of service unit names to add to the list.

• remove (List[str]) – A list of service unit names to remove from the list.

Returns:

The list of service unit names to stop and re-start.

Return type:

List[str]

Raises:

SnapmArgumentError if a unit named in add does not exist.

_check_recursion(origins)

Verify that each entry in origins corresponds to a device that is not a snapshot belonging to another snapshot set.

Parameters:

origins – A list of origin devices to check.

Raises:

SnapmRecursionError if an origin device is a snapshot.

_find_and_verify_plugins(sources, size_policies)

Find snapshot provider plugins for each source in sources and verify that a provider exists for each source present.

Parameters:

• sources – A list of source mount point or block device paths.

• size_policies – A dictionary mapping sources to size policies.

Returns:

A dictionary mapping sources to plugins.

_find_next_index(set_basename)

Find the next index value for the recurring snapset with basename set_basename.

Parameters:

basename – The basename of the recurring snapset

Returns:

An integer index value

_load_plugin_config(plugin_name: str) → ConfigParser

Load optional configuration file for plugin plugin_name.

Parameters:

plugin_name (str) – The name of the plugin to load config for.

Returns:

A (possibly empty) ConfigParser instance.

Return type:

ConfigParser

_load_services() → List[str]

Load service list from disk.

_recategorize_snapshot_sets()

Update the mapping of snapshot sets to “yearly”, “monthly”, “quarterly”, etc. categories.

_set_autoactivate(snapset, auto=False)

Set autoactivation for snapset.

Parameters:

• snapset – The SnapshotSet object to operate on.

• auto – True to enable autoactivation or False otherwise.

_snapset_from_name_or_uuid(name=None, uuid=None)

Look a snapshot set up by name or uuid. Returns a SnapshotSet corresponding to either name or uuid`, or raises ``SnapmError on error.

Parameters:

• name – The name of the snapshot set to look up.

• uuid – The UUID of the snapshot set to look up.

Returns:

A SnapshotSet corresponding to the given name or UUID.

Raises:

SnapmNotFoundError is the name or UUID cannot be found. SnapmInvalidIdentifierError if the name and UUID do not match.

_start_services(services: List[str])

Start systemd service units after snapshot creation.

Failure to start an individual service unit is logged and does not prevent the remaining units from being started.

Parameters:

services (List[str]) – The list of service unit names to start.

_stop_services(services: List[str])

Stop systemd service units before snapshot creation.

If a service unit cannot be stopped the units already stopped are re-started before the error is re-raised. A unit for which systemd accepted a stop job is re-started even though the stop could not be confirmed: the unit may stop at any time after the wait for it is abandoned.

Parameters:

services (List[str]) – The list of service unit names to stop.

Raises:

SnapmSystemdError if any service unit fails to stop.

_validate_snapset_name(name)

Validate a snapshot set name.

Returns if name is a valid snapshot set name or raises an appropriate exception otherwise.

Parameters:

name – The snapshot set name to validate.

Raises:

SnapmExistsError if the name is already in use, or SnapmInvalidIdentifierError if the name fails validation.

activate_snapshot_set(name=None, uuid=None)

Activate snapshot set by name or uuid.

Parameters:

• name – The name of the snapshot set to activate.

• uuid – The UUID of the snapshot set to activate.

activate_snapshot_sets(selection)

Activate snapshot sets matching selection criteria selection.

Parameters:

selection – Selection criteria for snapshot sets to activate.

create_snapshot_set(name, source_specs, default_size_policy=None, boot=False, revert=False, autoindex=False, services=None)

Create a snapshot set of the supplied mount points with the name name.

Parameters:

• name – The name of the snapshot set.

• source_specs – A list of mount point and block device paths to include in the set.

• default_size_policy – A default size policy to use for the set.

• boot – Create a snapshot boot entry for this snapshot set.

• revert – Create a revert boot entry for this snapshot set.

• autoindex – Treat name as the basename of a recurring snapshot set and generate and append an appropriate index value.

• services – A list of modifications to the configured service list to apply to this operation. Each element names a service unit to add to the list, or, when prefixed with ‘-’, a service unit to remove from it.

Raises:

SnapmExistsError if the name is already in use, SnapmInvalidIdentifierError if the name fails validation, or SnapmArgumentError if the service list modifications fail validation.

create_snapshot_set_boot_entry(name=None, uuid=None)

Create a snapshot boot entry for the specified snapshot set.

Parameters:

• name – The name of the snapshot set.

• uuid – The UUID of the snapshot set.

Raises:

SnapmExistsError if a boot entry already exists for the snapshot set.

Raises:

SnapmPluginError if autoactivation could not be enabled.

create_snapshot_set_revert_entry(name=None, uuid=None)

Create a revert boot entry for the specified snapshot set.

Parameters:

• name – The name of the snapshot set.

• uuid – The UUID of the snapshot set.

deactivate_snapshot_set(name=None, uuid=None)

Deactivate snapshot set by name or uuid.

Parameters:

• name – The name of the snapshot set to deactivate.

• uuid – The UUID of the snapshot set to deactivate.

deactivate_snapshot_sets(selection)

Deactivate snapshot sets matching selection criteria selection.

Parameters:

selection – Selection criteria for snapshot sets to deactivate.

delete_snapshot_set(name=None, uuid=None, recategorize=True)

Remove snapshot sets by name or UUID.

Parameters:

• name (str) – The name of the snapshot set to delete.

• uuid (UUID) – The UUID of the snapshot set to delete.

• recategorize (bool) – Recategorize snapshot sets after deletion.

delete_snapshot_sets(selection)

Remove snapshot sets matching selection criteria selection.

Parameters:

selection – Selection criteria for snapshot sets to remove.

discover_snapshot_sets()

Discover snapshot sets by calling into each plugin to find individual snapshots and then aggregating them together into snapshot sets.

Initialises the snapshot_sets, by_name and by_uuid members with the discovered snapshot sets.

find_snapshot_sets(selection=None)

Find snapshot sets matching selection criteria.

Parameters:

selection – Selection criteria to apply.

Returns:

A list of matching SnapshotSet objects.

find_snapshots(selection=None)

Find snapshots matching selection criteria.

Parameters:

selection – Selection criteria to apply.

Returns:

A list of matching Snapshot objects.

rename_snapshot_set(old_name, new_name)

Rename snapshot set old_name as new_name.

Parameters:

• old_name – The name of the snapshot set to be renamed.

• new_name – The new name of the snapshot set.

Raises:

SnapmExistsError if the name is already in use, SnapmInvalidIdentifierError if the name fails validation, or SnapmBusyError if the snapshot set is mounted.

resize_snapshot_set(source_specs, name=None, uuid=None, default_size_policy=None)

Resize snapshot set named name or having UUID uuid.

Request to resize each snapshot included in source_specs according to the given size policy, or apply default_size_policy if set.

Parameters:

• name – The name of the snapshot set to resize.

• uuid – The UUID of the snapshot set to resize.

• source_specs – A list of mount points and optional size policies.

• default_size_policy – A default size policy to apply to the resize.

revert_snapshot_set(name=None, uuid=None)

Revert snapshot set named name or having UUID uuid.

Request to revert each snapshot origin within each snapshot set to the state at the time the snapshot was taken.

Parameters:

• name – The name of the snapshot set to revert.

• uuid – The UUID of the snapshot set to revert.

revert_snapshot_sets(selection)

Revert snapshot sets matching selection criteria selection.

Request to revert each snapshot origin within each snapshot set to the state at the time the snapshot was taken.

Parameters:

selection – Selection criteria for snapshot sets to revert.

set_autoactivate(selection, auto=False)

Set autoactivation state for snapshot sets matching selection criteria selection.

Parameters:

• selection – Selection criteria for snapshot sets to set autoactivation.

• auto – True to enable autoactivation or False otherwise.

split_snapshot_set(name, new_name, source_specs)

Split the snapshot set named name into two snapshot sets, with new_name containing the listed sources and name containing all remaining snapshots.

If new_name is None the listed snapshot are removed from name and permanently deleted. This operation cannot be undone.

Parameters:

• name – The name of the snapshot set to split

• new_name – The name of the newly split off snapshot set

• sources – The list of sources to include in the new snapshot set

Raises:

SnapmBusyError if the snapshot set is mounted.

class snapm.manager.Schedule(name: str, sources: List[str], default_size_policy: str | None, autoindex: bool, calendarspec: str, gc_policy: GcPolicy, boot=False, revert=False)

Bases: object

An individual snapshot schedule instance with create and garbage collection timers. Tracks timer configuration, name, sources, size policies, enabled/disabled, nr snapshots, next elapse.

__init__(name: str, sources: List[str], default_size_policy: str | None, autoindex: bool, calendarspec: str, gc_policy: GcPolicy, boot=False, revert=False)

Initialise a new Schedule instance.

Parameters:

• name (str) – The name of the Schedule.

• sources (list[str]) – The source specs to include in this Schedule.

• default_size_policy (str) – The default size policy for this Schedule.

• autoindex (bool) – Enable autoindex names for this Schedule.

• calendarspec (str) – The OnCalendar expression for this Schedule.

• policy (GcPolicy) – The garbage collection policy for this Schedule.

Returns:

The new Schedule instance.

Return type:

Schedule

__repr__()

Return a machine-readable representation of this Schedule.

Returns:

A machine-readable string.

Return type:

str

__str__()

Return a human-readable representation of this Schedule.

Returns:

A human-readable string.

Return type:

str

property autoindex

The autoindex property for this Schedule.

property boot

Return True if this Schedule is configured to enable snapshot set boot entries when creating snapshot sets.

property calendarspec

The OnCalendar expression for the timer associated with this Schedule instance.

property default_size_policy

The default size policy for this Schedule.

delete_config()

Delete this Schedule’s on-disk configuration.

disable()

Disable this Schedule and its corresponding timer.

enable()

Enable this Schedule and its corresponding timers.

property enabled

Return True if this Schedule and its corresponding timer are enabled, and False otherwise.

classmethod from_dict(data)

Instantiate a Schedule object from values in data.

Parameters:

data (dict) – A dictionary describing a Schedule instance.

Returns:

An instance of Schedule reflecting the values in data.

Return type:

GcPolicyParams

classmethod from_file(sched_file)

Initialise a new Schedule instance from an on-disk JSON configuration file.

Parameters:

sched_file (str) – The path to the schedule configuration file.

Returns:

A new Schedule instance.

Return type:

Schedule

gc(sets: List[SnapshotSet]) → List[str]

Apply the configured garbage collection policy for this Schedule.

property gc_policy

The garbage collection policy for this Schedule.

json(pretty=False)

Return a JSON representation of this Schedule.

Returns:

This Schedule as a JSON string.

Return type:

str

property name

The name of this Schedule.

property next_elapse

Return the next_elapse time of this Schedule’s timer.

property revert

Return True if this Schedule is configured to enable snapshot set revert entries when creating snapshot sets.

property running

Return True if this Schedule and its corresponding timer are enabled, and False otherwise.

property sources

The source specs value for this Schedule.

start()

Start this the timer for this Schedule.

stop()

Stop the timer for this Schedule.

to_dict()

Return a dictionary representation of this Schedule.

Returns:

This Schedule as a dictionary.

Return type:

dict

write_config(sched_dir: str)

Write this Schedule’s configuration to disk.

Parameters:

sched_dir (str) – The path at which to write the configuration file.

snapm.command module

The snapm.command module provides both the snapm command line interface infrastructure, and a simple procedural interface to the snapm library modules.

The procedural interface is used by the snapm command line tool, and may be used by application programs, or interactively in the Python shell by users who do not require all the features present in the snapm object API.

snapm.command.PR_DIFF = 16

DiffRecord report object type

snapm.command.PR_PLUGIN = 4

Plugin report object type

snapm.command.PR_SCHEDULE = 8

Schedule report object type

snapm.command.PR_SNAPSET = 1

Snapshot set report object type

snapm.command.PR_SNAPSHOT = 2

Snapshot report object type

class snapm.command.ReportObj(snapset=None, snapshot=None, plugin=None, schedule=None, diff=None)

Bases: object

Common report object for snapm reports

__init__(snapset=None, snapshot=None, plugin=None, schedule=None, diff=None)

snapm.command._activate_cmd(cmd_args)

Activate snapshot set command handler.

Attempt to activate the snapshot sets that match the given selection criteria.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._add_autoactivate_args(parser)

snapm.command._add_create_args(parser)

Add create command arguments.

snapm.command._add_diff_args(parser)

snapm.command._add_identifier_args(parser, snapset=False, snapshot=False)

Add snapshot set or snapshot identifier command line arguments.

snapm.command._add_json_arg(parser)

snapm.command._add_plugin_subparser(type_subparser)

Add subparser for ‘plugin’ commands.

Parameters:

type_subparser – Command type subparser

snapm.command._add_policy_args(parser, required=True)

snapm.command._add_report_args(parser)

Add common reporting command line arguments.

snapm.command._add_schedule_config_arg(parser)

snapm.command._add_schedule_subparser(type_subparser)

Add subparser for ‘schedule’ commands.

Parameters:

type_subparser – Command type subparser

snapm.command._add_snapset_subparser(type_subparser)

Add subparser for ‘snapset’ commands.

Parameters:

type_subparser – Command type subparser

snapm.command._add_snapshot_subparser(type_subparser)

Add subparser for ‘snapshot’ commands.

Parameters:

type_subparser – Command type subparser

snapm.command._autoactivate_cmd(cmd_args)

Autoactivation status snapshot set command handler.

Attempt to set the autoactivation status for snapshot set that match the given selection criteria.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._create_cmd(cmd_args)

create snapshot set command handler. attempt to create the specified snapshot set.

Parameters:

cmd_args – command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._deactivate_cmd(cmd_args)

Deactivate snapshot set command handler.

Attempt to deactivate the snapshot sets that match the given selection criteria.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._debug_manager(cmd_args: Namespace)

Return True if “manager” subsystem debugging is enabled in the supplied cmd_args and False otherwise.

Parameters:

cmd_args (Namespace.) – A Namespace instance resulting from parsing command line arguments. This function assumes that the supplied object has a ``debug` property.

Returns:

True if manager subsystem debugging is enabled.

Return type:

bool

snapm.command._delete_cmd(cmd_args)

Delete snapshot set command handler.

Attempt to delete the specified snapshot set.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._diff_cache_opts(cache: str, expires: int | None)

Helper to set cache options for diff commands.

Parameters:

• cache (str) – Cache mode string (“auto”, “always”, “never”)

• expires (Optional[int]) – Expiry value in seconds (0 to disable, -1 to use built-in defaults).

Returns:

A 2-tuple of (cache_enabled: bool, expires: int).

Return type:

(bool, int)

snapm.command._diff_cmd(cmd_args)

Diff snapshot set command handler.

Compare between snapshot sets or the running system.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._diffreport_cmd(cmd_args)

Diff Report snapshot set command handler.

Compare between snapshot sets or the running system.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._do_print_type(report_fields, selected, output_fields=None, opts=None, sort_keys=None, title=None)

Print an object type report (snapshot set, snapshot)

Helper for list function that generate reports.

Format a set of snapshot set or snapshot objects matching the given criteria and format them as a report, returning the output as a string.

Selection criteria may be expressed via a Selection object passed to the call using the selection parameter.

Parameters:

• selection – A Selection object giving selection criteria for the operation

• output_fields – a comma-separated list of output fields

• opts – output formatting and control options

• sort_keys – a comma-separated list of sort keys

Return type:

str

snapm.command._exec_cmd(cmd_args)

Execute in snapshot set command handler.

Execute the specified command in the given snapshot set.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._expand_fields(default_fields, output_fields)

Expand output fields list from command line arguments.

snapm.command._generic_list_cmd(cmd_args, select, opts, manager, print_fn, data: object = None)

Generic list command implementation.

Implements a simple list command that applies selection criteria and calls a print_*() API function to display results.

Callers should initialise identifier and select appropriately for the specific command arguments.

Parameters:

• cmd_args – the command arguments

• select – selection criteria

• opts – reporting options object

• print_fn – the API call to display results. The function must accept the selection, output_fields, opts, and sort_keys keyword arguments

• data (object) – Optional precomputed data for reporting.

Returns:

Integer status code.

Return type:

int

snapm.command._list_cmd(cmd_args)

List snapshot sets command handler.

List the snapshot sets that match the given selection criteria as a tabular report, with one snapshot set per row.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._log_debug_command(msg, *args, **kwargs)

A wrapper for command subsystem debug logs.

snapm.command._mount_cmd(cmd_args)

Mount snapshot set command handler.

Mount the specified snapshot set (by default snapshot set mounts appear under /run/snapm/mounts/<name>).

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._plugin_list_cmd(cmd_args)

List available plugins.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._prune_cmd(cmd_args)

Prune snapshot set command handler.

Attempt to prune the specified sources from the given snapshot set.

Parameters:

cmd_args – Command line arguments for the command.

Returns:

integer status code returned from main()

snapm.command._rename_cmd(cmd_args)

Rename snapshot set command handler.

Attempt to rename the specified snapshot set.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._report_obj_types = [<snapm.report.ReportObjType object>, <snapm.report.ReportObjType object>, <snapm.report.ReportObjType object>, <snapm.report.ReportObjType object>, <snapm.report.ReportObjType object>]

Report object types table for snapm.command reports

snapm.command._report_opts_from_args(cmd_args)

snapm.command._resize_cmd(cmd_args)

Resize snapshot set command handler.

Attempt to resize the snapshots contained in the given snapshot set according to the corresponding size policy.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._revert_cmd(cmd_args)

Revert snapshot set command handler.

Attempt to revert the specified snapshot set.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._schedule_create_cmd(cmd_args)

Create schedule.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._schedule_delete_cmd(cmd_args)

Delete schedule.

Parameters:

cmd_args – Command line arguments for the command.

Returns:

integer status code returned from main()

snapm.command._schedule_disable_cmd(cmd_args)

Disable schedule.

Parameters:

cmd_args – Command line arguments for the command.

Returns:

integer status code returned from main()

snapm.command._schedule_edit_cmd(cmd_args)

Edit schedule.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._schedule_enable_cmd(cmd_args)

Enable schedule.

Parameters:

cmd_args – Command line arguments for the command.

Returns:

integer status code returned from main()

snapm.command._schedule_gc_cmd(cmd_args)

Garbage collect schedule.

Parameters:

cmd_args – Command line arguments for the command.

Returns:

integer status code returned from main().

snapm.command._schedule_list_cmd(cmd_args)

List configured schedules.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._schedule_show_cmd(cmd_args)

List configured schedules.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._shell_cmd(cmd_args)

Shell snapshot set command handler.

Start an interactive shell in the given snapshot set. The shell invoked defaults to the SHELL environment variable if set, or the program at /bin/bash otherwise.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._show_cmd(cmd_args)

Show snapshot set command handler.

Show the snapshot sets that match the given selection criteria as a multi-line report.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._snapshot_activate_cmd(cmd_args)

Activate snapshot command handler.

Attempt to activate the snapshots that match the given selection criteria.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._snapshot_autoactivate_cmd(cmd_args)

Autoactivate snapshot command handler.

Attempt to set autoactivation status for the snapshots that match the given selection criteria.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._snapshot_deactivate_cmd(cmd_args)

Deactivate snapshot command handler.

Attempt to deactivate the snapshots that match the given selection criteria.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._snapshot_list_cmd(cmd_args)

List snapshots command handler.

List the snapshot that match the given selection criteria as a tabular report, with one snapshot per row.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._snapshot_show_cmd(cmd_args)

Show snapshots command handler.

Show the snapshots that match the given selection criteria as a multi-line report.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command._split_cmd(cmd_args)

Split snapshot set command handler.

Attempt to split the specified sources from the first snapshot set argument into the second.

Parameters:

cmd_args – Command line arguments for the command.

Returns:

integer status code returned from main()

snapm.command._str_indent(string, indent)

Indent all lines of a multi-line string.

Indent each line of the multi line string string to the specified indentation level.

Parameters:

• string – The string to be indented

• indent – The number of characters to indent by

Returns:

str

snapm.command._umount_cmd(cmd_args)

Unmount snapshot set command handler.

Unmount the specified snapshot set. Snapshot sets mounted at a custom location are discovered automatically; –mount-root is only needed to select between them if the set is mounted at more than one path.

Parameters:

cmd_args – Command line arguments for the command

Returns:

integer status code returned from main()

snapm.command.create_schedule(manager: Manager, name: str, sources: List[str], default_size_policy: str, autoindex: bool, calendarspec: str | CalendarSpec, policy: GcPolicy, boot=False, revert=False)

Create a new schedule from a list of mount point and block device source paths.

Parameters:

• manager – The manager context to use.

• name – The name of the new schedule.

• sources – A list of mount point or block devices to snapshot.

• default_size_policy – The default size policy for this schedule.

• autoindex – Treat name as the basename of a recurring scheduled snapshot set and generate and append an appropriate index value.

• calendarspec – The calendar trigger expression for this schedule.

• policy – The garbage collection policy for this schedule.

• boot – Create a boot entry for snapshot sets created by this schedule.

• revert – Create a revert boot entry for snapshot sets created by this schedule.

snapm.command.create_snapset(manager, name, sources, size_policy=None, boot=False, revert=False, autoindex=False, services=None)

Create a new snapshot set from a list of mount point and block device source paths.

Parameters:

• manager – The manager context to use

• name – The name of the new snapshot set

• sources – A list of mount point or block devices to snapshot

• size_policy – The default size policy for this snapshot set.

• boot – Create a boot entry for this snapshot set.

• revert – Create a revert boot entry for this snapshot set.

• autoindex – Treat name as the basename of a recurring snapshot set and generate and append an appropriate index value.

• services – A list of modifications to the configured service list to apply to this snapshot set. Each element names a service unit to add to the list, or, when prefixed with ‘-’, a service unit to remove from it.

snapm.command.delete_schedule(manager: Manager, name: str)

Delete schedule by name. This disables the schedule and removes all on-disk configuration data.

Parameters:

• manager – The manager context to use.

• name – The name of the schedule to delete.

snapm.command.delete_snapset(manager, selection)

Delete snapshot set matching selection criteria.

Parameters:

• manager – The manager context to use

• selection – Selection criteria for the snapshot set to remove.

snapm.command.diff_snapsets(manager: Manager, diff_from: str, diff_to: str, options: DiffOptions, cache: bool = True, cache_expires: int = -1, color: str = 'auto') → FsDiffResults

Find differences between snapshot sets or the running system.

Use the difference engine to compare two snapshot sets or a snapshot set and the running system.

The reserved name ‘.’ is used to reference the running system in either diff_from or diff_to.

Parameters:

• manager (Manager) – The Manager instance to use.

• diff_from (str) – The name of the snapshot set (or ‘.’ for the running system) to use as the left side of the comparison.

• diff_to (str) – The name of the snapshot set (or ‘.’ for the running system) to use as the right side of the comparison.

• options (DiffOptions) – Options controlling the comparison.

• cache (bool) – Set to False to disable use of the diffcache.

• cache_expires (int) – Specify cache expiry time in seconds (0 to disable expiry, -1 for built-in default).

Returns:

A file system diff results container.

Return type:

FsDiffResults

snapm.command.disable_schedule(manager: Manager, name: str)

Enable an existing schedule. This disables the systemd timer units associated with this schedule.

Parameters:

• manager – The manager context to use.

• name – The name of the schedule to disable.

snapm.command.edit_schedule(manager: Manager, name: str, sources: List[str] | None, default_size_policy: str | None, autoindex: bool, calendarspec: str | CalendarSpec | None, policy: GcPolicy | None, boot: bool | None = None, revert: bool | None = None)

Edit an existing schedule from a list of mount point and block device source paths.

Parameters:

• manager – The manager context to use.

• name – The name of the schedule to edit.

• sources – A list of mount point or block devices to snapshot.

• default_size_policy – The default size policy for this schedule.

• autoindex – Treat name as the basename of a recurring scheduled snapshot set and generate and append an appropriate index value.

• calendarspec – The calendar trigger expression for this schedule.

• policy – The garbage collection policy for this schedule.

• boot – Create a boot entry for snapshot sets created by this schedule.

• revert – Create a revert boot entry for snapshot sets created by this schedule.

snapm.command.enable_schedule(manager: Manager, name: str, start: bool)

Enable an existing schedule. This enables the systemd timer units associated with this schedule.

Parameters:

• manager – The manager context to use.

• name – The name of the schedule to enable.

snapm.command.gc_schedule(manager: Manager, name: str) → List[str]

Run garbage collection for an existing schedule. This executes the configured garbage collection policy for the schedule named name.

Parameters:

• manager – The manager context to use.

• name – The name of the schedule to run garbage collection for.

snapm.command.main(args)

Main entry point for snapm.

snapm.command.print_diffs(_manager, selection=None, output_fields=None, opts=None, sort_keys=None, data: FsDiffResults | None = None)

Print diffs matching selection criteria.

Format a set of snapm.fsdiff.FsDiffRecord objects matching the given criteria, and output them as a report to the file given in opts.report_file.

Selection criteria may be expressed via a Selection object passed to the call using the selection parameter.

Parameters:

• _manager (Manager) – Manager context (unused).

• selection (Selection) – A Selection object giving selection criteria for the operation (unused).

• output_fields – a comma-separated list of output fields

• opts – output formatting and control options

• sort_keys – a comma-separated list of sort keys

• data (Optional[FsDiffResults]) – Pre-computed data for diff report.

snapm.command.print_plugins(manager, selection=None, output_fields=None, opts=None, sort_keys=None, data: object = None)

Print plugins matching selection criteria.

Format a set of snapm.manager.Plugin objects matching the given criteria, and output them as a report to the file given in opts.report_file.

Selection criteria are currently ignored for plugin reports.

Parameters:

• selection – A Selection object giving selection criteria for the operation

• output_fields – a comma-separated list of output fields

• opts – output formatting and control options

• sort_keys – a comma-separated list of sort keys

• data (NoneType) – Optional precomputed data for reporting (currently unused for this report type).

snapm.command.print_schedules(manager, selection: None | Selection = None, output_fields: None | str = None, opts: None | ReportOpts = None, sort_keys: [None, <class 'str'>]=None, data: object = None)

Print schedules matching selection criteria.

Format a set of snapm.manager.Schedule objects matching the given criteria, and output them as a report to the file given in opts.report_file.

Parameters:

• selection (Selection) – A Selection object giving selection criteria for the operation.

• output_fields (str) – A comma-separated list of output fields.

• opts (ReportOpts) – Output formatting and control options.

• sort_keys (str) – A comma-separated list of sort keys.

• data (NoneType) – Optional precomputed data for reporting (currently unused for this report type).

snapm.command.print_snapsets(manager, selection=None, output_fields=None, opts=None, sort_keys=None, data: object = None)

Print snapshot sets matching selection criteria.

Format a set of snapm.manager.SnapshotSet objects matching the given criteria, and output them as a report to the file given in opts.report_file.

Selection criteria may be expressed via a Selection object passed to the call using the selection parameter.

Parameters:

• selection – A Selection object giving selection criteria for the operation

• output_fields – a comma-separated list of output fields

• opts – output formatting and control options

• sort_keys – a comma-separated list of sort keys

• data (NoneType) – Optional precomputed data for reporting (currently unused for this report type).

snapm.command.print_snapshots(manager, selection=None, output_fields=None, opts=None, sort_keys=None, data: object = None)

Print snapshots matching selection criteria.

Format a set of snapm.manager.Snapshot objects matching the given criteria, and output them as a report to the file given in opts.report_file.

Selection criteria may be expressed via a Selection object passed to the call using the selection parameter.

Parameters:

• selection – A Selection object giving selection criteria for the operation

• output_fields – a comma-separated list of output fields

• opts – output formatting and control options

• sort_keys – a comma-separated list of sort keys

• data (NoneType) – Optional precomputed data for reporting (currently unused for this report type).

snapm.command.prune_snapset(manager, name, sources)

Prune snapshots from an existing snapshot set.

Remove snapshots from an existing snapshot set named name. The snapshot sources listed in sources are pruned (deleted) from the named snapshot set.

Parameters:

• name – The name of the snapshot set to prune.

• sources – The sources to prune from name.

Returns:

A SnapshotSet object representing the pruned snapshot set

snapm.command.rename_snapset(manager, old_name, new_name)

Rename a snapshot set from old_name to new_name.

snapm.command.resize_snapset(manager, sources, name=None, uuid=None, default_size_policy=None)

Resize snapshot set by name or UUID.

Parameters:

• manager – The manager context to use

• sources – A list of mount point or block devices to snapshot

• name – The name of the snapshot set to resize.

• uuid – The uuid of the snapshot set to resize.

• default_size_policy – The default size policy for this snapshot set.

snapm.command.revert_snapset(manager, name=None, uuid=None)

Revert snapshot set by name or UUID.

Parameters:

• manager – The manager context to use

• name – The name of the snapshot set to revert

• uuid – The UUID of the snapshot set to revert

snapm.command.set_debug(debug_arg)

Set debugging mask from command line argument.

snapm.command.setup_logging(cmd_args)

Set up snapm logging.

snapm.command.show_schedules(manager, selection=None, _members=False, json=False)

Show schedules matching selection criteria.

snapm.command.show_snapsets(manager, selection=None, members=False, json=False)

Show snapshot sets matching selection criteria.

snapm.command.show_snapshots(manager, selection=None, json=False)

Show snapshots matching selection criteria.

snapm.command.shutdown_logging()

Shut down snapm logging.

snapm.command.split_snapset(manager, name, new_name, sources)

Split an existing snapshot set.

Split the snapshot set named name into two snapshot sets, including all sources listed in sources in the newly created snapshot set with name new_name.

Parameters:

• name – The name of the snapshot set to split.

• new_name – The name for the newly created snapshot set.

• sources – The sources to split from name to new_name.

Returns:

A SnapshotSet object representing the snapshot set named new_name.

snapm.progress module

Terminal control and progress indicator

snapm.progress.DEFAULT_FPS = 10

Default frames-per-second for Throbber classes

class snapm.progress.NullProgress(register: bool = True)

Bases: ProgressBase

A progress class that produces no output.

class snapm.progress.NullThrobber(header, register: bool = True)

Bases: ThrobberBase

A throbber class that produces no output.

end(message: str | None = None)

Silent end for NullThrobber.

throb()

Silent throb for NullThrobber.

class snapm.progress.Progress(header, register: bool = True, term_stream: TextIO | None = None, width: int | None = None, width_frac: float | None = None, no_clear: bool = False, tc: TermControl | None = None)

Bases: ProgressBase

A 2-line Unicode or ASCII progress bar, which looks like:

Header: 20% [███████████░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░]

progress message

or:

Header: 20% [===========———————————-]

progress message

The progress bar is colored, if the terminal supports color output and adjusts to the width of the terminal.

BAR = '${BOLD}${CYAN}%s${NORMAL}: %3d%% ${GREEN}[${BOLD}%s%s${NORMAL}${GREEN}]${NORMAL}\n'

Progress bar format string

FIXED = 9

Length of fixed characters in BAR.

__init__(header, register: bool = True, term_stream: TextIO | None = None, width: int | None = None, width_frac: float | None = None, no_clear: bool = False, tc: TermControl | None = None)

Initialise a two-line terminal progress renderer.

Parameters:

• header (str) – The progress header to display.

• register (bool) – Register this Progress for log callbacks.

• term_stream (TextIO) – The terminal stream to write to.

• width (int) – An optional width value in characters. If specified the progress bar will occupy this width. Cannot be used with width_frac.

• width_frac (Optional[float]) – An optional width fraction between [0..1]. If specified the terminal width is multiplied by this value and rounded to the nearest integer to determine the width of the progress bar. Cannot be used with width.

• no_clear (bool) – For progress indicators that clear and re-draw the content on progress, do not erase the progress bar when ProgressBase.end() is called, leaving it on the terminal for the user to refer to. Ignored by ProgressBase child classes that do not use TerminalControl.

• tc (Optional[TermControl]) – An optional TermControl object already initialised with a term_stream value. If this argument is set it will override any term_stream argument.

Raises:

ValueError – If terminal lacks required capabilities.

budget: int

did_erase: bool

first_update: bool

fps: int

header: str | None

pbar: str | None

skip_line: bool

stream: TextIO | None

term: TermControl | None

width: int

class snapm.progress.ProgressBase(register: bool = True)

Bases: ABC

An abstract progress reporting class.

FIXED = -1

__init__(register: bool = True)

Initialize base progress state.

Sets default state for lifecycle and rendering configuration.

Parameters:

register (bool) – Register this ProgressBase for log callbacks.

cancel(message: str | None = None)

End the progress run with error and finalize the display.

Parameters:

message (Optional[str]) – An optional error message.

end(message: str | None = None)

End the progress run and finalize the display.

Parameters:

message (Optional[str]) – An optional completion message.

first_update: bool

header: str | None

progress(done: int, message: str | None = None)

Advance the progress indicator to the specified done count.

Parameters:

• done (int) – The number of completed progress items.

• message (Optional[str]) – An optional progress message.

register: bool = <bound method ABCMeta.register of <class 'snapm.progress.ProgressBase'>>

registered: bool

reset_position()

Mark progress bar as displaced by external output.

skip_line: bool

start(total: int)

Begin a progress run with the specified total.

Parameters:

total (int) – The total number of expected progress items.

stream: TextIO | None

term: TermControl | None

total: int

width: int

class snapm.progress.ProgressFactory

Bases: object

A factory for constructing progress objects.

static get_progress(header: str, quiet: bool = False, term_stream: TextIO | None = None, term_control: TermControl | None = None, width: int | None = None, width_frac: float | None = None, no_clear: bool = False, register: bool = True) → ProgressBase

Return an appropriate ProgressBase implementation.

Parameters:

• header (str) – The progress report header.

• quiet (bool) – Suppress all output.

• term_stream (Optional[TextIO]) – An optional TextIO output object. Defaults to sys.stdout if unspecified.

• term_control (Optional[TermControl]) – An optional TermControl object to use for the progress report.

• width (int) – An optional width value in characters. If specified the progress bar will occupy this width. Cannot be used with width_frac.

• width_frac (Optional[float]) – An optional width fraction between [0..1]. If specified the terminal width is multiplied by this value and rounded to the nearest integer to determine the width of the progress bar. Cannot be used with width.

• no_clear (bool) – For progress indicators that clear and re-draw the content on progress, do not erase the progress bar when ProgressBase.end() is called, leaving it on the terminal for the user to refer to. Ignored by ProgressBase child classes that do not use TerminalControl.

• register (bool) – Register the new object with the log system for notification callbacks.

Returns:

An appropriate progress implementation.

Return type:

ProgressBase

static get_throbber(header: str, quiet: bool = False, style: str | None = None, term_stream: TextIO | None = None, term_control: TermControl | None = None, no_clear: bool = False, register: bool = True) → ThrobberBase

Return an appropriate ThrobberBase implementation.

Parameters:

• header (str) – The throbber header.

• quiet (bool) – Suppress all output.

• style (Optional[str]) – An optional Unicode throbber style string. Ignored if the terminal does not support Unicode encoding.

• term_stream (Optional[TextIO]) – An optional TextIO output object. Defaults to sys.stdout if unspecified.

• term_control (Optional[TermControl]) – An optional TermControl object to use for the throbber.

• no_clear (bool) – For throbber indicators that clear and re-draw the content on throbber, do not erase the throbber bar when ThrobberBase.end() is called, leaving it on the terminal for the user to refer to. Ignored by ThrobberBase child classes that do not use TerminalControl.

• register (bool:) – Register the new object with the log system for notification callbacks.

Returns:

An appropriate throbber implementation.

Return type:

ThrobberBase

static get_throbber_styles() → List[str]

Return a list of known Throbber style strings.

Returns:

A list of throbber styles.

Return type:

List[str]

class snapm.progress.SimpleProgress(header, register: bool = True, term_stream: TextIO | None = None, width: int | None = None, width_frac: float | None = None)

Bases: ProgressBase

A simple progress bar that does not rely on terminal capabilities.

BAR = '%s: %3d%% [%s%s] (%s)'

Progress bar format string

DID = '='

Bar character for completed work.

FIXED = 12

Length of fixed characters in BAR.

TODO = '-'

Bar character for uncompleted work.

__init__(header, register: bool = True, term_stream: TextIO | None = None, width: int | None = None, width_frac: float | None = None)

Initialise a new SimpleProgress object.

Parameters:

• header (str) – The progress header to display.

• register (bool) – Register this SimpleProgress for log callbacks.

• term_stream (Optional[TextIO]) – The terminal stream to write to.

• width (int) – An optional width value in characters. If specified the progress bar will occupy this width. Cannot be used with width_frac.

• width_frac (Optional[float]) – An optional width fraction between [0..1]. If specified the terminal width is multiplied by this value and rounded to the nearest integer to determine the width of the progress bar. Cannot be used with width.

header: str | None

stream: TextIO | None

width: int

class snapm.progress.SimpleThrobber(header: str, register: bool = True, term_stream: TextIO | None = None)

Bases: ThrobberBase

A simple throbber that does not rely on terminal capabilities.

__init__(header: str, register: bool = True, term_stream: TextIO | None = None)

Initialise a simple ascii throbber.

Parameters:

• header (str) – The throbber header.

• register (bool) – Register this SimpleThrobber for log callbacks.

• term_stream (Optional[TextIO]) – The terminal stream to write to.

class snapm.progress.TermControl(term_stream: TextIO | None = None, color: str = 'auto')

Bases: object

A class for portable terminal control and output.

Uses the curses package to set up appropriate terminal control sequences for the current terminal.

Inspired by and adapted from:

https://code.activestate.com/recipes/475116-using-terminfo-for-portable-color-output-cursor-co/

Copyright Edward Loper and released under the PSF license.

TermControl defines a set of instance variables whose values are initialized to the control sequence necessary to perform a given action. These can be simply included in normal output to the terminal:

>>> term = TermControl()
>>> print 'This is '+term.GREEN+'green'+term.NORMAL

Alternatively, the render() method can used, which replaces ‘${action}’ with the string required to perform ‘action’:

>>> term = TermControl()
>>> print term.render('This is ${GREEN}green${NORMAL}')

If the terminal doesn’t support a given action, then the value of the corresponding instance variable will be set to ‘’. As a result, the above code will still work on terminals that do not support color, except that their output will not be colored. Also, this means that you can test whether the terminal supports a given action by simply testing the truth value of the corresponding instance variable:

>>> term = TermControl()
>>> if term.CLEAR_SCREEN:
...     print 'This terminal supports clearning the screen.'

Finally, if the width and height of the terminal are known, then they will be stored in the columns and lines attributes.

BELL: str = ''

Ring the terminal bell

BG_BLACK: str = ''

Black background color

BG_BLUE: str = ''

Blue background color

BG_CYAN: str = ''

Cyan background color

BG_GREEN: str = ''

Green background color

BG_MAGENTA: str = ''

Magenta background color

BG_RED: str = ''

Red background color

BG_WHITE: str = ''

White background color

BG_YELLOW: str = ''

Yellow background color

BLACK: str = ''

Black foreground color

BLINK: str = ''

Turn on blink mode

BLUE: str = ''

Blue foreground color

BOL: str = ''

Move the cursor to the beginning of the line

BOLD: str = ''

Turn on bold mode

CLEAR_BOL: str = ''

Clear to the beginning of the line.

CLEAR_EOL: str = ''

Clear to the end of the line.

CLEAR_EOS: str = ''

Clear to the end of the screen

CLEAR_SCREEN: str = ''

Clear the screen and move to home position

CYAN: str = ''

Cyan foreground color

DIM: str = ''

Turn on half-bright mode

DOWN: str = ''

Move the cursor down one line

GREEN: str = ''

Green foreground color

HIDE_CURSOR: str = ''

Make the cursor invisible

LEFT: str = ''

Move the cursor left one char

MAGENTA: str = ''

Magenta foreground color

NORMAL: str = ''

Turn off all modes

RED: str = ''

Red foreground color

REVERSE: str = ''

Turn on reverse-video mode

RIGHT: str = ''

Move the cursor right one char

SHOW_CURSOR: str = ''

Make the cursor visible

UP: str = ''

Move the cursor up one line

WHITE: str = ''

White foreground color

YELLOW: str = ''

Yellow foreground color

__init__(term_stream: TextIO | None = None, color: str = 'auto')

Initialize terminal capabilities and size information.

If the output stream is not a tty or terminal setup fails, the instance will have no terminal capabilities (all control attributes remain empty strings or None).

Parameters:

• term_stream (Optional[TextIO]) – Output stream to probe for capabilities.

• color (str) – A string to control color rendering: “auto”, “always”, or “never”.

columns: int | None = None

Terminal width

lines: int | None = None

Terminal height

render(template)

Replace each $-substitution with the corresponding control.

Replace each $-substitution in the template string with the corresponding terminal control string (if defined) or an empty string (if not defined).

Parameters:

template (str) – Template string containing ${NAME} patterns.

Returns:

Rendered string with substitutions applied.

Return type:

str

class snapm.progress.Throbber(header: str, register: bool = True, style: str | None = None, term_stream: TextIO | None = None, no_clear: bool = False, tc: TermControl | None = None)

Bases: ThrobberBase

A ThrobberBase subclass to display a one line Unicode or ASCII throbber for capable terminals.

STYLES: ClassVar[Dict[str, str | List[str]]] = {'arrowspinner': '←↖↑↗→↘↓↙', 'ascii': '-\\|/', 'bouncingball': ['[●     ]', '[ ●    ]', '[  ●   ]', '[   ●  ]', '[    ● ]', '[     ●]', '[    ● ]', '[   ●  ]', '[  ●   ]', '[ ●    ]', '[●     ]'], 'bouncingbar': ['[|     ]', '[ |    ]', '[  |   ]', '[   |  ]', '[    | ]', '[     |]', '[    | ]', '[   |  ]', '[  |   ]', '[ |    ]', '[|     ]'], 'braillecircle': ['⢎⡰', '⢎⡡', '⢎⡑', '⢎⠱', '⠎⡱', '⢊⡱', '⢌⡱', '⢆⡱'], 'braillewave': '⣾⣽⣻⢿⡿⣟⣯⣷', 'horizontalbars': '█▉▊▋▌▍▎▏▎▍▌▋▊▉█', 'verticalbars': '▁▂▃▄▅▆▇█▇▆▅▄▃▂▁', 'wave': '⠁⠂⠄⡀⢀⠠⠐⠈'}

__init__(header: str, register: bool = True, style: str | None = None, term_stream: TextIO | None = None, no_clear: bool = False, tc: TermControl | None = None)

Initialise a new one line throbber instance.

Parameters:

• header (str) – The header string to print.

• register (bool) – Register this Throbber for log callbacks.

• style (Optional[str]) – A Unicode throbber style string. Ignored if the terminal does not support Unicode encoding. Defaults to “wave” if unset.

• term_stream (TextIO) – The terminal stream to write to.

• no_clear (bool) – For throbber indicators that clear and re-draw the content on progress, do not erase the throbber frame when ThrobberBase.end() is called, leaving it on the terminal for the user to refer to. Ignored by ThrobberBase child classes that do not use TerminalControl.

• tc (Optional[TermControl]) – An optional TermControl object already initialised with a term_stream value. If this argument is set it will override any term_stream argument.

Raises:

ValueError – If terminal lacks required capabilities.

stream: TextIO | None

term: TermControl | None

class snapm.progress.ThrobberBase(header, register: bool = True)

Bases: ABC

An abstract busy indicator class. Unlike ProgressBase classes the throbber reports progress of a time consuming task where the total number of items is unknown (for instance, generating tree walk lists).

__init__(header, register: bool = True)

Initialize base throbber state.

Sets default state for lifecycle and rendering configuration.

Parameters:

register (bool) – Register this ThrobberBase for log callbacks.

end(message: str | None = None)

End the throbber run and finalize the display.

Parameters:

message (Optional[str]) – An optional completion message.

first_update: bool

fps: int

frames: str

header: str

nr_frames: int

register: bool = <bound method ABCMeta.register of <class 'snapm.progress.ThrobberBase'>>

registered: bool

reset_position()

Mark throbber as displaced by external output.

start()

Begin a throbber run.

started: bool

stream: TextIO | None

term: TermControl | None

throb()

Maintain liveness for this throbber and output frame if required.

snapm.report module

The snapm reporting module contains a set of classes for creating simple text based tabular reports for a user-defined set of object types and fields. No restrictions are placed on the types of object that can be reported: users of the Report classes may define additional object types outside the snapm package and include these types in reports generated by the module.

The fields displayed in a specific report may be selected from the available set of fields by specifying a simple comma-separated string list of field names (in display order). In addition, custom multi-column sorting is possible using a similar string notation.

The Report module is closely based on the device-mapper reporting engine and shares many features and behaviours with device mapper reports.

class snapm.report.Field

Bases: object

A Field represents an instance of a FieldType including its associated data values.

__init__(report, props)

Initialise a new Field object.

Initialise a Field object and configure the supplied report and props attributes.

Parameters:

• report – The Report that owns this field

• props – The FieldProperties object for this field

report_idx(value)

Report an index value for this Field object.

Set the value for this field to the supplied value.

Parameters:

value – The index value to set

Return type:

None

report_num(value)

Report a numeric value for this Field object.

Set the value for this field to the supplied value.

Parameters:

value – The numeric value to set

Return type:

None

report_sha(value)

Report a SHA value for this Field object.

Set the value for this field to the supplied value.

Parameters:

value – The SHA value to set

Return type:

None

report_size(value)

Report a size value for this Field object.

Set the value for this field to the supplied value, converted to a human readable string.

Parameters:

value – The size value to set in bytes.

Return type:

None

report_str(value)

Report a string value for this Field object.

Set the value for this field to the supplied value.

Parameters:

value – The string value to set

Return type:

None

report_str_list(value, nosort=False)

Report a string list value for this Field object.

Set the value for this field to the strings contained in value.

Parameters:

• value – A list of strings

• nosort (bool) – Disable sorting of the provided string list.

Return type:

None

report_time(value)

Report a time value for this Field object.

Set the value for this field to the supplied value.

Parameters:

value – The time value to set

Return type:

None

report_uuid(value)

Report a uuid value for this Field object.

Set the value for this field to the supplied value.

Parameters:

value – The uuid value to set

Return type:

None

set_value(report_string, sort_value=None)

Report an arbitrary value for this Field object.

Set the value for this field to the supplied value, and set the field’s sort_value to the supplied sort_value.

Parameters:

• report_string – The string value to set

• sort_value – The sort value

Return type:

None

class snapm.report.FieldProperties(field_num=None, initial_width=0, width=0, objtype=None, dtype=None, align=None, hidden=False, implicit=False, sort_key=False, sort_dir=None, compact_one=False, compacted=False, sort_posn=None)

Bases: object

Properties of a field instance.

__init__(field_num=None, initial_width=0, width=0, objtype=None, dtype=None, align=None, hidden=False, implicit=False, sort_key=False, sort_dir=None, compact_one=False, compacted=False, sort_posn=None)

class snapm.report.FieldType

Bases: object

The FieldType class describes the properties of a field available in a Report instance.

__init__(objtype, name, head, desc, width, dtype, report_fn, align=None)

Initialise new FieldType object.

Initialise a new FieldType object with the specified properties.

Parameters:

• objtype – The numeric object type ID (power of two)

• name – The field name used to select display fields

• desc – A human-readable description of the field

• width – The default (initial) field width

• dtype – The Report data type of the field

• report_fn – The field reporting function

• align – The field alignment value

Returns:

A new ReportFieldType object

Return type:

ReportFieldType

class snapm.report.Report

Bases: object

A class representing a configurable text report with multiple caller-defined fields. An optional title may be provided and the fields argument must contain a list of Field objects describing the required report.

__init__(types, fields, output_fields, opts, sort_keys, title)

Initialise Report.

Initialise a new Report object with the specified fields and output control options.

Parameters:

• types – List of ReportObjType used in this report.

• fields – A list of Field field descriptions.

• output_fields – An optional list of output fields to be rendered by this report.

• opts – An instance of ReportOpts or None.

Returns:

A new report object.

Return type:

Report.

opts = None

report_object(obj)

Report data for object.

Add a row of data to this Report. The data argument should be an object of the type understood by this report’s fields. It will be passed in turn to each field to obtain data for the current row.

Parameters:

obj – the object to report on for this row.

report_output()

Output report data.

Output this report’s data to the configured report file, using the configured output controls and fields.

On success the number of rows output is returned. On error an exception is raised.

Returns:

the number of rows of output written.

Return type:

int

class snapm.report.ReportObjType

Bases: object

Class representing a type of object to be reported on. Instances of ReportObjType must specify an identifier, a description, and a data function that will return the correct type of object from a compound object containing data objects of different types. For reports that use only a single object type the data_fn member may be simply lambda x: x.

__init__(objtype, desc, prefix, data_fn)

Initialise ReportObjType.

Initialise a new ReportObjType object with the specified objtype, desc, optional prefix and data_fn. The objtype must be an integer power of two that is unique within a given report. The data_fn should accept an object as its only argument and return an object of the requested type.

class snapm.report.ReportOpts

Bases: object

Options controlling the formatting and output of a report.

__init__(columns=80, headings=True, buffered=True, separator=' ', field_name_prefix='', unquoted=True, aligned=True, json=False, columns_as_rows=False, report_file=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Initialise ReportOpts object.

Initialise a ReportOpts object to control output of a Report.

Parameters:

• columns – the number of columns to use for output.

• headings – a boolean indicating whether to output column headings for this report.

• buffered – a boolean indicating whether to buffer output from this report.

• report_file – a file to which output will be sent.

Returns:

a new ReportOpts object.

Return type:

<class ReportOpts>

__str__()

Return str(self).

snapm.manager._boot module

Boot integration for snapshot manager

class snapm.manager._boot.BootCache

Bases: object

Set of caches mapping snapshot sets to boot entries and revert entries.

__init__()

refresh_cache()

Refresh the cache of boot entry mappings held by this BootCache instance.

class snapm.manager._boot.BootEntryCache(entry_arg)

Bases: dict

Cache mapping snapshot sets to boom BootEntry instances.

Boot entries in the cache are either snapshot set boot entries or revert entries depending on the value of entry_arg.

__init__(entry_arg)

_parse_entry(boot_entry)

Parse a boom BootEntry options string and return the value of the snapm.snapset argument if present.

Parameters:

boot_entry – The boot entry to process.

Returns:

The snapset name for the boot entry.

refresh_cache()

Generate mappings from snapshot sets to boot entries.

snapm.manager._boot.REVERT_ARG = 'snapm.revert'

Snapshot set revert kernel command line argument

snapm.manager._boot.SNAPSET_ARG = 'snapm.snapset'

Snapshot set kernel command line argument

snapm.manager._boot._DBUS_MACHINE_ID = '/var/lib/dbus/machine-id'

Path to the legacy system machine-id file

snapm.manager._boot._DEV_PREFIX = '/dev/'

/dev path prefix

snapm.manager._boot._MACHINE_ID = '/etc/machine-id'

Path to the system machine-id file

snapm.manager._boot._OS_RELEASE = '/etc/os-release'

Path to the system os-release file

snapm.manager._boot._build_swap_list(fstab: FsTabReader) → List[str]

Build a list of command line swap unit definitions for the running system. Swap entries are extracted from /etc/fstab and returned as a list of “WHAT:OPTIONS” strings.

snapm.manager._boot._create_boom_boot_entry(version, title, tag_arg, root_device, mounts=None, swaps=None)

Create a boom boot entry according to the passed arguments.

Parameters:

• version – The UTS release name for the boot entry

• title – The title for the boot entry.

• tag_arg – A tag argument to be added to the kernel command line and used to associate the entry with a snapshot set name or UUID.

• root_device – The root device for the entry. Passed to root=…

• mounts – An optional list of mount specifications to use for the boot entry. If defined fstab=no will be appended to the generated kernel command line.

• swaps – An optional list of swap specifications to use for the boot entry. Each entry should be in the format “WHAT:OPTIONS”.

snapm.manager._boot._create_default_os_profile()

Create a default boom OsProfile for the running system. This uses the boom API to run the equivalent of: boom profile create --from-host.

snapm.manager._boot._delete_boot_entry(boot_id)

Delete a boom boot entry by ID.

Parameters:

boot_id – The boot identifier to delete.

snapm.manager._boot._escape(orig: str) → str

Convert literal ‘:’ characters into the hexadecimal escape (:): systemd will decode these when reading systemd.{mount,swap}-extra values.

Parameters:

orig – The original string possibly containing literal ‘:’ characters.

Returns:

An escaped string with ‘:’ replaced by ‘:’.

snapm.manager._boot._get_machine_id() → str | None

Return the current host’s machine-id.

Get the machine-id value for the running system by reading from /etc/machine-id and return it as a string.

Returns:

The machine_id as a string, or None if not available.

Return type:

Optional[str]

snapm.manager._boot._get_uts_release()

Return the UTS release (kernel version) of the running system.

Returns:

A string representation of the UTS release value.

snapm.manager._boot._mount_list_to_units(mount_list: List[Tuple[str, str, str, str]]) → List[str]

Transform a list of (what, where, fstype, options) tuples into systemd command line mount unit notation.

Parameters:

mount_list – The list of mount tuples to transform.

Returns:

A list of systemd.mount-extra= argument strings.

Return type:

List[str]

snapm.manager._boot.check_boom_config()

Check for boom configuration and create the default config if not found.

snapm.manager._boot.create_snapset_boot_entry(snapset, title=None)

Create a boom boot entry to boot into the snapshot set represented by snapset.

Parameters:

• snapset – The snapshot set for which to create a boot entry.

• title – An optional title for the boot entry. If title is None the boot entry will be titled as “Snapshot snapset_name snapset_time”.

snapm.manager._boot.create_snapset_revert_entry(snapset, title=None)

Create a boom boot entry to revert the snapshot set represented by snapset.

Parameters:

• snapset – The snapshot set for which to create a revert entry.

• title – An optional title for the revert entry. If title is None the revert entry will be titled as “Revert snapset_name snapset_time”.

snapm.manager._boot.delete_snapset_boot_entry(snapset)

Delete the boot entry corresponding to snapset.

Parameters:

snapset – The snapshot set for which to remove a boot entry.

snapm.manager._boot.delete_snapset_revert_entry(snapset)

Delete the revert entry corresponding to snapset.

Parameters:

snapset – The snapshot set for which to remove a revert entry.

snapm.manager._calendar module

Calendar event abstraction for Snapshot Manager

class snapm.manager._calendar.CalendarSpec(calendarspec: str)

Bases: object

Class representing systemd CalendarSpec expressions.

__init__(calendarspec: str)

Validate and parse a systemd calendarspec expression into a CalendarSpec object.

Parameters:

calendarspec – A string containing an calendarspec expression.

Raises:

ValueError if calendarspec is not a valid calendarspec expression.

__repr__()

Return a string representation of this CalendarSpec instance in the form of a call to the CalendarSpec initializer.

__str__()

Return a string representation of this CalendarSpec instance.

_cond_refresh()

Refresh this CalendarSpec object’s time-dependent properties if the current next_elapse value is in the past.

_refresh()

Refresh this CalendarSpec object’s time-dependent properties.

property from_now

Return a string representation of the time remaining until this CalendarSpec next elapses.

Returns:

The time remaining until the next elapse as a string.

Return type:

str

property in_utc

Return the next elapse time for this CalendarSpec object as an instance of datetime.datetime in UTC.

Returns:

The next elapse time as a UTC datetime object.

Return type:

datetime.datetime

property next_elapse

Return the next elapse time for this CalendarSpec object as an instance of datetime.datetime.

Returns:

The next elapse time as a datetime object.

Return type:

datetime.datetime

property occurs

True if this CalendarSpec object’s calendar expression will occur in the future, or False otherwise.

property original

The original form of this CalendarSpec as a string.

snapm.manager._mounts module

Mount integration for snapshot manager

class snapm.manager._mounts.Mount(snapset: SnapshotSet, mount_root: str, discover: bool = False, root_device: str | None = None)

Bases: MountBase

Representation of a mounted snapshot set, including snapshot set mounts, non-snapshot mounts from /etc/fstab, and bind mounts for API file systems.

TYPE = 'Snapshot set'

Description of the type of mount object.

__init__(snapset: SnapshotSet, mount_root: str, discover: bool = False, root_device: str | None = None)

Initialize a new Mount instance for the given snapshot set. If discover is True validate that the mount_root is an existing mount point for the snapshot set and check for the presence of expected submounts.

Parameters:

• snapset (SnapshotSet) – The snapshot set to mount or represent.

• mount_root (str) – The absolute path to the root directory where the snapshot set will be mounted or is already mounted.

• discover (bool) – If True, validate that mount_root is an existing mount point and discover its submounts. If False, prepare a new mount object without validation.

• root_device (Optional[str]) – The device to mount at mount_root. One is looked up with find_snapset_root() when it is None.

Raises:

SnapmPathError – If mount_root is not a directory, or if discover is True and mount_root is not a mount point.

_abc_impl = <_abc._abc_data object>

_bind_api_filesystems(options: str)

Recursively bind mount the host’s API file systems into this mount.

Parameters:

options (str) – Mount options to apply to each bind mount.

_chroot()

Return a chrooted command list for this snapshot set mount.

Returns:

A command list including an appropriate chroot invocation if necessary.

Return type:

List[str]

_do_mount()

Mount the configured snapshot set and its submounts at self.root.

_do_umount()

Unmount the configured snapshot set and its submounts from self.root.

_rollback_mount(err: Exception)

Unmount whatever a failed mount operation left behind.

Parameters:

err (Exception) – The error that failed the mount operation.

class snapm.manager._mounts.Mounts(manager, mounts_dir: str, state_dir: str, lock_dir: str)

Bases: object

A high-level interface for mounting, unmounting and enumerating snapshot set mounts in the snapm runtime directory (normally /run/snapm/mounts).

__init__(manager, mounts_dir: str, state_dir: str, lock_dir: str)

Initialise a new Mounts object.

Parameters:

• manager – The Manager object that this instance belongs to.

• mounts_dir – The directory under which mounts are managed.

• state_dir – The directory holding the mount provenance records (#1025).

• lock_dir – The directory holding the lock that serialises changes to the mount state.

_canonical_mount(snapset: SnapshotSet) → Mount | None

Return the canonical mount for snapset.

A snapshot set may be mounted at more than one path at a time: the canonical mount is the first one discovered or created, and is the mount reported as the set’s mount root. Entries describing a path that is no longer mounted, or one the records no longer account for, are dropped by _forget_stale_mounts() and _recorded_mounts(), which every caller runs first.

Parameters:

snapset – The snapshot set to look up.

Returns:

A Mount object, or None if the set is not mounted.

_check_mount_point(path: str) → str

Validate a mount point nominated in place of the default and return the resolved path to mount the snapshot set at.

A symbolic link in a leading component is resolved: mount(8) follows it, and the kernel then reports the resolved path, so a mount rooted at an unresolved one can never be found again (#1029). A link as the final component is refused rather than resolved, since that is the component nominating the mount point.

Parameters:

path – The mount point to validate.

Returns:

The resolved mount point path.

Raises:

SnapmPathError – If path is within the mounts directory, is not an existing directory, or is a symbolic link.

_check_mounts_dir() → None

Check the mounts directory before creating a mount point in it.

Mount points created here are verified by name and then mounted by name, so the directory holding them has to be one that no other user can write to.

Raises:

SnapmPathError – If the mounts directory is not a directory owned by this process and writable by its owner alone.

_default_mount_path(snapset: SnapshotSet) → str

Return the path at which snapset is mounted beneath the default mounts directory.

Parameters:

snapset – The snapshot set to build a mount path for.

Returns:

The absolute mount path for the snapshot set.

Raises:

SnapmPathError – If the set name does not name a path directly beneath the mounts directory.

static _device_matches(record: MountStateRecord, entry: MountInfoEntry, mountinfo: MountInfoReader) → bool

Test whether the mount entry is a mount of the device that record was written for.

The device number recorded once the mount was made is the one the kernel reports for the mount, so it is compared with the mount table directly. A record with none was written by a command that never reached the end of its mount, and names only the device it set out to mount: ask the mount table whether that device is the one mounted there, which it answers through its index whether or not the file system has an anonymous superblock.

Parameters:

• record – The mount provenance record to verify.

• entry – The mount table entry for the recorded path.

• mountinfo – A MountInfoReader instance to use.

Returns:

True if the recorded device is the one mounted.

_discard_record(record: MountStateRecord) → None

Discard the record for a mount that is no longer present, removing the mount point with it if it was snapm’s to remove.

Parameters:

record – The mount provenance record to discard.

_forget_mounts(mounts: List[Mount]) → None

Drop mounts from the mount table in memory and recompute the mount root reported for each set that loses one.

This is the whole of forgetting a mount: an entry removed without the second step leaves its set reported as mounted at a path the object no longer believes in. Both notions of a mount worth forgetting, one with nothing mounted at it and one the records no longer account for, come through here so that the two steps cannot come apart.

Nothing here leaves this object. Removing what a dropped entry left on disk is _remove_stale_mounts()’s to do, and needs the lock.

Parameters:

mounts – The entries to drop. Each must be in the mount table: a caller selects them from it.

_forget_stale_mounts(snapset: SnapshotSet | None = None) → List[Mount]

Drop the mount table entries that no longer describe a mount, and return them.

A snapshot set unmounted outside snapm, or one whose mount point could not be removed when it was unmounted, leaves an entry naming a path with nothing mounted on it. Until it is dropped the set looks mounted there: the path is reported as the set’s mount root, it blocks mounting the set there again, and it counts towards the ambiguity check in umount().

Nothing here leaves this object: the entries are forgotten in memory, which follows from the mount table as it is now. Removing what the dropped entries left on disk is _remove_stale_mounts()’s to do, and needs the lock.

Parameters:

snapset – Optional snapshot set to limit this to. All snapshot sets are considered when it is None.

Returns:

The entries that were dropped, in mount table order.

_lock_timeout

How long to wait for the mount state lock. Overridden by the tests, which have no reason to wait out the default.

_locked_mount(snapset: SnapshotSet, mount_root: str | None = None) → Mount

Mount the snapshot set snapset with the mount state lock held.

Parameters:

• snapset – The snapshot set to operate on.

• mount_root – Optional path to mount the snapshot set at.

_locked_umount(snapset: SnapshotSet, mount_root: str | None = None)

Unmount the snapshot set snapset with the mount state lock held.

Parameters:

• snapset – The snapshot set to operate on.

• mount_root – Optional mount point selecting the mount to unmount.

static _mount_is_marked(mount_path: str, snapset: SnapshotSet) → bool

Test whether the mount just made at mount_path carries the option marking it as a mount of snapset made by snapm.

The option is passed to every mount in the tree, so a mount that does not have it is one whose userspace mount table entry was never written: mount(8) keeps no such table on this system, or could not write to it. Verification notes the difference rather than insisting, since requiring an option that is never recorded would make every mount here unrecognisable to the next command.

A table that could not be read says nothing either way, and what it would have said cannot be recovered later: the answer is what the record carries for the rest of its life. Fail rather than write the weaker form into it.

Parameters:

• mount_path – The mount point to read the options of.

• snapset – The snapshot set mounted there.

Returns:

True if the mount is marked as snapm’s.

Raises:

SnapmSystemError – If the userspace mount table exists and cannot be read.

_mount_is_recorded(mount: Mount, mountinfo: MountInfoReader, utab: UtabReader) → bool

Test whether the records still account for mount, logging the reason when they do not.

Parameters:

• mount – The candidate mount to test.

• mountinfo – A MountInfoReader instance to use.

• utab – A UtabReader instance to use.

Returns:

True if the records account for mount.

static _mounted_devno(mount_path: str) → int | None

Return the device number the kernel reports for the file system mounted at mount_path.

Parameters:

mount_path – The mount point to read the device number of.

Returns:

The device number of the mount at mount_path, or None if nothing is mounted there.

static _option_matches(record: MountStateRecord, utab: UtabReader) → bool

Test whether the mount at the path record names still carries the option marking it as the mount snapm made.

A device number identifies a file system, not one mount of it. A record outlives a mount taken down outside snapm until the next command that writes reaps it, so an administrator who mounts the same snapshot device at the same path in the meantime makes a mount that answers to the recorded device. The option tells the two apart: mount(8) records it with the mount and umount(8) takes it away with it, where the kernel’s mount ID is reused by the next mount at the same path.

Parameters:

• record – The mount provenance record to verify.

• utab – A UtabReader instance to use.

Returns:

True if the mount carries the recorded option.

_owns_mount_point(path: str) → bool

Test whether path is a mount point snapm created, and may therefore remove.

The answer is the one recorded when the set was mounted there: a default mount point in the mounts directory is snapm’s to remove, and a mount point nominated with --mount-root belongs to whoever made it. A path with no record is not snapm’s at all.

Parameters:

path – The mount point to test.

Returns:

True if snapm created path, or False otherwise.

_reap_stale_records() → None

Remove the mount state records that no longer describe a mount, along with the mount points snapm created for them.

Discovery leaves them alone, so this is where they go. A record is written before the mount it describes is made, which means a record for a path with nothing mounted on it is as likely to belong to a mount another process is making right now as to one that is over. Reaping from the commands that mount and unmount, rather than from discovery, keeps every read-only command out of the way of a mount in progress: a ‘snapset list’ or a scheduled garbage collection can no longer unlink the record of a mount that is halfway made, nor remove the directory it is about to be mounted on.

The caller holds the mount state lock, so no other command is mounting or unmounting while this runs.

Raises:

SnapmSystemError – If a record is verified by its mount mark and the userspace mount table exists and cannot be read.

static _record_is_live(record: MountStateRecord, mountinfo: MountInfoReader, utab: UtabReader) → bool

Test whether the mount described by record is still mounted.

The recorded mount path has to still be a mount point, the device mounted there has to still be the one recorded, and the mount has to still carry the option marking it as snapm’s. Checking more than the path distinguishes a snapshot set that is still mounted from a mount point that snapm’s mount was taken off and something else was then mounted on: unmounting the latter would take out a mount that is not snapm’s (#1025).

Each check demands what the record carries and no more. A record written by a command that did not reach the end of its mount has no mount device number, and one written on a system whose mount(8) keeps no userspace mount table has no option: requiring either of those unconditionally would make a mount that is really there unrecognisable to the next command, and orphan it.

Parameters:

• record – The mount provenance record to verify.

• mountinfo – A MountInfoReader instance to use.

• utab – A UtabReader instance to use.

Returns:

True if the recorded mount is still present.

_recorded_mounts(mounts: List[Mount]) → List[Mount]

Return those of mounts the records still account for.

An entry says what was mounted at a path when it was discovered, and nothing but the record says what is mounted there now: the path being a mount point only says that something is. By the time the entry is used another command may have unmounted it, and an administrator or another command may have mounted something else there, including a different snapshot set.

An entry is kept only when a record for its path exists, names the same snapshot set, and still describes what is mounted there. Anything else is a mount this object no longer knows the truth about: acting on one would unmount a file system that is not the one asked for, and reporting one would name a mount as a set’s that it is not (#1025).

An entry that is not kept is forgotten, as one with nothing mounted at it is. Filtering only the result would leave the set reported as mounted at a path that umount() then declines to act on, which is the same mount state described two ways. Nothing here leaves this object or touches the records, so this is as true of a command that only reports as of one holding the lock.

Parameters:

mounts – The candidate mounts to filter.

Returns:

Those of mounts the records account for, in order.

_refresh_state() → None

Re-read the mount state records.

The copy in memory was read when this object was built, which may be a long time ago for a caller that keeps a Manager around, and another command may have mounted or unmounted something since. What follows has to be decided from the records as they are now rather than as they were at discovery.

A command that is about to change the mount state does this with the lock held, so that what it read stays true for as long as it acts on it. A command that only reports does it without, and answers with what was true when it looked.

Raises:

SnapmPathError – If the record directory cannot be read.

_remove_stale_mounts(stale: List[Mount]) → None

Remove what the mounts dropped by _forget_stale_mounts() left behind: the record of each, and the mount point when snapm made it.

A nominated mount root is left alone: it belongs to whoever made it.

This writes to the mount state, so the caller holds the mount state lock.

Parameters:

stale – The dropped entries, as returned by _forget_stale_mounts().

static _reusable_mount_point(path: str) → bool

Test whether path is a leftover mount point that snapm may re-use: an empty directory, with nothing mounted on it, that is not a symbolic link.

Parameters:

path – The path to test.

Returns:

True if path may be re-used as a mount point, or False otherwise.

_snapset_mounts(snapset: SnapshotSet) → List[Mount]

Return the mounts for snapset, in discovery order.

Parameters:

snapset – The snapshot set to look up.

Returns:

A list of Mount objects, empty if the set is not mounted.

_update_mount_root(snapset: SnapshotSet) → None

Point the snapshot set’s reported mount root at its canonical mount, or clear it if the set is no longer mounted anywhere.

Parameters:

snapset – The snapshot set to update.

discover_mounts()

Discover and validate existing snapshot set mounts.

Mounts are found from the provenance records snapm writes as it mounts a snapshot set, so a set mounted at a custom mount root is found again without the path having to be derived from the mount table, and a snapshot that an administrator mounted by hand is not mistaken for a mount of snapm’s (#1025).

Nothing here writes. A record describing a mount that is no longer present is passed over rather than discarded: discovery runs for every command, including those that only report, and removing a record is _reap_stale_records()’s to do.

find_mounts(selection: Selection | None = None) → List[Mount]

Return a list of Mount objects describing the currently mounted snapshot sets managed by this Mounts instance.

Snapshot sets unmounted outside snapm are dropped before matching, and the records are read again and consulted, so that every mount returned is one that is still mounted and still the mount the entry says it is. Without that a report can name a path that another command has since unmounted and mounted a different snapshot set on.

None of this writes, so reporting on mounts neither takes the mount state lock nor waits on one. The answer is what was true when it was read: a report cannot be made to stay true, only to have been true, and holding the lock would not change that.

get_sys_mount() → SysMount

Return an object representing the system root file system mounts.

Returns:

A SysMount instance representing ‘/’.

Return type:

SysMount

mount(snapset: SnapshotSet, mount_root: str | None = None) → Mount

Mount the snapshot set snapset.

The mount state lock is held for the whole operation, so that the record written before the mount and the update written after it are one change as far as any other command is concerned.

Parameters:

• snapset – The snapshot set to operate on.

• mount_root – Optional path to mount the snapshot set at, in place of the default mount point beneath the mounts directory. The directory must already exist: snapm neither creates it nor removes it when the set is unmounted.

umount(snapset: SnapshotSet, mount_root: str | None = None)

Unmount the snapshot set snapset.

The mount state lock is held for the whole operation, so that the unmount and the discarding of the record that describes it are one change as far as any other command is concerned.

Parameters:

• snapset – The snapshot set to operate on.

• mount_root – Optional mount point selecting the mount to unmount. Only required when snapset is mounted at more than one path.

Raises:

• SnapmNotFoundError – If no matching mount is found.

• SnapmArgumentError – If snapset is mounted at more than one path and mount_root does not select exactly one of them.

class snapm.manager._mounts.SysMount

Bases: MountBase

A class representing the system root file system tree mounted at ‘/’.

TYPE = 'System root'

Description of the type of mount object.

__init__()

Initialize a new SysMount instance for the root file system.

Raises:

SnapmPathError – If root is not a directory, or not a mount point. This should never happen for the root file system.

_abc_impl = <_abc._abc_data object>

_chroot()

Return a non-chrooted command list for the root file system mount.

Returns:

A command list including an appropriate chroot invocation if necessary.

Return type:

List[str]

_do_mount()

Dummy template mount method hook for the system root file system.

_do_umount()

Template unmount method hook. Always rejects attempts to unmount the root file system.

snapset: SnapshotSet | None

snapm.manager._schedule module

Snapshot set scheduling abstractions for Snapshot Manager.

class snapm.manager._schedule.GcPolicy(policy_name: str, policy_type: GcPolicyType, params: dict)

Bases: object

An instance of a garbage collection policy.

__init__(policy_name: str, policy_type: GcPolicyType, params: dict)

Initialise a new GcPolicy.

Parameters:

• policy_type (GcPolicyType) – The policy type.

• params (dict) – A dictionary of parameters suitable for policy_type.

__repr__()

Return a machine readable string representation of this GcPolicy.

Returns:

A machine readable string.

Return type:

str

__str__()

Return a human readable string representation of this GcPolicy.

Returns:

A human readable string.

Return type:

str

disable()

Disable this GcPolicy and its corresponding timer.

enable()

Enable this GcPolicy and its corresponding timer.

property enabled

Return True if this GcPolicy and its corresponding timer are enabled, and False otherwise.

evaluate(sets: List[SnapshotSet])

Evaluate the list of SnapshotSet objects in sets against this GcPolicy and return a list of SnapshotSet objects that should be garbage collected.

Parameters:

sets (list[SnapshotSet].) – The list of SnapshotSet objects to evaluate, sorted in order of increasing creation date.

Returns:

A list of SnapshotSet objects to garbage collect.

Return type:

list[SnapshotSet]

classmethod from_cmd_args(cmd_args)

Initialise garbage collection policy from command line arguments.

Construct a new GcPolicy object from the command line arguments in cmd_args. The GcPolicyType is determined by the policy_type argument, and the type-specific policy parameters are initialised from the remaining arguments.

Parameters:

cmd_args – The command line selection arguments.

Returns:

A new GcPolicy instance

Return type:

GcPolicy

classmethod from_dict(data)

Instantiate a GcPolicy object from values in data.

Parameters:

data (dict) – A dictionary describing a GcPolicy instance.

Returns:

An instance of GcPolicy reflecting the values in data.

Return type:

GcPolicy

classmethod from_json(value)

Instantiate a GcPolicy object from a JSON string in value.

Parameters:

value – A JSON string describing a GcPolicyParams instance.

Returns:

An instance of a GcPolicyParams subclass reflecting the values in the JSON string value.

Return type:

GcPolicyParams

property has_params

Return True if this GcPolicy object has defined policy parameters or False otherwise.

Returns:

True if parameters are set else False.

Return type:

bool

json(pretty=False)

Return a JSON representation of this GcPolicy.

Returns:

This GcPolicy as a JSON string.

Return type:

str

property name

This GcPolicy instance’s name.

property params

This GcPolicy instance’s GcPolicyParams value.

property running

Return True if this GcPolicy and its corresponding timer are enabled, and False otherwise.

start()

Start this the timer for this GcPolicy.

stop()

Stop the timer for this GcPolicy.

to_dict()

Return a dictionary representation of this GcPolicy.

Returns:

This GcPolicy as a dictionary.

Return type:

dict

property type

This GcPolicy instance’s GcPolicyType value.

class snapm.manager._schedule.GcPolicyParams

Bases: object

Abstract base class for garbage collection policy parameters.

__str__()

Return a human readable string representation of this GcPolicyParams instance.

Returns:

A human readable string.

Return type:

str

evaluate(sets: List[SnapshotSet]) → List[SnapshotSet]

Evaluate the list of SnapshotSet objects in sets against this set of GcPolicyParams and return a list of SnapshotSet objects that should be garbage collected.

Parameters:

sets (list[SnapshotSet].) – The list of SnapshotSet objects to evaluate, sorted in order of increasing creation date.

Returns:

A list of SnapshotSet objects to garbage collect.

Return type:

list[SnapshotSet]

property has_params

Return True if this GcPolicyParams object has defined parameters or False otherwise.

Returns:

True if parameters are set else False.

Return type:

bool

json(pretty=False)

Return a JSON representation of this GcPolicyParams.

Returns:

This GcPolicyParams as a JSON string.

Return type:

str

to_dict()

Return a dictionary representation of this GcPolicyParams.

Returns:

This GcPolicyParams as a dictionary.

Return type:

dict

class snapm.manager._schedule.GcPolicyType(*values)

Bases: Enum

Garbage collection policy types enum.

AGE = 'Age'

ALL = 'All'

COUNT = 'Count'

TIMELINE = 'Timeline'

class snapm.manager._schedule.Schedule(name: str, sources: List[str], default_size_policy: str | None, autoindex: bool, calendarspec: str, gc_policy: GcPolicy, boot=False, revert=False)

Bases: object

An individual snapshot schedule instance with create and garbage collection timers. Tracks timer configuration, name, sources, size policies, enabled/disabled, nr snapshots, next elapse.

__init__(name: str, sources: List[str], default_size_policy: str | None, autoindex: bool, calendarspec: str, gc_policy: GcPolicy, boot=False, revert=False)

Initialise a new Schedule instance.

Parameters:

• name (str) – The name of the Schedule.

• sources (list[str]) – The source specs to include in this Schedule.

• default_size_policy (str) – The default size policy for this Schedule.

• autoindex (bool) – Enable autoindex names for this Schedule.

• calendarspec (str) – The OnCalendar expression for this Schedule.

• policy (GcPolicy) – The garbage collection policy for this Schedule.

Returns:

The new Schedule instance.

Return type:

Schedule

__repr__()

Return a machine-readable representation of this Schedule.

Returns:

A machine-readable string.

Return type:

str

__str__()

Return a human-readable representation of this Schedule.

Returns:

A human-readable string.

Return type:

str

property autoindex

The autoindex property for this Schedule.

property boot

Return True if this Schedule is configured to enable snapshot set boot entries when creating snapshot sets.

property calendarspec

The OnCalendar expression for the timer associated with this Schedule instance.

property default_size_policy

The default size policy for this Schedule.

delete_config()

Delete this Schedule’s on-disk configuration.

disable()

Disable this Schedule and its corresponding timer.

enable()

Enable this Schedule and its corresponding timers.

property enabled

Return True if this Schedule and its corresponding timer are enabled, and False otherwise.

classmethod from_dict(data)

Instantiate a Schedule object from values in data.

Parameters:

data (dict) – A dictionary describing a Schedule instance.

Returns:

An instance of Schedule reflecting the values in data.

Return type:

GcPolicyParams

classmethod from_file(sched_file)

Initialise a new Schedule instance from an on-disk JSON configuration file.

Parameters:

sched_file (str) – The path to the schedule configuration file.

Returns:

A new Schedule instance.

Return type:

Schedule

gc(sets: List[SnapshotSet]) → List[str]

Apply the configured garbage collection policy for this Schedule.

property gc_policy

The garbage collection policy for this Schedule.

json(pretty=False)

Return a JSON representation of this Schedule.

Returns:

This Schedule as a JSON string.

Return type:

str

property name

The name of this Schedule.

property next_elapse

Return the next_elapse time of this Schedule’s timer.

property revert

Return True if this Schedule is configured to enable snapshot set revert entries when creating snapshot sets.

property running

Return True if this Schedule and its corresponding timer are enabled, and False otherwise.

property sources

The source specs value for this Schedule.

start()

Start this the timer for this Schedule.

stop()

Stop the timer for this Schedule.

to_dict()

Return a dictionary representation of this Schedule.

Returns:

This Schedule as a dictionary.

Return type:

dict

write_config(sched_dir: str)

Write this Schedule’s configuration to disk.

Parameters:

sched_dir (str) – The path at which to write the configuration file.

snapm.manager._timers module

Systemd timer integration for Snapshot Manager.

class snapm.manager._timers.Timer(timer_type: TimerType, name: str, calendarspec: str | CalendarSpec)

Bases: object

High-level interface for managing schedling timers.

__init__(timer_type: TimerType, name: str, calendarspec: str | CalendarSpec)

Initialise a new Timer object from the provided arguments.

Parameters:

• type – A TimerType enum value representing the type of timer to create.

• name – The name of the timer. The value is used as the instance part of the timer unit name.

• calendarspec – A valid calendarspec string, or an instance of the CalendarSpec class representing the trigger condition for the timer.

disable()

Attempt to disable this Timer instance. Following a successful call to Timer.disable() the systemd unit is un-configured and no longer loaded. The Timer.status is TimerStatus.DISABLED.

Return type:

None

enable()

Attempt to enable this Timer instance. Following a successful call to Timer.enable() the systemd unit is configured and loaded, and the Timer.status is TimerStatus.ENABLED.

Return type:

None

property enabled

True if this Timer is enabled, and False otherwise.

Return type:

bool

property from_now

Return a string representation of the time remaining until this Timer next elapses.

Returns:

The time remaining until the next elapse as a string.

Return type:

str

property next_elapse

Return the next elapse time for this Timer object as an instance of datetime.datetime.

Returns:

The next elapse time as a datetime object.

Return type:

datetime.datetime

property occurs

True if this Timer will elapse in the future, and False otherwise.

Return type:

bool

property running

True if this Timer is running, and False otherwise.

Return type:

bool

start()

Attempt to start this Timer instance. Following a successful call to Timer.start() the systemd unit is ‘active’ / ‘waiting’ and will be invoked when the next elapse time is reached. The Timer.status is TimerStatus.RUNNING.

Return type:

None

property status

Return a TimerStatus instance reflecting the status of this timer.

Return type:

TimerStatus

stop()

Attempt to stop this Timer instance. Following a successful call to Timer.stop() the systemd unit is ‘loaded’ / ‘dead’ and the Timer.status is TimerStatus.ENABLED.

Return type:

None

class snapm.manager._timers.TimerStatus(*values)

Bases: Enum

Enum class representing the possible timer status values.

DISABLED = 'disable'

ENABLED = 'enabled'

INVALID = 'invalid'

RUNNING = 'running'

STOPPED = 'stopped'

class snapm.manager._timers.TimerType(*values)

Bases: Enum

Enum class representing the available timer types.

CREATE = 'create'

GC = 'gc'

snapm.manager.plugins module

Snapshot manager plugin interface.

class snapm.manager.plugins.Plugin(logger, plugin_cfg)

Bases: ABC

Abstract base class for snapshot manager plugins.

__init__(logger, plugin_cfg)

_abc_impl = <_abc._abc_data object>

static _check_limit(limit, tracking, key)

Check whether adding a new snapshot would exceed a configured limit.

Parameters:

• limit – The configured maximum (0 means unlimited).

• tracking – A dict mapping keys to current snapshot counts.

• key – The key to look up in tracking.

Returns:

True if adding a new snapshot would exceed the limit, or False otherwise.

Return type:

bool

_log_debug(*args)

Log at debug level.

_log_error(*args)

Log at error level.

_log_info(*args)

Log at info level.

_log_warn(*args)

Log at warning level.

abstractmethod activate_snapshot(name)

Activate the snapshot named name

Parameters:

name – The name of the snapshot to be activated.

abstractmethod can_snapshot(source)

Test whether this plugin can snapshot the specified mount point or block device.

Parameters:

source – The block device or mount point path to test.

Returns:

True if this plugin can snapshot the file system mounted at mount_point, or False otherwise.

abstractmethod check_create_snapshot(origin, snapset_name, timestamp, mount_point, size_policy)

Perform pre-creation checks before creating a snapshot.

Parameters:

• origin – The origin volume for the snapshot.

• snapset_name – The name of the snapshot set to be created.

• timestamp – The snapshot set timestamp.

• mount_point – The mount point path for this snapshot.

Raises:

SnapmNoSpaceError if there is insufficient free space to create the snapshot.

abstractmethod check_resize_snapshot(name, origin, mount_point, size_policy)

Check whether this snapshot can be resized to the requested size_policy. This method returns if the resize can be satisfied and raises an exception if not.

Returns:

None

Raises:

SnapmNoSpaceError if insufficient space is available to satisfy the requested size policy or SnapmPluginError if another reason prevents the snapshot from being resized.

abstractmethod check_revert_snapshot(name, origin)

Check whether this snapshot can be reverted or not. This method returns if the current snapshot can be reverted and raises an exception if not.

Returns:

None

Raises:

NotImplementedError if this plugin does not support the revert operation, SnapmBusyError if the snapshot is already in the process of being reverted to another snapshot state or SnapmPluginError if another reason prevents the snapshot from being merged.

abstractmethod create_snapshot(origin, snapset_name, timestamp, mount_point, size_policy)

Create a snapshot of origin in the snapset named snapset_name.

Parameters:

• origin – The origin volume for the snapshot.

• snapset_name – The name of the snapshot set to be created.

• timestamp – The snapshot set timestamp.

• mount_point – The mount point path for this snapshot.

Raises:

SnapmNoSpaceError if there is insufficient free space to create the snapshot.

abstractmethod deactivate_snapshot(name)

Deactivate the snapshot named name

Parameters:

name – The name of the snapshot to be deactivated.

abstractmethod delete_snapshot(name)

Delete the snapshot named name

Parameters:

name – The name of the snapshot to be removed.

abstractmethod discover_snapshots()

Discover snapshots managed by this plugin class.

Returns a list of objects that are a subclass of Snapshot. :returns: A list of snapshots discovered by this plugin class.

end_transaction()

End a snapshot set creation transaction in this plugin.

info()

Return plugin name and version.

name = 'plugin'

abstractmethod origin_from_mount_point(mount_point)

Return a string representing the origin from a given mount point path.

Parameters:

mount_point – The mount point path.

abstractmethod rename_snapshot(old_name, origin, snapset_name, timestamp, mount_point)

Rename the snapshot named old_name according to the provided snapshot field values.

Parameters:

• old_name – The original name of the snapshot to be renamed.

• origin – The origin volume for the snapshot.

• snapset_name – The new name of the snapshot set.

• timestamp – The snapshot set timestamp.

• mount_point – The mount point of the snapshot.

abstractmethod resize_snapshot(name, origin, mount_point, size_policy)

Attempt to resize the snapshot name to the requested size_policy. This method returns if the resize can be satisfied and raises an exception if not.

Returns:

None

Raises:

SnapmNoSpaceError if insufficient space is available to satisfy the requested size policy or SnapmPluginError if another reason prevents the snapshot from being resized.

abstractmethod revert_snapshot(name)

Request to revert a snapshot and revert the content of the origin volume to its state at the time of the snapshot.

This may be deferred until the next device activation or mount operation for the respective volume.

Parameters:

name – The name of the snapshot to revert.

abstractmethod set_autoactivate(name, auto=False)

Set the autoactivation state of the snapshot named name.

Parameters:

• name – The name of the snapshot to be modified.

• auto – True to enable autoactivation or False otherwise.

snapshot_class

alias of Snapshot

start_transaction()

Begin a snapshot set creation transaction in this plugin.

version = '0.2.0'

class snapm.manager.plugins.PluginLimits(cfg)

Bases: object

Per-plugin resource limits.

__init__(cfg)

Initialise a new PluginLimits instance.

snapm.manager.plugins.decode_mount_point(mount_str)

Parse the mount point component of a snapshot name.

snapm.manager.plugins.device_from_mount_point(mount_point)

Convert a mount point path to a corresponding device path.

Return the device corresponding to the file system mount point mount_point according to /proc/self/mounts.

snapm.manager.plugins.encode_mount_point(mount_point)

Encode mount point paths.

Encode a mount point for use in a snapshot name by replacing the path separator with ‘-‘.

snapm.manager.plugins.format_snapshot_name(origin, snapset_name, timestamp, mount_point)

Format structured snapshot name.

Format a snapshot name according to the snapshot manager naming format.

snapm.manager.plugins.mount_point_space_used(mount_point)

Retrieve space usage for a mount point path.

Return the space used on the file system mounted at mount_point as a value in bytes.

snapm.manager.plugins.parse_snapshot_name(full_name, origin)

Attempt to parse a snapshot set snapshot name.

Returns a tuple of (snapset_name, timestamp, mount_point) if full_name is a valid snapset snapshot name, or None otherwise.

snapm.fsdiff.changes module

File system entry change detection and classification.

class snapm.fsdiff.changes.ChangeDetector

Bases: object

Class to detect and classify changes in FsEntry objects.

_effective_changes(changes: List[FileChange], options: DiffOptions) → List[FileChange]

Elide ignored change types in changes.

Parameters:

• changes (List[FileChange]) – The list of changes to examine.

• options (DiffOptions) – Effective diff options.

Returns:

Changes pruned according to options.

Return type:

List[FileChange]

detect_added(new_entry: FsEntry, options: DiffOptions) → List[FileChange]

Detect all changes for a newly added file.

Parameters:

• new_entry (FsEntry) – The updated file system entry.

• options (DiffOptions) – Change detection options.

Returns:

A list of changes to the file system entry.

Return type:

List[FileChange]

detect_changes(old_entry: FsEntry, new_entry: FsEntry, options: DiffOptions) → List[FileChange]

Detect all changes between two file system entries.

Parameters:

• old_entry (FsEntry) – The original file system entry.

• new_entry (FsEntry) – The updated file system entry.

• options (DiffOptions) – Change detection options.

Returns:

A list of changes to the file system entry.

Return type:

List[FileChange]

detect_removed(old_entry: FsEntry, options: DiffOptions) → List[FileChange]

Detect all changes for a removed file.

Parameters:

• old_entry (FsEntry) – The updated file system entry.

• options (DiffOptions) – Change detection options.

Returns:

A list of changes to the file system entry.

Return type:

List[FileChange]

class snapm.fsdiff.changes.ChangeType(*values)

Bases: Enum

Enum representing different types of file system entry change.

CONTENT = 'content'

OWNERSHIP = 'ownership'

PERMISSIONS = 'permissions'

SYMLINK_TARGET = 'symlink_target'

TIMESTAMPS = 'timestamps'

XATTRS = 'extended_attributes'

class snapm.fsdiff.changes.FileChange(change_type: ChangeType, old_value: str | None = None, new_value: str | None = None, description: str = '')

Bases: object

Representation of a file system entry change.

__init__(change_type: ChangeType, old_value: str | None = None, new_value: str | None = None, description: str = '')

Initialise a new FileChange object.

Parameters:

• change_type (ChangeType) – The type of change.

• old_value (Optional[str]) – The original value.

• new_value (Optional[str]) – The changed value.

• description (str) – A string description of the change.

__str__() → str

Return a string representation of this FileChange object.

Returns:

A human readable string representation of this instance.

Return type:

str

to_dict() → Dict[str, str | None]

Convert this FileChange object into a dictionary representation suitable for encoding as JSON.

Returns:

A dictionary mapping this instance’s keys to values.

Return type:

Dict[str, Optional[str]]

snapm.fsdiff.changes._log_debug_fsdiff(msg, *args, **kwargs)

A wrapper for fsdiff subsystem debug logs.

snapm.fsdiff.contentdiff module

Content-aware diff support.

class snapm.fsdiff.contentdiff.BinaryContentDiffer

Bases: ContentDifferBase

Binary file content differ.

_abc_impl = <_abc._abc_data object>

binary_types = (FileTypeCategory.BINARY, FileTypeCategory.IMAGE, FileTypeCategory.AUDIO, FileTypeCategory.VIDEO, FileTypeCategory.ARCHIVE, FileTypeCategory.EXECUTABLE, FileTypeCategory.DATABASE, FileTypeCategory.DOCUMENT)

can_handle(file_type_info: FileTypeInfo) → bool

Return True if this differ can handle the given file type.

Parameters:

file_type_info (FileTypeInfo) – File type information for the file to compare.

Returns:

True if this content differ can handle this file.

Return type:

bool

generate_diff(_old_path: Path | None, _new_path: Path | None, old_entry: FsEntry | None, new_entry: FsEntry | None) → ContentDiff

Generate binary diff summary.

Parameters:

• _old_path (Optional[Path]) – Path describing the original path to diff (unused).

• _new_path (Optional[Path]) – Path describing the updated path to diff (unused).

• old_entry (Optional[FsEntry]) – The original FsEntry to diff.

• new_entry (Optional[FsEntry]) – The updated FsEntry to diff.

Returns:

A diff of the two entries.

Return type:

ContentDiff

property priority: int

Priority for selection when multiple differs match (higher = preferred)

Returns:

Integer priority level.

Return type:

int

class snapm.fsdiff.contentdiff.ContentDiff(diff_type: str, old_content: str | None = None, new_content: str | None = None, summary: str = '')

Bases: object

Represents a content-aware diff between two files.

__init__(diff_type: str, old_content: str | None = None, new_content: str | None = None, summary: str = '')

Initialise a new ContentDiff object.

Parameters:

• diff_type (str) – The type of diff to perform.

• old_content (str) – The original content.

• new_content (str) – The updated content.

• summary (str) – A summary of the difference.

__str__()

Return a string representation of this ContentDiff object.

Returns:

A human readable string representing this instance.

Return type:

str

to_dict() → Dict[str, Any]

Convert this ContentDiff object into a dictionary representation suitable for encoding as JSON.

Returns:

A dictionary mapping this instance’s keys to values.

Return type:

Dict[str, Any]

class snapm.fsdiff.contentdiff.ContentDifferBase

Bases: ABC

Base class for content-aware diff implementations.

_abc_impl = <_abc._abc_data object>

abstractmethod can_handle(file_type_info: FileTypeInfo) → bool

Return True if this differ can handle the given file type.

Parameters:

file_type_info (FileTypeInfo) – File type information for the file to compare.

Returns:

True if this content differ can handle this file.

Return type:

bool

abstractmethod generate_diff(old_path: Path | None, new_path: Path | None, old_entry: FsEntry | None, new_entry: FsEntry | None) → ContentDiff

Generate content diff between two files.

Parameters:

• old_path (Optional[Path]) – Path describing the original path to diff.

• new_path (Optional[Path]) – Path describing the updated path to diff.

• old_entry (Optional[FsEntry]) – The original FsEntry to diff.

• new_entry (Optional[FsEntry]) – The updated FsEntry to diff.

Returns:

A diff of the two entries.

Return type:

ContentDiff

abstract property priority: int

Priority for selection when multiple differs match (higher = preferred)

Returns:

Integer priority level.

Return type:

int

class snapm.fsdiff.contentdiff.ContentDifferManager

Bases: object

Manager for content-aware diff implementations.

__init__()

Initialise a new ContentDifferManager instance.

_register_default_differs()

Register built-in content differs.

generate_content_diff(old_path: Path | None, new_path: Path | None, old_entry: FsEntry | None, new_entry: FsEntry | None) → ContentDiff | None

Generate content diff using appropriate differ.

Parameters:

• old_path (Optional[Path]) – Path describing the original path to diff.

• new_path (Optional[Path]) – Path describing the updated path to diff.

• old_entry (Optional[FsEntry]) – The original FsEntry to diff.

• new_entry (Optional[FsEntry]) – The updated FsEntry to diff.

Returns:

A diff of the two entries.

Return type:

Optional[ContentDiff]

get_differ_for_file(file_type_info: FileTypeInfo) → ContentDifferBase

Get the best content differ for a file type.

Parameters:

file_type_info (FileTypeInfo) – The file type to find a differ for.

Returns:

An appropriate differ for file_type_info.

Return type:

A ContentDifferBase subclass.

register_differ(differ: ContentDifferBase)

Register a new content differ.

class snapm.fsdiff.contentdiff.JsonContentDiffer

Bases: ContentDifferBase

JSON-aware content differ.

_abc_impl = <_abc._abc_data object>

can_handle(file_type_info: FileTypeInfo) → bool

Return True if this differ can handle the given file type.

Parameters:

file_type_info (FileTypeInfo) – File type information for the file to compare.

Returns:

True if this content differ can handle this file.

Return type:

bool

generate_diff(old_path: Path | None, new_path: Path | None, old_entry: FsEntry | None, new_entry: FsEntry | None) → ContentDiff

Generate structured JSON diff.

Parameters:

• old_path (Optional[Path]) – Path describing the original path to diff.

• new_path (Optional[Path]) – Path describing the updated path to diff.

• old_entry (Optional[FsEntry]) – The original FsEntry to diff.

• new_entry (Optional[FsEntry]) – The updated FsEntry to diff.

Returns:

A diff of the two entries.

Return type:

ContentDiff

property priority: int

Priority for selection when multiple differs match (higher = preferred)

Returns:

Integer priority level.

Return type:

int

class snapm.fsdiff.contentdiff.TextContentDiffer

Bases: ContentDifferBase

Default text-based content differ.

_abc_impl = <_abc._abc_data object>

can_handle(file_type_info: FileTypeInfo) → bool

Return True if this differ can handle the given file type.

Parameters:

file_type_info (FileTypeInfo) – File type information for the file to compare.

Returns:

True if this content differ can handle this file.

Return type:

bool

generate_diff(old_path: Path | None, new_path: Path | None, old_entry: FsEntry | None, new_entry: FsEntry | None) → ContentDiff

Generate unified diff for text files.

Parameters:

• old_path (Optional[Path]) – Path describing the original path to diff.

• new_path (Optional[Path]) – Path describing the updated path to diff.

• old_entry (Optional[FsEntry]) – The original FsEntry to diff.

• new_entry (Optional[FsEntry]) – The updated FsEntry to diff.

Returns:

A diff of the two entries.

Return type:

ContentDiff

property priority: int

Priority for selection when multiple differs match (higher = preferred)

Returns:

Integer priority level.

Return type:

int

snapm.fsdiff.contentdiff._log_debug_fsdiff(msg, *args, **kwargs)

A wrapper for fsdiff subsystem debug logs.

snapm.fsdiff.engine module

File system diff engine

class snapm.fsdiff.engine.DiffEngine

Bases: object

Core class for generating fsdiff comparisons.

__init__()

Initialise a new DiffEngine instance.

static _best_sibling_proximity(orig_path: str, candidates: List[Tuple[str, FsEntry]]) → Tuple[str, FsEntry]

Sibling proximity heuristic: choose the “best” move destination when multiple candidates exist.

Parameters:

• orig_path (str) – The original path before the move.

• candidates (List[Tuple[str, FsEntry]]) – Possible candidates for the destination, as a list of (path, entry) pairs.

Returns:

The winning (path, entry) pair.

Return type:

Tuple[str, FsEntry]

_detect_moves(diffs: List[FsDiffRecord], tree_a: Dict[str, FsEntry], tree_b: Dict[str, FsEntry], options: DiffOptions, term_control: TermControl | None = None) → List[FsDiffRecord]

Detect file moves/renames by matching content hashes

Parameters:

• diffs (List[FsDiffRecord]) – A list of file system diff records to inspect.

• tree_a (Dict[str, FsEntry]) – The first file system tree to compare: a dictionary of path name -> file system entry mappings.

• tree_b (Dict[str, FsEntry]) – The second file system tree to compare: a dictionary of path name -> file system entry mappings.

• options (DiffOptions) – Options to apply to the diff generation.

• term_control (TermControl) – A TermControl instance to use for rendering color output.

Returns:

Updated list of diff records with moves detected.

Return type:

List[FsDiffRecord]

static _is_move_diff(diff, src_path, dest_path)

Return True if diff reflects a move diff type for the current path/dest_path or False otherwise.

Parameters:

• diff – The FsDiffRecord to inspect.

• src_path – The original path to check.

• dest_path – The destination path to check.

compute_diff(tree_a: Dict[str, FsEntry], tree_b: Dict[str, FsEntry], options: DiffOptions = None, term_control: TermControl | None = None) → FsDiffResults

Main diff computation logic.

Parameters:

• tree_a (Dict[str, FsEntry]) – The first file system tree to compare: a dictionary of path name -> file system entry mappings.

• tree_b (Dict[str, FsEntry]) – The second file system tree to compare: a dictionary of path name -> file system entry mappings.

• options (DiffOptions) – Options to apply to the diff generation.

• term_control (TermControl) – A TermControl instance to use for rendering color output.

Returns:

An FsDiffResults instance containing FsDiffRecord objects.

Return type:

FsDiffResults

class snapm.fsdiff.engine.FsDiffRecord(path: str, diff_type: DiffType, old_entry: FsEntry | None = None, new_entry: FsEntry | None = None)

Bases: object

Diff record compatible with snapm.report system Represents a single file/directory change

__init__(path: str, diff_type: DiffType, old_entry: FsEntry | None = None, new_entry: FsEntry | None = None)

Initialise a new FsDiffRecord object.

Parameters:

• path (str) – A string describing the path for this diff record.

• diff_type (DiffType) – The diff type for this diff record.

• old_entry – The original entry for the comparison.

• new_entry – The updated entry for the comparison.

__str__() → str

Return a string representation of this FsDiffRecord object.

Returns:

A human readable representation of this FsDiffRecord.

Return type:

str

_get_file_category() → str

Get file category for reporting

Returns:

The file category for this diff record.

Return type:

str

_get_file_type() → str

Get file type for reporting.

Returns:

A description of the file type for this diff record.

Return type:

str

_get_file_type_desc() → str

Get file type description for reporting.

Returns:

A detailed description of the file type for this diff record.

Return type:

str

add_change(change: FileChange)

Add a detected change.

Parameters:

change (FileChange) – The file change to record.

get_change_summary() → str

Get human-readable change summary

json(pretty=False) → str

Return a string representation of this FsDiffRecord in JSON notation.

Parameters:

pretty (bool) – Indent JSON to be human readable.

Returns:

A JSON representation of this instance.

Return type:

str

set_content_diff(content_diff: ContentDiff)

Set content-level diff.

Parameters:

content_diff (ContentDiff) – The content diff for this diff record.

to_dict() → Dict[str, Any]

Convert this FsDiffRecord object into a dictionary representation suitable for encoding as JSON.

Returns:

A dictionary mapping this instance’s keys to values.

Return type:

Dict[str, Any]

class snapm.fsdiff.engine.FsDiffResults(records: List[FsDiffRecord], options: DiffOptions, timestamp: int, count: int = 0)

Bases: object

Container for filesystem diff results with formatting methods.

DIFF_FORMATS: ClassVar[List[str]] = ['paths', 'full', 'short', 'json', 'diff', 'summary', 'tree']

Constant for the names of the string diff formats

__init__(records: List[FsDiffRecord], options: DiffOptions, timestamp: int, count: int = 0)

__repr__() → str

Return a machine-readable representation of this instance.

Returns:

FsDiffResults constructor style string.

Return type:

str

property added: List[FsDiffRecord]

Return added changes in this FsDiffResults instance.

Returns:

Changes with DiffType.ADDED type.

Return type:

List[FsDiffRecord]

property content_changes: int

Return the number of content changes in this FsDiffResults instance.

Returns:

Count of changes with content diff.

Return type:

int

diff(diffstat: bool = False, color: str = 'auto', term_control: TermControl | None = None) → str

Return unified diff representation of content changes for this instance.

Parameters:

• diffstat (bool) – Include “diffstat”-like change summary.

• color (str) – A string to control color diff rendering: “auto”, “always”, or “never”.

• term_control (Optional[TermControl]) – An optional TermControl instance to use for formatting. The supplied instance overrides any color argument if set.

Returns:

unified diff string description of file system changes.

Return type:

str

full() → str

Return a string with full FsDiffRecord content for this instance.

Returns:

String description of file system changes.

Return type:

str

json(pretty: bool = False) → str

Return JSON representation of FsDiffRecord content for this instance.

Returns:

JSON string description of file system changes.

Return type:

str

property modified: List[FsDiffRecord]

Return modified changes in this FsDiffResults instance.

Returns:

Changes with DiffType.MODIFIED type.

Return type:

List[FsDiffRecord]

property moved: List[FsDiffRecord]

Return moved changes in this FsDiffResults instance.

Returns:

Changes with DiffType.MOVED type.

Return type:

List[FsDiffRecord]

paths() → List[str]

Return a list of paths that changed in this FsDiffResults.

Returns:

Path list.

Return type:

List[str]

property removed: List[FsDiffRecord]

Return removed changes in this FsDiffResults instance.

Returns:

Changes with DiffType.REMOVED type.

Return type:

List[FsDiffRecord]

short() → str

Return brief summary of FsDiffRecord content for this instance.

Returns:

Brief string description of file system changes.

Return type:

str

summary(diffstat: bool = False, color: str = 'auto', term_control: TermControl | None = None) → str

Return a summary of this FsDiffResults instance.

Parameters:

• diffstat (bool) – Include “diffstat”-like change summary.

• color (str) – A string to control color rendering: “auto”, “always”, or “never”.

• term_control (Optional[TermControl]) – An optional TermControl instance to use for formatting. The supplied instance overrides any color argument if set.

Returns:

A string summarizing this instance.

Return type:

str

property total_changes: int

Return the total number of changes in this FsDiffResults instance: equivalent to len(self).

Returns:

Count of changes.

Return type:

int

tree(color: str = 'auto', desc: str = 'none', term_control: TermControl | None = None) → str

Render a DiffTree of this FsDiffResults instance.

Parameters:

• color (str) – A string to control color tree rendering: “auto”, “always”, or “never”.

• desc (Optional[str]) – Include descriptions: “short” for brief description, “full” for complete change descriptions or “none” to omit.

• term_control (Optional[TermControl]) – An optional TermControl instance to use for formatting. The supplied instance overrides any color argument if set.

Returns:

A string representation of a difference tree

Return type:

str

property type_changed: List[FsDiffRecord]

Return type_changed changes in this FsDiffResults instance.

Returns:

Changes with DiffType.TYPE_CHANGED type.

Return type:

List[FsDiffRecord]

snapm.fsdiff.engine._log_debug_fsdiff(msg, *args, **kwargs)

A wrapper for fsdiff subsystem debug logs.

snapm.fsdiff.engine._log_debug_fsdiff_extra(msg, *args, **kwargs)

A wrapper for fsdiff subsystem debug logs.

snapm.fsdiff.engine.render_diff_stat(records: List[FsDiffRecord], term_control: TermControl) → str

Render a diffstat-style summary for the given records.

Parameters:

• records (FsDiffRecord) – Diff records with content diffs.

• term_control (TermControl) – A TermControl instance to use for rendering color output.

Returns:

Diffstat-style summary string.

Return type:

str

snapm.fsdiff.engine.render_unified_diff(record: FsDiffRecord, tc: TermControl | None) → str

Render a unified diff for a modified file.

Parameters:

• record (FsDiffRecord) – The diff record to render.

• tc (Optional[TermControl]) – An optional TermControl instance to use for rendering color output.

Returns:

Rendered unified diff string.

Return type:

str

snapm.fsdiff.filetypes module

File type information support.

class snapm.fsdiff.filetypes.FileTypeCategory(*values)

Bases: Enum

Enum for file type categories.

ARCHIVE = 'archive'

AUDIO = 'audio'

BINARY = 'binary'

BINARY_LOG = 'binary_log'

BLOCK = 'block'

CERTIFICATE = 'certificate'

CHAR = 'char'

CONFIG = 'config'

DATABASE = 'database'

DIRECTORY = 'directory'

DOCUMENT = 'document'

EXECUTABLE = 'executable'

FIFO = 'FIFO'

IMAGE = 'image'

LOG = 'log'

SOCK = 'socket'

SOURCE_CODE = 'source_code'

SYMLINK = 'symlink'

TEXT = 'text'

UNKNOWN = 'unknown'

VIDEO = 'video'

class snapm.fsdiff.filetypes.FileTypeDetector

Bases: object

Detect file types using magic from python3-file-magic.

_categorize_file(mime_type: str, file_path: Path) → FileTypeCategory

Categorize file based on MIME type and path patterns.

Parameters:

• mime_type (str) – Detected file MIME type.

• file_path (Path) – Path to the file to categorize.

Returns:

File type categorization.

Return type:

FileTypeCategory

_guess_file_type(file_path: Path) → FileTypeInfo

Attempt to guess file type based on extension and file path without using python-magic.

Parameters:

file_path (Path) – The path to guess file type for.

Returns:

A FileTypeInfo object with a best-effort guess of the file type.

Return type:

FileTypeInfo

category_rules: ClassVar[Dict[str, FileTypeCategory]] = {'application/gzip': FileTypeCategory.ARCHIVE, 'application/java-archive': FileTypeCategory.ARCHIVE, 'application/javascript': FileTypeCategory.SOURCE_CODE, 'application/json': FileTypeCategory.CONFIG, 'application/ld+json': FileTypeCategory.CONFIG, 'application/mbox': FileTypeCategory.DATABASE, 'application/msword': FileTypeCategory.DOCUMENT, 'application/pdf': FileTypeCategory.DOCUMENT, 'application/pkix-cert': FileTypeCategory.CERTIFICATE, 'application/rtf': FileTypeCategory.DOCUMENT, 'application/toml': FileTypeCategory.CONFIG, 'application/vnd.android.package-archive': FileTypeCategory.ARCHIVE, 'application/vnd.microsoft.portable-executable': FileTypeCategory.EXECUTABLE, 'application/vnd.ms-excel': FileTypeCategory.DOCUMENT, 'application/vnd.ms-powerpoint': FileTypeCategory.DOCUMENT, 'application/vnd.oasis.opendocument.spreadsheet': FileTypeCategory.DOCUMENT, 'application/vnd.oasis.opendocument.text': FileTypeCategory.DOCUMENT, 'application/vnd.openxmlformats-officedocument.presentationml.presentation': FileTypeCategory.DOCUMENT, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet': FileTypeCategory.DOCUMENT, 'application/vnd.openxmlformats-officedocument.wordprocessingml.document': FileTypeCategory.DOCUMENT, 'application/vnd.sqlite3': FileTypeCategory.DATABASE, 'application/x-7z-compressed': FileTypeCategory.ARCHIVE, 'application/x-bat': FileTypeCategory.SOURCE_CODE, 'application/x-bzip2': FileTypeCategory.ARCHIVE, 'application/x-csh': FileTypeCategory.SOURCE_CODE, 'application/x-dbf': FileTypeCategory.DATABASE, 'application/x-dosexec': FileTypeCategory.EXECUTABLE, 'application/x-elf': FileTypeCategory.EXECUTABLE, 'application/x-executable': FileTypeCategory.EXECUTABLE, 'application/x-fish': FileTypeCategory.SOURCE_CODE, 'application/x-gzip': FileTypeCategory.ARCHIVE, 'application/x-iso9660-image': FileTypeCategory.ARCHIVE, 'application/x-javascript': FileTypeCategory.SOURCE_CODE, 'application/x-ksh': FileTypeCategory.SOURCE_CODE, 'application/x-lzip': FileTypeCategory.ARCHIVE, 'application/x-lzma': FileTypeCategory.ARCHIVE, 'application/x-mach-binary': FileTypeCategory.EXECUTABLE, 'application/x-msdownload': FileTypeCategory.EXECUTABLE, 'application/x-object': FileTypeCategory.EXECUTABLE, 'application/x-pem-file': FileTypeCategory.CERTIFICATE, 'application/x-pie-executable': FileTypeCategory.EXECUTABLE, 'application/x-rar': FileTypeCategory.ARCHIVE, 'application/x-rar-compressed': FileTypeCategory.ARCHIVE, 'application/x-sh': FileTypeCategory.SOURCE_CODE, 'application/x-sharedlib': FileTypeCategory.EXECUTABLE, 'application/x-sqlite3': FileTypeCategory.DATABASE, 'application/x-tar': FileTypeCategory.ARCHIVE, 'application/x-x509-ca-cert': FileTypeCategory.CERTIFICATE, 'application/x-xz': FileTypeCategory.ARCHIVE, 'application/x-yaml': FileTypeCategory.CONFIG, 'application/x-zsh': FileTypeCategory.SOURCE_CODE, 'application/xml': FileTypeCategory.CONFIG, 'application/yaml': FileTypeCategory.CONFIG, 'application/zip': FileTypeCategory.ARCHIVE, 'application/zstd': FileTypeCategory.ARCHIVE, 'audio/': FileTypeCategory.AUDIO, 'font/': FileTypeCategory.BINARY, 'image/': FileTypeCategory.IMAGE, 'inode/blockdevice': FileTypeCategory.BLOCK, 'inode/chardevice': FileTypeCategory.CHAR, 'inode/directory': FileTypeCategory.DIRECTORY, 'inode/fifo': FileTypeCategory.FIFO, 'inode/socket': FileTypeCategory.SOCK, 'inode/symlink': FileTypeCategory.SYMLINK, 'model/': FileTypeCategory.BINARY, 'text/': FileTypeCategory.TEXT, 'text/asciidoc': FileTypeCategory.DOCUMENT, 'text/css': FileTypeCategory.SOURCE_CODE, 'text/html': FileTypeCategory.SOURCE_CODE, 'text/javascript': FileTypeCategory.SOURCE_CODE, 'text/markdown': FileTypeCategory.DOCUMENT, 'text/rtf': FileTypeCategory.DOCUMENT, 'text/troff': FileTypeCategory.DOCUMENT, 'text/vbs': FileTypeCategory.SOURCE_CODE, 'text/x-awk': FileTypeCategory.SOURCE_CODE, 'text/x-bibtex': FileTypeCategory.DOCUMENT, 'text/x-c': FileTypeCategory.SOURCE_CODE, 'text/x-c++': FileTypeCategory.SOURCE_CODE, 'text/x-diff': FileTypeCategory.SOURCE_CODE, 'text/x-ini': FileTypeCategory.CONFIG, 'text/x-java-source': FileTypeCategory.SOURCE_CODE, 'text/x-lua': FileTypeCategory.SOURCE_CODE, 'text/x-makefile': FileTypeCategory.SOURCE_CODE, 'text/x-nfo': FileTypeCategory.DOCUMENT, 'text/x-perl': FileTypeCategory.SOURCE_CODE, 'text/x-powershell': FileTypeCategory.SOURCE_CODE, 'text/x-python': FileTypeCategory.SOURCE_CODE, 'text/x-script.python': FileTypeCategory.SOURCE_CODE, 'text/x-sed': FileTypeCategory.SOURCE_CODE, 'text/x-shellscript': FileTypeCategory.SOURCE_CODE, 'text/x-tcl': FileTypeCategory.SOURCE_CODE, 'text/x-tex': FileTypeCategory.DOCUMENT, 'text/x-toml': FileTypeCategory.CONFIG, 'text/xml': FileTypeCategory.CONFIG, 'text/yaml': FileTypeCategory.CONFIG, 'video/': FileTypeCategory.VIDEO}

detect_file_type(file_path: Path, strip_prefix: str = '', use_magic=False) → FileTypeInfo

Detect comprehensive file type information, optionally using python-magic for MIME type detection.

Parameters:

• file_path (Path.) – The path to the file to inspect.

• strip_prefix (str) – An optional prefix (mount root) to strip from the file_path when guessing file types.

Returns:

File type information for file_path.

Return type:

FileTypeInfo

class snapm.fsdiff.filetypes.FileTypeInfo(mime_type: str, description: str, category: FileTypeCategory, encoding: str | None = None)

Bases: object

Class representing file type information and encoding.

TEXT_DOCUMENTS = ('application/rtf', 'application/x-lyx')

__init__(mime_type: str, description: str, category: FileTypeCategory, encoding: str | None = None)

Initialise a new FileTypeInfo object.

Parameters:

• mime_type (str) – The detected MIME type.

• description (str) – Type description returned by magic.

• category (FileTypeCategory) – File type category.

• encoding (Optional[str]) – Optional file encoding.

__str__()

Return a string representation of this FileTypeInfo object.

Returns:

A human readable string describing this instance.

Return type:

str

snapm.fsdiff.filetypes._generic_guess_file(file_path: Path, extension_map: Dict[str, Tuple[str, str]], filename_map: Dict[str, Tuple[str, str]], encoding: str) → Tuple[str, str, str] | None

Attempt to guess a file’s MIME type and description based on the file name and extension.

Parameters:

• file_path (Path) – A Path instance containing the file path to check.

• extension_map (Dict[str, Tuple[str, str]]) – A map of “.extension”: (mime_type, description) tuples to use.

• filename_map – A map of “filename”: (mime_type, description) tuples to use.

Returns:

A 3-tuple containing (mime_type, description, encoding) if the type could be guessed or None otherwise.

Return type:

Optional[Tuple[str, str, str]]

snapm.fsdiff.filetypes._guess_binary_file(file_path: Path) → Tuple[str, str, str] | None

Attempt to guess a binary file’s MIME type and description based on the file name and extension.

Parameters:

file_path (Path) – A Path instance containing the file path to check.

Returns:

A 3-tuple containing (mime_type, description, encoding) if the type could be guessed or None otherwise.

Return type:

Optional[Tuple[str, str, str]]

snapm.fsdiff.filetypes._guess_file(file_path: Path) → Tuple[str, str, str]

Attempt to guess a file’s MIME type and description based on the file name and extension.

Parameters:

file_path (Path) – A Path instance containing the file path to check.

Returns:

A 3-tuple containing (mime_type, description, encoding).

Return type:

Tuple[str, str, str]

snapm.fsdiff.filetypes._guess_text_file(file_path: Path) → Tuple[str, str, str] | None

Attempt to guess a text file’s MIME type and description based on the file name and extension.

Parameters:

file_path (Path) – A Path instance containing the file path to check.

Returns:

A 3-tuple containing (mime_type, description, encoding) if the type could be guessed or None otherwise.

Return type:

Optional[Tuple[str, str, str]]

snapm.fsdiff.filetypes._is_binary_log(file_path: Path) → bool

Determine based on file name pattern match whether file_path refers to a log file that normally contains binary data.

Parameters:

file_path (Path) – The path to test.

Returns:

True if file_path is likely a binary log, or False otherwise.

Return type:

bool

snapm.fsdiff.filetypes._log_debug_fsdiff(msg, *args, **kwargs)

A wrapper for fsdiff subsystem debug logs.

snapm.fsdiff.fsdiffer module

Top-level fsdiff interface.

class snapm.fsdiff.fsdiffer.FsDiffer(manager: Manager, options: DiffOptions | None = None, cache: bool = True, cache_expires: int = -1, color: str = 'auto', term_control: TermControl | None = None)

Bases: object

Top-level interface for generating file system comparisons.

__init__(manager: Manager, options: DiffOptions | None = None, cache: bool = True, cache_expires: int = -1, color: str = 'auto', term_control: TermControl | None = None)

Initialise a new FsDiffer to compute file system differences.

Parameters:

• manager (Manager) – The manager context for snapshot operations (used by future methods).

• options (DiffOptions) – Options to control this FsDiffer instance.

• cache (bool) – Use diff cache if available.

• cache_expires (int) – Cache expiry in seconds. Use 0 to disable cache expiry or -1 to use the cache default (15m).

• color (str) – A string to control color tree rendering: “auto”, “always”, or “never”.

• term_control (Optional[TermControl]) – An optional TermControl instance to use for formatting. The supplied instance overrides any color argument if set.

_term_control: TermControl | None

compare_roots(mount_a: Mount, mount_b: Mount) → FsDiffResults

Compare two mounted snapshot sets and return diff results.

Parameters:

• mount_a (Mount) – The first (left hand) mount to compare.

• mount_b (Mount) – The second (right hand) mount to compare.

Returns:

The diff results for the comparison.

Return type:

FsDiffResults

diff_engine: DiffEngine

manager: Manager

Manager context for snapshot operations (used by future methods)

options: DiffOptions

tree_walker: TreeWalker

snapm.fsdiff.fsdiffer._MAX_RSS_FRACTION = 0.333

Maximum fraction of memory used to attempt diffing after building trees

snapm.fsdiff.fsdiffer._should_diff(options: DiffOptions) → bool

Determine whether it is safe to attempt computing the diff given system memory constraints and current state.

Parameters:

options (DiffOptions) – The effective options for this run.

Returns:

True if diffing should be attempted or False otherwise.

Return type:

bool

snapm.fsdiff.fsdiffer.close(theself)

Work around Magic destructor breakage.

Parameters:

theself (magic.Magic) – The magic.Magic instance.

snapm.fsdiff.options module

File system diff options and categories.

class snapm.fsdiff.options.DiffCategories(*values)

Bases: Enum

Enum for categories of difference based on type or file system location.

APPLICATION = 'application'

CRITICAL_SYSTEM = 'critical_system'

LOG_FILES = 'log_files'

PACKAGE_MANAGEMENT = 'package_mgmt'

TEMPORARY = 'temporary'

USER_DATA = 'user_data'

class snapm.fsdiff.options.DiffOptions(ignore_timestamps: bool = False, ignore_permissions: bool = False, ignore_ownership: bool = False, content_only: bool = False, include_system_dirs: bool = False, include_content_diffs: bool = True, use_magic_file_type: bool = False, follow_symlinks: bool = False, max_file_size: int = 0, max_content_diff_size: int = 1048576, max_content_hash_size: int = 1048576, file_patterns: Tuple[str, ...]=<factory>, exclude_patterns: Tuple[str, ...]=<factory>, from_path: str | Tuple[str] | None = ('/', ), quiet: bool = False, no_mem_check: bool = False)

Bases: object

File system comparison options.

__init__(ignore_timestamps: bool = False, ignore_permissions: bool = False, ignore_ownership: bool = False, content_only: bool = False, include_system_dirs: bool = False, include_content_diffs: bool = True, use_magic_file_type: bool = False, follow_symlinks: bool = False, max_file_size: int = 0, max_content_diff_size: int = 1048576, max_content_hash_size: int = 1048576, file_patterns: Tuple[str, ...]=<factory>, exclude_patterns: Tuple[str, ...]=<factory>, from_path: str | Tuple[str] | None = ('/', ), quiet: bool = False, no_mem_check: bool = False) → None

__repr__()

Return repr(self).

__str__()

Return a human readable string representation of this DiffOptions instance.

Returns:

A human readable string.

Return type:

str

content_only: bool = False

Only consider content changes

exclude_patterns: Tuple[str, ...]

File patterns to exclude (glob notation)

file_patterns: Tuple[str, ...]

File patterns to include (glob notation)

follow_symlinks: bool = False

Follow symlinks when walking file system trees

classmethod from_cmd_args(cmd_args: Namespace) → DiffOptions

Initialise DiffOptions from command line arguments.

Construct a new DiffOptions object from the command line arguments in cmd_args.

Parameters:

cmd_args (Namespace) – The command line selection arguments.

Returns:

A new DiffOptions instance

Return type:

DiffOptions

from_path: str | Tuple[str] | None = ('/',)

Start path(s) for file system comparison

ignore_ownership: bool = False

Ignore ownership in diff comparisons

ignore_permissions: bool = False

Ignore permissions in diff comparisons

ignore_timestamps: bool = False

Ignore timestamps in diff comparisons

include_content_diffs: bool = True

Generate content diffs for detected file modifications

include_system_dirs: bool = False

Include system directories in diff comparisons

max_content_diff_size: int = 1048576

Maximum size for generating content diffs

max_content_hash_size: int = 1048576

Maximum file size for generating content hashes

max_file_size: int = 0

Maximum file size for diff comparisons

no_mem_check: bool = False

Disable RSS memory pressure safety checks

quiet: bool = False

Do not output progress or status updates

use_magic_file_type: bool = False

Generate file type information using magic

snapm.fsdiff.treewalk module

Tree walking support for fsdiff.

class snapm.fsdiff.treewalk.FsEntry(path: str, strip_prefix: str, stat_info: stat_result, content_hash: str | None = None, file_type_info: FileTypeInfo | None = None)

Bases: object

Representation of a single file system entry for comparison.

__init__(path: str, strip_prefix: str, stat_info: stat_result, content_hash: str | None = None, file_type_info: FileTypeInfo | None = None)

Initialise a new FsEntry object.

Parameters:

• path (str) – The full path to this FsEntry represents.

• strip_prefix (str) – An optional leading path prefix (mount root) to strip from path before storing.

• stat_info (os.stat_result) – An os.stat_result for this FsEntry.

• content_hash (str) – A hash of the content of this FsEntry.

• file_type_info (FileTypeInfo) – A FileTypeInfo object for this FsEntry.

__str__()

Return a string representation of this FsEntry object.

Returns:

A human readable representation of this FsEntry.

Return type:

str

broken_symlink

Not a broken symlink

content_hash: str | None

Calculated content hash or None if not available.

file_type_info

An optional instance of FileTypeInfo describing this path.

full_path: Path

The full path from the host perspective

gid: int

File group returned by stat()

property is_block: bool

True if this FsEntry is a block special file.

Returns:

True if this FsEntry corresponds to a block special file or False otherwise.

Return type:

bool

property is_char: bool

True if this FsEntry is a character special file.

Returns:

True if this FsEntry corresponds to a character special file or False otherwise.

Return type:

bool

property is_dir: bool

True if this FsEntry is a directory.

Returns:

True if this FsEntry corresponds to a directory or False otherwise.

Return type:

bool

property is_fifo: bool

True if this FsEntry is a FIFO special file.

Returns:

True if this FsEntry corresponds to a FIFO special file or False otherwise.

Return type:

bool

property is_file: bool

True if this FsEntry is a regular file.

Returns:

True if this FsEntry corresponds to a regular file or False otherwise.

Return type:

bool

property is_sock: bool

True if this FsEntry is a socket.

Returns:

True if this FsEntry corresponds to a socket or False otherwise.

Return type:

bool

property is_symlink: bool

True if this FsEntry is a symlink.

Returns:

True if this FsEntry corresponds to a symlink or False otherwise.

Return type:

bool

property is_text_like: bool

True if this FsEntry is a text-like file.

Returns:

True if this FsEntry corresponds to a text-like file or False otherwise.

Return type:

bool

mode: int

File mode returned by stat()

mtime: float

File modification time returned by stat()

path: Path

The path relative to this tree’s mount point

size: int

File size returned by stat()

stat: stat_result

An os.stat_result for this path

symlink_target

The symbolic link target: None for non-symlinks

to_dict() → Dict[str, Any]

Convert this FsEntry object into a dictionary representation suitable for encoding as JSON.

Returns:

A dictionary mapping this instance’s keys to values.

Return type:

Dict[str, Any]

property type_desc

Return a string description of the entry type: “file”, “dir”, or “symlink”.

Returns:

A string description of the entry type.

Return type:

str

uid: int

File owner returned by stat()

xattrs: Dict[str, bytes]

Extended attributes as Dict[str, bytes]

class snapm.fsdiff.treewalk.TreeWalker(options, hash_algorithm: str = 'sha256')

Bases: object

Simple file system tree walker for comparisons.

__init__(options, hash_algorithm: str = 'sha256')

Initialise a new TreeWalker object.

Parameters:

• options (DiffOptions) – Options to control this TreeWalker instance.

• hash_algorithm (str) – A string describing the hash algorithm to be used for this TreeWalker instance.

_calculate_content_hash(file_path: str, strip_prefix: str = '') → str

Calculate content hash for regular files

Parameters:

• file_path (str) – The path to the file to hash.

• strip_prefix (str) – An optional leading path prefix to strip from stored path values.

Returns:

A string representation of the hash of the file content using the configured hash algorithm.

Return type:

str

_process_dir(dir_path: str, dir_stat: stat_result, strip_prefix: str) → FsEntry

Process a single directory at dir_path with stat data dir_stat.

Parameters:

• dir_path (str) – The path to examine.

• dir_stat (os.stat_result) – os.stat() result for dir_path.

Returns:

A new FsEntry representing the directory.

Return type:

FsEntry

_process_file(file_path: str, file_stat: stat_result, strip_prefix: str) → FsEntry

Process a single file at file_path with stat data file_stat.

Parameters:

• file_path (str) – The path to examine.

• file_stat (os.stat_result) – os.stat() result for file_path.

Returns:

A new FsEntry representing the file.

Return type:

FsEntry

exclude_patterns: Tuple[str, ...]

file_patterns: Tuple[str, ...] | None

file_type_detector: FileTypeDetector

hash_algorithm: str

options: DiffOptions

walk_tree(mount: Mount, strip_prefix: str = '', quiet: bool = False, term_control: TermControl | None = None) → Dict[str, FsEntry]

Walk filesystem tree starting from mount.root and return indexed entries.

Parameters:

• mount (snapm.manager._mounts.Mount) – A snapm.manager._mounts.Mount object representing the file system to walk.

• strip_prefix (str) – An optional leading path prefix to strip from stored path values.

• quiet (bool) – Suppress progress output.

• term_control (Optional[TermControl]) – Optional pre-initialised terminal control object.

Returns:

A dictionary mapping path strings to FsEntry objects.

Return type:

Dict[str, FsEntry]

snapm.fsdiff.treewalk._ALWAYS_EXCLUDE_PATTERNS = ('*/proc/kcore', '*/proc/kmsg', '*/proc/*/mem', '*/proc/*/pagemap', '*/proc/*/fd/*', '*/proc/*/task/*/mem', '*/proc/*/clear_refs', '*/proc/sys/fs/binfmt_misc/register', '*/proc/sys/net/ipv*/route/flush', '*/proc/sysrq-trigger', '*/proc/acpi/event', '/dev/zero', '/dev/full', '/dev/random', '/dev/urandom', '/dev/kmsg', '/dev/mem', '/dev/kmem', '/dev/port', '/dev/nvram', '/dev/console', '/dev/tty*', '/dev/pts/*', '/dev/ptmx', '/dev/input/*', '/dev/uinput', '/dev/watchdog*', '/dev/st*', '/dev/nst*', '*/sys/kernel/debug/*', '*/sys/kernel/tracing/trace_pipe', '*/sys/kernel/tracing/*/*/*_raw', '*/sys/kernel/tracing/*/*/trace_pipe', '*/sys/fs/cgroup/*', '*/sys/*/power/autosuspend_delay_ms', '*/sys/devices/pci*/rescan', '*/sys/*/bind', '*/sys/*/unbind', '*/sys/*/uevent')

Paths that are always excluded, no matter what the user says.

snapm.fsdiff.treewalk._EXCLUDE_SYSTEM_DIRS = ('/proc/*', '/sys/*', '/dev/*', '/tmp/*', '/run/*', '/var/run/*', '/var/lock/*')

Default system directories to exclude

snapm.fsdiff.treewalk._NO_CHUNK_PATHS = ('/proc/sys/fs/binfmt_misc/*', '/proc/sys/net/ipv*/conf/*', '/proc/irq/*/*', '/sys/*', '/dev/*')

Paths where we should avoid trying to read content in chunks.

snapm.fsdiff.treewalk._log_debug_fsdiff(msg, *args, **kwargs)

A wrapper for fsdiff subsystem debug logs.

snapm.fsdiff.treewalk._stat_to_dict(st: stat_result) → Dict[str, Any]

Convert an os.stat_result into a dictionary.

Parameters:

st (os.stat_result) – The stat result to convert.

Returns:

A dictionary mapping stat fields to values.

Return type:

Dict[str, Any]