Skip to content

Snapm

Snapm — snapshot manager for creating, scheduling, and reverting coordinated snapshots across multiple LVM volumes. Integrates with Boom for snapshot-based boot environments.

sts.snapm.base

Snapshot Manager base functionality.

SnapmBase pydantic-model

Bases: CliTool

Base class for Snapshot Manager operations.

Show JSON schema:
{
  "additionalProperties": false,
  "description": "Base class for Snapshot Manager operations.",
  "properties": {
    "debugopts": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Debugopts"
    },
    "verbose": {
      "default": false,
      "title": "Verbose",
      "type": "boolean"
    }
  },
  "title": "SnapmBase",
  "type": "object"
}

Fields:

  • debugopts (list[str] | None)
  • verbose (bool)
Source code in sts_libs/src/sts/snapm/base.py
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class SnapmBase(CliTool):
    """Base class for Snapshot Manager operations."""

    CLI_NAME: ClassVar[str] = CLI_NAME

    debugopts: list[str] | None = None
    verbose: bool = False

    def _global_args(self) -> list[str]:
        """Return global CLI arguments derived from ``verbose``/``debugopts``."""
        args: list[str] = []
        if self.verbose:
            args.append('--verbose')
        if self.debugopts:
            args.extend(self.debugopts)
        return args

    @cached_property
    def version(self) -> VersionInfo:
        """Get snapm version."""
        return VersionInfo.from_string(self.run_command(options={'--version': None}).stdout.strip())

version cached property

Get snapm version.

sts.snapm.plugin

Snapshot Manager plugin management.

Plugin pydantic-model

Bases: ReportModel

Snapshot Manager plugin report parsed from CLI JSON output.

Show JSON schema:
{
  "description": "Snapshot Manager plugin report parsed from CLI JSON output.",
  "properties": {
    "plugin_name": {
      "title": "Plugin Name",
      "type": "string"
    },
    "plugin_version": {
      "default": "",
      "title": "Plugin Version",
      "type": "string"
    },
    "plugin_type": {
      "default": "",
      "title": "Plugin Type",
      "type": "string"
    }
  },
  "required": [
    "plugin_name"
  ],
  "title": "Plugin",
  "type": "object"
}

Config:

  • populate_by_name: True

Fields:

  • name (str)
  • version (str)
  • plugin_type (str)
Source code in sts_libs/src/sts/snapm/plugin.py
23
24
25
26
27
28
29
30
class Plugin(ReportModel):
    """Snapshot Manager plugin report parsed from CLI JSON output."""

    model_config = ConfigDict(populate_by_name=True)

    name: str = Field(alias='plugin_name')
    version: str = Field(default='', alias='plugin_version')
    plugin_type: str = Field(default='', alias='plugin_type')

PluginManager pydantic-model

Bases: SnapmBase

Plugin management.

Show JSON schema:
{
  "additionalProperties": false,
  "description": "Plugin management.",
  "properties": {
    "debugopts": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Debugopts"
    },
    "verbose": {
      "default": false,
      "title": "Verbose",
      "type": "boolean"
    }
  },
  "title": "PluginManager",
  "type": "object"
}

Fields:

  • debugopts (list[str] | None)
  • verbose (bool)
Source code in sts_libs/src/sts/snapm/plugin.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class PluginManager(SnapmBase):
    """Plugin management."""

    # Class-level constants
    SUBCOMMAND: ClassVar[str] = 'plugin'

    def list_plugins(self, *, fields: str | None = None, json_output: bool = False) -> CommandResult:
        """List all plugins.

        Args:
            fields: Comma-separated list of fields to display
                   (e.g. "plugin_name,plugin_version,plugin_type")
            json_output: Whether to output in JSON format
        """
        options: SnapmOptions = {}

        # Add fields if provided
        if fields:
            options['--options'] = fields

        # Add JSON output if requested
        if json_output:
            options['--json'] = None

        return self.run_command(subcommand=self.SUBCOMMAND, action='list', options=options)

    def get_plugins(self) -> list[Plugin]:
        """Get all available plugins as Plugin instances."""
        plugins: list[Plugin] = []

        # Get plugin list in JSON format for easier parsing
        result = self.list_plugins(json_output=True)

        if result.failed:
            logger.error(f'Failed to list plugins: {result.stderr}')
            return plugins

        if not result.stdout:
            return plugins

        try:
            # Parse JSON output
            data: dict[str, Any] | list[Any] = json.loads(result.stdout)

            # Extract plugins from the 'Plugins' key
            items: list[dict[str, Any]] = data.get('Plugins', []) if isinstance(data, dict) else data

            # Create plugin instances from each entry
            for item in items:
                plugin = Plugin.model_validate(item)
                plugins.append(plugin)

        except json.JSONDecodeError as e:
            logger.warning(f'Failed to parse plugins (invalid JSON): {e}')
        except (TypeError, ValueError) as e:
            logger.warning(f'Failed to process plugin data: {e}')

        return plugins

get_plugins()

Get all available plugins as Plugin instances.

Source code in sts_libs/src/sts/snapm/plugin.py
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_plugins(self) -> list[Plugin]:
    """Get all available plugins as Plugin instances."""
    plugins: list[Plugin] = []

    # Get plugin list in JSON format for easier parsing
    result = self.list_plugins(json_output=True)

    if result.failed:
        logger.error(f'Failed to list plugins: {result.stderr}')
        return plugins

    if not result.stdout:
        return plugins

    try:
        # Parse JSON output
        data: dict[str, Any] | list[Any] = json.loads(result.stdout)

        # Extract plugins from the 'Plugins' key
        items: list[dict[str, Any]] = data.get('Plugins', []) if isinstance(data, dict) else data

        # Create plugin instances from each entry
        for item in items:
            plugin = Plugin.model_validate(item)
            plugins.append(plugin)

    except json.JSONDecodeError as e:
        logger.warning(f'Failed to parse plugins (invalid JSON): {e}')
    except (TypeError, ValueError) as e:
        logger.warning(f'Failed to process plugin data: {e}')

    return plugins

list_plugins(*, fields=None, json_output=False)

List all plugins.

Parameters:

Name Type Description Default
fields str | None

Comma-separated list of fields to display (e.g. "plugin_name,plugin_version,plugin_type")

None
json_output bool

Whether to output in JSON format

False
Source code in sts_libs/src/sts/snapm/plugin.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def list_plugins(self, *, fields: str | None = None, json_output: bool = False) -> CommandResult:
    """List all plugins.

    Args:
        fields: Comma-separated list of fields to display
               (e.g. "plugin_name,plugin_version,plugin_type")
        json_output: Whether to output in JSON format
    """
    options: SnapmOptions = {}

    # Add fields if provided
    if fields:
        options['--options'] = fields

    # Add JSON output if requested
    if json_output:
        options['--json'] = None

    return self.run_command(subcommand=self.SUBCOMMAND, action='list', options=options)

sts.snapm.schedule

Snapshot schedule management.

GcPolicyInfo pydantic-model

Bases: ReportModel

Garbage collection policy information.

Retention fields are policy-specific: keep_count for COUNT, keep_years/months/weeks/days for AGE, and keep_yearly/quarterly/monthly/weekly/daily/hourly for TIMELINE.

Show JSON schema:
{
  "description": "Garbage collection policy information.\n\nRetention fields are policy-specific: ``keep_count`` for COUNT,\n``keep_years/months/weeks/days`` for AGE, and\n``keep_yearly/quarterly/monthly/weekly/daily/hourly`` for TIMELINE.",
  "properties": {
    "policy_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Policy Name"
    },
    "policy_type": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Policy Type"
    },
    "keep_count": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Count"
    },
    "keep_years": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Years"
    },
    "keep_months": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Months"
    },
    "keep_weeks": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Weeks"
    },
    "keep_days": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Days"
    },
    "keep_yearly": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Yearly"
    },
    "keep_quarterly": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Quarterly"
    },
    "keep_monthly": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Monthly"
    },
    "keep_weekly": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Weekly"
    },
    "keep_daily": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Daily"
    },
    "keep_hourly": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Keep Hourly"
    }
  },
  "title": "GcPolicyInfo",
  "type": "object"
}

Fields:

  • policy_name (str | None)
  • policy_type (str | None)
  • keep_count (int | None)
  • keep_years (int | None)
  • keep_months (int | None)
  • keep_weeks (int | None)
  • keep_days (int | None)
  • keep_yearly (int | None)
  • keep_quarterly (int | None)
  • keep_monthly (int | None)
  • keep_weekly (int | None)
  • keep_daily (int | None)
  • keep_hourly (int | None)
Source code in sts_libs/src/sts/snapm/schedule.py
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class GcPolicyInfo(ReportModel):
    """Garbage collection policy information.

    Retention fields are policy-specific: ``keep_count`` for COUNT,
    ``keep_years/months/weeks/days`` for AGE, and
    ``keep_yearly/quarterly/monthly/weekly/daily/hourly`` for TIMELINE.
    """

    policy_name: str | None = None
    policy_type: str | None = None
    # COUNT policy
    keep_count: int | None = None
    # AGE policy
    keep_years: int | None = None
    keep_months: int | None = None
    keep_weeks: int | None = None
    keep_days: int | None = None
    # TIMELINE policy
    keep_yearly: int | None = None
    keep_quarterly: int | None = None
    keep_monthly: int | None = None
    keep_weekly: int | None = None
    keep_daily: int | None = None
    keep_hourly: int | None = None

Schedule pydantic-model

Bases: SnapmBase

Schedule management.

A Schedule defines automatic creation of snapshot sets.

Show JSON schema:
{
  "$defs": {
    "GcPolicyInfo": {
      "description": "Garbage collection policy information.\n\nRetention fields are policy-specific: ``keep_count`` for COUNT,\n``keep_years/months/weeks/days`` for AGE, and\n``keep_yearly/quarterly/monthly/weekly/daily/hourly`` for TIMELINE.",
      "properties": {
        "policy_name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Policy Name"
        },
        "policy_type": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Policy Type"
        },
        "keep_count": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Count"
        },
        "keep_years": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Years"
        },
        "keep_months": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Months"
        },
        "keep_weeks": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Weeks"
        },
        "keep_days": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Days"
        },
        "keep_yearly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Yearly"
        },
        "keep_quarterly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Quarterly"
        },
        "keep_monthly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Monthly"
        },
        "keep_weekly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Weekly"
        },
        "keep_daily": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Daily"
        },
        "keep_hourly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Hourly"
        }
      },
      "title": "GcPolicyInfo",
      "type": "object"
    },
    "ScheduleInfo": {
      "description": "Schedule information parsed from ``snapm schedule show`` output.",
      "properties": {
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "sources": {
          "items": {
            "type": "string"
          },
          "title": "Sources",
          "type": "array"
        },
        "default_size_policy": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Default Size Policy"
        },
        "autoindex": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Autoindex"
        },
        "calendarspec": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Calendarspec"
        },
        "boot": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Boot"
        },
        "revert": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Revert"
        },
        "gc_policy": {
          "anyOf": [
            {
              "$ref": "#/$defs/GcPolicyInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "enabled": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Enabled"
        },
        "running": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Running"
        },
        "next_elapse": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Next Elapse"
        }
      },
      "title": "ScheduleInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Schedule management.\n\nA Schedule defines automatic creation of snapshot sets.",
  "properties": {
    "debugopts": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Debugopts"
    },
    "verbose": {
      "default": false,
      "title": "Verbose",
      "type": "boolean"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "report": {
      "anyOf": [
        {
          "$ref": "#/$defs/ScheduleInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "title": "Schedule",
  "type": "object"
}

Fields:

  • debugopts (list[str] | None)
  • verbose (bool)
  • name (str | None)
  • report (ScheduleInfo | None)
Source code in sts_libs/src/sts/snapm/schedule.py
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
class Schedule(SnapmBase):
    """Schedule management.

    A Schedule defines automatic creation of snapshot sets.
    """

    SUBCOMMAND: ClassVar[str] = 'schedule'

    name: str | None = None
    report: ScheduleInfo | None = Field(default=None, repr=False)

    @staticmethod
    def _add_keep_options(options: SnapmOptions, keep_options: dict[str, str]) -> None:
        """Add keep_* options to the command options dictionary.

        Converts keep_* kwargs to their corresponding CLI options.
        Unknown options are also passed through by converting underscores to dashes.

        Args:
            options: Command options dictionary to update
            keep_options: Dictionary of keep_* options from kwargs
        """
        # Known keep options mapping: kwarg name -> CLI option
        known_keep_options = {
            'keep_count': '--keep-count',
            'keep_years': '--keep-years',
            'keep_months': '--keep-months',
            'keep_weeks': '--keep-weeks',
            'keep_days': '--keep-days',
            'keep_yearly': '--keep-yearly',
            'keep_quarterly': '--keep-quarterly',
            'keep_monthly': '--keep-monthly',
            'keep_weekly': '--keep-weekly',
            'keep_daily': '--keep-daily',
            'keep_hourly': '--keep-hourly',
        }

        for key, value in keep_options.items():
            if key in known_keep_options:
                options[known_keep_options[key]] = str(value)
            else:
                # Pass through unknown options by converting underscores to dashes
                cli_option = f'--{key.replace("_", "-")}'
                options[cli_option] = str(value)

    def refresh_report(self) -> bool:
        """Refresh schedule information from show and list commands."""
        if not self.name:
            logger.error('No schedule name available')
            return False

        result = self.show()

        if result.failed:
            logger.error(f'Failed to get schedule info: {result.stderr}')
            return False

        if not result.stdout:
            return False

        try:
            data_raw: list[dict[str, Any]] = json.loads(result.stdout)

            if not data_raw:
                logger.error(f'Failed to get schedule info: {result.stderr}')
                return False
            data = data_raw[0]

            # Create or update report from show command
            self.report = ScheduleInfo.model_validate(data)

            # Get runtime status from list command
            self._refresh_status()

        except json.JSONDecodeError as e:
            logger.warning(f'Failed to parse schedule info (invalid JSON): {e}')
            return False
        except (KeyError, TypeError) as e:
            logger.warning(f'Failed to process schedule info (unexpected format): {e}')
            return False

        return True

    def _refresh_status(self) -> None:
        """Update enabled/next_elapse from list output (not available from show)."""
        if not self.name or not self.report:
            return

        result = self.list_items(json_output=True)
        if result.failed or not result.stdout:
            return

        try:
            data: dict[str, Any] | list[Any] = json.loads(result.stdout)

            # Handle the Schedules wrapper
            items: list[dict[str, Any]]
            if isinstance(data, dict) and 'Schedules' in data:
                items = data['Schedules']
            elif isinstance(data, list):
                items = data
            else:
                return

            # Find matching schedule by name
            for item in items:
                item_name: str | None = item.get('schedule_name')
                if item_name == self.name:
                    enabled: bool | None = item.get('schedule_enabled')
                    next_elapse: str | None = item.get('schedule_nextelapse')
                    self.report = self.report.model_copy(
                        update={
                            'enabled': enabled,
                            'next_elapse': next_elapse,
                        }
                    )
                    break

        except (json.JSONDecodeError, TypeError, KeyError):
            # Status update is best-effort, don't fail if it doesn't work
            pass

    def create(
        self,
        schedule_name: str | None = None,
        sources: list[str] | None = None,
        policy_type: str | None = None,
        calendarspec: str | None = None,
        size_policy: str | None = None,
        *,
        bootable: bool = False,
        revert: bool = False,
        **keep_options: str,
    ) -> CommandResult:
        """Create a schedule.

        Args:
            schedule_name: Schedule name (uses self.name if not provided)
            sources: Source mount points or block devices
            policy_type: GC policy type (ALL, COUNT, AGE, TIMELINE)
            calendarspec: Calendar event expression for scheduling
            size_policy: Default size policy for sources
            bootable: Whether to create boot entries for snapsets
            revert: Whether to create revert entries for snapsets
            **keep_options: Retention kwargs matching the GC policy
                (e.g. ``keep_count=5``, ``keep_daily=7``, ``keep_weekly=4``)

        Example:
            ```python
            schedule = Schedule()
            schedule.create(
                'backup',
                ['/', '/home', '/var'],
                policy_type='TIMELINE',
                calendarspec='daily',
                keep_daily=7,
                keep_weekly=4,
                keep_monthly=12,
                bootable=True,
                revert=True,
            )
            ```
        """
        name = schedule_name or self.name

        options: SnapmOptions = {}

        # Policy type option
        if policy_type is not None:
            options['--policy-type'] = policy_type

        # Optional options
        if calendarspec:
            options['--calendarspec'] = calendarspec
        if size_policy:
            options['--size-policy'] = size_policy
        if bootable:
            options['--bootable'] = None
        if revert:
            options['--revert'] = None

        # Process keep_* options from kwargs
        self._add_keep_options(options, keep_options)

        # Build positional args
        positional_args: list[str] = []
        if name:
            positional_args.append(name)
        if sources:
            positional_args.extend(sources)

        result = self.run_command(
            subcommand=self.SUBCOMMAND, action='create', options=options, positional_args=positional_args
        ).assert_ok()

        if name:
            self.name = name
        self.refresh_report()

        return result

    def delete(self) -> CommandResult:
        """Delete a schedule. Existing snapsets are not deleted."""
        positional_args = [self.name] if self.name else []

        return self.run_command(
            subcommand=self.SUBCOMMAND, action='delete', positional_args=positional_args
        ).assert_ok()

    def edit(
        self,
        sources: list[str] | None = None,
        policy_type: str | None = None,
        calendarspec: str | None = None,
        size_policy: str | None = None,
        *,
        bootable: bool | None = None,
        revert: bool | None = None,
        **keep_options: str,
    ) -> CommandResult:
        """Edit a schedule. Only specified options are changed.

        Args:
            sources: Source mount points or block devices
            policy_type: GC policy type (ALL, COUNT, AGE, TIMELINE)
            calendarspec: Calendar event expression for scheduling
            size_policy: Default size policy for sources
            bootable: Whether to create boot entries for snapsets
            revert: Whether to create revert entries for snapsets
            **keep_options: Retention kwargs matching the GC policy
        """
        options: SnapmOptions = {}

        # Optional options - only add if specified
        if policy_type is not None:
            options['--policy-type'] = policy_type
        if calendarspec is not None:
            options['--calendarspec'] = calendarspec
        if size_policy is not None:
            options['--size-policy'] = size_policy
        if bootable is not None:
            options['--bootable'] = None
        if revert is not None:
            options['--revert'] = None

        # Process keep_* options from kwargs
        self._add_keep_options(options, keep_options)

        # Build positional args
        positional_args: list[str] = []
        if self.name:
            positional_args.append(self.name)
        if sources:
            positional_args.extend(sources)

        result = self.run_command(
            subcommand=self.SUBCOMMAND, action='edit', options=options, positional_args=positional_args
        ).assert_ok()

        self.refresh_report()

        return result

    def enable(self, *, start: bool = False) -> CommandResult:
        """Enable a schedule.

        Args:
            start: Whether to start the schedule immediately
        """
        options: SnapmOptions = {}
        if start:
            options['--start'] = None

        positional_args = [self.name] if self.name else []

        result = self.run_command(
            subcommand=self.SUBCOMMAND, action='enable', options=options, positional_args=positional_args
        ).assert_ok()

        self.refresh_report()

        return result

    def disable(self) -> CommandResult:
        """Disable a schedule."""
        positional_args = [self.name] if self.name else []

        result = self.run_command(
            subcommand=self.SUBCOMMAND, action='disable', positional_args=positional_args
        ).assert_ok()

        self.refresh_report()

        return result

    def gc(self, schedule_name: str | None = None) -> CommandResult:
        """Run garbage collection, deleting snapsets according to policy.

        Args:
            schedule_name: Schedule name (uses self.name if not provided)
        """
        name = schedule_name or self.name

        options: SnapmOptions = {}
        if name:
            options['--config'] = name

        return self.run_command(subcommand=self.SUBCOMMAND, action='gc', options=options)

    def list_items(self, *, fields: str | None = None, json_output: bool = False) -> CommandResult:
        """List all schedules.

        Args:
            fields: Comma-separated list of fields to display
            json_output: Whether to output in JSON format
        """
        options: SnapmOptions = {}

        if fields:
            options['--options'] = fields

        if json_output:
            options['--json'] = None

        return self.run_command(subcommand=self.SUBCOMMAND, action='list', options=options)

    def show(self, *, json_output: bool = True) -> CommandResult:
        """Show detailed schedule information."""
        options: SnapmOptions = {}

        if json_output:
            options['--json'] = None

        positional_args: list[str] = []
        if self.name:
            positional_args.append(self.name)

        return self.run_command(
            subcommand=self.SUBCOMMAND, action='show', options=options, positional_args=positional_args
        )

    @classmethod
    def get_all(cls) -> list[Schedule]:
        """Get all schedules with their reports populated."""
        schedules: list[Schedule] = []

        base = cls()

        # Use 'list --json' to get all schedules (show without name returns empty)
        options: SnapmOptions = {'--json': None}
        result = base.run_command(subcommand=cls.SUBCOMMAND, action='list', options=options)

        if result.failed or not result.stdout:
            return schedules

        try:
            data: dict[str, Any] | list[Any] = json.loads(result.stdout)

            # Handle the Schedules wrapper from list output
            items: list[dict[str, Any]]
            if isinstance(data, dict) and 'Schedules' in data:
                items = data['Schedules']
            elif isinstance(data, list):
                items = data
            else:
                items = [data]

            for item in items:
                # List output uses 'schedule_name' key
                schedule_name: str | None = item.get('schedule_name') or item.get('name')
                if schedule_name:
                    schedule = cls(name=schedule_name)
                    schedule.refresh_report()
                    schedules.append(schedule)

        except (json.JSONDecodeError, TypeError) as e:
            logger.warning(f'Failed to parse schedules: {e}')

        return schedules

create(schedule_name=None, sources=None, policy_type=None, calendarspec=None, size_policy=None, *, bootable=False, revert=False, **keep_options)

Create a schedule.

Parameters:

Name Type Description Default
schedule_name str | None

Schedule name (uses self.name if not provided)

None
sources list[str] | None

Source mount points or block devices

None
policy_type str | None

GC policy type (ALL, COUNT, AGE, TIMELINE)

None
calendarspec str | None

Calendar event expression for scheduling

None
size_policy str | None

Default size policy for sources

None
bootable bool

Whether to create boot entries for snapsets

False
revert bool

Whether to create revert entries for snapsets

False
**keep_options str

Retention kwargs matching the GC policy (e.g. keep_count=5, keep_daily=7, keep_weekly=4)

{}
Example
schedule = Schedule()
schedule.create(
    'backup',
    ['/', '/home', '/var'],
    policy_type='TIMELINE',
    calendarspec='daily',
    keep_daily=7,
    keep_weekly=4,
    keep_monthly=12,
    bootable=True,
    revert=True,
)
Source code in sts_libs/src/sts/snapm/schedule.py
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def create(
    self,
    schedule_name: str | None = None,
    sources: list[str] | None = None,
    policy_type: str | None = None,
    calendarspec: str | None = None,
    size_policy: str | None = None,
    *,
    bootable: bool = False,
    revert: bool = False,
    **keep_options: str,
) -> CommandResult:
    """Create a schedule.

    Args:
        schedule_name: Schedule name (uses self.name if not provided)
        sources: Source mount points or block devices
        policy_type: GC policy type (ALL, COUNT, AGE, TIMELINE)
        calendarspec: Calendar event expression for scheduling
        size_policy: Default size policy for sources
        bootable: Whether to create boot entries for snapsets
        revert: Whether to create revert entries for snapsets
        **keep_options: Retention kwargs matching the GC policy
            (e.g. ``keep_count=5``, ``keep_daily=7``, ``keep_weekly=4``)

    Example:
        ```python
        schedule = Schedule()
        schedule.create(
            'backup',
            ['/', '/home', '/var'],
            policy_type='TIMELINE',
            calendarspec='daily',
            keep_daily=7,
            keep_weekly=4,
            keep_monthly=12,
            bootable=True,
            revert=True,
        )
        ```
    """
    name = schedule_name or self.name

    options: SnapmOptions = {}

    # Policy type option
    if policy_type is not None:
        options['--policy-type'] = policy_type

    # Optional options
    if calendarspec:
        options['--calendarspec'] = calendarspec
    if size_policy:
        options['--size-policy'] = size_policy
    if bootable:
        options['--bootable'] = None
    if revert:
        options['--revert'] = None

    # Process keep_* options from kwargs
    self._add_keep_options(options, keep_options)

    # Build positional args
    positional_args: list[str] = []
    if name:
        positional_args.append(name)
    if sources:
        positional_args.extend(sources)

    result = self.run_command(
        subcommand=self.SUBCOMMAND, action='create', options=options, positional_args=positional_args
    ).assert_ok()

    if name:
        self.name = name
    self.refresh_report()

    return result

delete()

Delete a schedule. Existing snapsets are not deleted.

Source code in sts_libs/src/sts/snapm/schedule.py
266
267
268
269
270
271
272
def delete(self) -> CommandResult:
    """Delete a schedule. Existing snapsets are not deleted."""
    positional_args = [self.name] if self.name else []

    return self.run_command(
        subcommand=self.SUBCOMMAND, action='delete', positional_args=positional_args
    ).assert_ok()

disable()

Disable a schedule.

Source code in sts_libs/src/sts/snapm/schedule.py
348
349
350
351
352
353
354
355
356
357
358
def disable(self) -> CommandResult:
    """Disable a schedule."""
    positional_args = [self.name] if self.name else []

    result = self.run_command(
        subcommand=self.SUBCOMMAND, action='disable', positional_args=positional_args
    ).assert_ok()

    self.refresh_report()

    return result

edit(sources=None, policy_type=None, calendarspec=None, size_policy=None, *, bootable=None, revert=None, **keep_options)

Edit a schedule. Only specified options are changed.

Parameters:

Name Type Description Default
sources list[str] | None

Source mount points or block devices

None
policy_type str | None

GC policy type (ALL, COUNT, AGE, TIMELINE)

None
calendarspec str | None

Calendar event expression for scheduling

None
size_policy str | None

Default size policy for sources

None
bootable bool | None

Whether to create boot entries for snapsets

None
revert bool | None

Whether to create revert entries for snapsets

None
**keep_options str

Retention kwargs matching the GC policy

{}
Source code in sts_libs/src/sts/snapm/schedule.py
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
def edit(
    self,
    sources: list[str] | None = None,
    policy_type: str | None = None,
    calendarspec: str | None = None,
    size_policy: str | None = None,
    *,
    bootable: bool | None = None,
    revert: bool | None = None,
    **keep_options: str,
) -> CommandResult:
    """Edit a schedule. Only specified options are changed.

    Args:
        sources: Source mount points or block devices
        policy_type: GC policy type (ALL, COUNT, AGE, TIMELINE)
        calendarspec: Calendar event expression for scheduling
        size_policy: Default size policy for sources
        bootable: Whether to create boot entries for snapsets
        revert: Whether to create revert entries for snapsets
        **keep_options: Retention kwargs matching the GC policy
    """
    options: SnapmOptions = {}

    # Optional options - only add if specified
    if policy_type is not None:
        options['--policy-type'] = policy_type
    if calendarspec is not None:
        options['--calendarspec'] = calendarspec
    if size_policy is not None:
        options['--size-policy'] = size_policy
    if bootable is not None:
        options['--bootable'] = None
    if revert is not None:
        options['--revert'] = None

    # Process keep_* options from kwargs
    self._add_keep_options(options, keep_options)

    # Build positional args
    positional_args: list[str] = []
    if self.name:
        positional_args.append(self.name)
    if sources:
        positional_args.extend(sources)

    result = self.run_command(
        subcommand=self.SUBCOMMAND, action='edit', options=options, positional_args=positional_args
    ).assert_ok()

    self.refresh_report()

    return result

enable(*, start=False)

Enable a schedule.

Parameters:

Name Type Description Default
start bool

Whether to start the schedule immediately

False
Source code in sts_libs/src/sts/snapm/schedule.py
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def enable(self, *, start: bool = False) -> CommandResult:
    """Enable a schedule.

    Args:
        start: Whether to start the schedule immediately
    """
    options: SnapmOptions = {}
    if start:
        options['--start'] = None

    positional_args = [self.name] if self.name else []

    result = self.run_command(
        subcommand=self.SUBCOMMAND, action='enable', options=options, positional_args=positional_args
    ).assert_ok()

    self.refresh_report()

    return result

gc(schedule_name=None)

Run garbage collection, deleting snapsets according to policy.

Parameters:

Name Type Description Default
schedule_name str | None

Schedule name (uses self.name if not provided)

None
Source code in sts_libs/src/sts/snapm/schedule.py
360
361
362
363
364
365
366
367
368
369
370
371
372
def gc(self, schedule_name: str | None = None) -> CommandResult:
    """Run garbage collection, deleting snapsets according to policy.

    Args:
        schedule_name: Schedule name (uses self.name if not provided)
    """
    name = schedule_name or self.name

    options: SnapmOptions = {}
    if name:
        options['--config'] = name

    return self.run_command(subcommand=self.SUBCOMMAND, action='gc', options=options)

get_all() classmethod

Get all schedules with their reports populated.

Source code in sts_libs/src/sts/snapm/schedule.py
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
@classmethod
def get_all(cls) -> list[Schedule]:
    """Get all schedules with their reports populated."""
    schedules: list[Schedule] = []

    base = cls()

    # Use 'list --json' to get all schedules (show without name returns empty)
    options: SnapmOptions = {'--json': None}
    result = base.run_command(subcommand=cls.SUBCOMMAND, action='list', options=options)

    if result.failed or not result.stdout:
        return schedules

    try:
        data: dict[str, Any] | list[Any] = json.loads(result.stdout)

        # Handle the Schedules wrapper from list output
        items: list[dict[str, Any]]
        if isinstance(data, dict) and 'Schedules' in data:
            items = data['Schedules']
        elif isinstance(data, list):
            items = data
        else:
            items = [data]

        for item in items:
            # List output uses 'schedule_name' key
            schedule_name: str | None = item.get('schedule_name') or item.get('name')
            if schedule_name:
                schedule = cls(name=schedule_name)
                schedule.refresh_report()
                schedules.append(schedule)

    except (json.JSONDecodeError, TypeError) as e:
        logger.warning(f'Failed to parse schedules: {e}')

    return schedules

list_items(*, fields=None, json_output=False)

List all schedules.

Parameters:

Name Type Description Default
fields str | None

Comma-separated list of fields to display

None
json_output bool

Whether to output in JSON format

False
Source code in sts_libs/src/sts/snapm/schedule.py
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
def list_items(self, *, fields: str | None = None, json_output: bool = False) -> CommandResult:
    """List all schedules.

    Args:
        fields: Comma-separated list of fields to display
        json_output: Whether to output in JSON format
    """
    options: SnapmOptions = {}

    if fields:
        options['--options'] = fields

    if json_output:
        options['--json'] = None

    return self.run_command(subcommand=self.SUBCOMMAND, action='list', options=options)

refresh_report()

Refresh schedule information from show and list commands.

Source code in sts_libs/src/sts/snapm/schedule.py
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def refresh_report(self) -> bool:
    """Refresh schedule information from show and list commands."""
    if not self.name:
        logger.error('No schedule name available')
        return False

    result = self.show()

    if result.failed:
        logger.error(f'Failed to get schedule info: {result.stderr}')
        return False

    if not result.stdout:
        return False

    try:
        data_raw: list[dict[str, Any]] = json.loads(result.stdout)

        if not data_raw:
            logger.error(f'Failed to get schedule info: {result.stderr}')
            return False
        data = data_raw[0]

        # Create or update report from show command
        self.report = ScheduleInfo.model_validate(data)

        # Get runtime status from list command
        self._refresh_status()

    except json.JSONDecodeError as e:
        logger.warning(f'Failed to parse schedule info (invalid JSON): {e}')
        return False
    except (KeyError, TypeError) as e:
        logger.warning(f'Failed to process schedule info (unexpected format): {e}')
        return False

    return True

show(*, json_output=True)

Show detailed schedule information.

Source code in sts_libs/src/sts/snapm/schedule.py
391
392
393
394
395
396
397
398
399
400
401
402
403
404
def show(self, *, json_output: bool = True) -> CommandResult:
    """Show detailed schedule information."""
    options: SnapmOptions = {}

    if json_output:
        options['--json'] = None

    positional_args: list[str] = []
    if self.name:
        positional_args.append(self.name)

    return self.run_command(
        subcommand=self.SUBCOMMAND, action='show', options=options, positional_args=positional_args
    )

ScheduleInfo pydantic-model

Bases: ReportModel

Schedule information parsed from snapm schedule show output.

Show JSON schema:
{
  "$defs": {
    "GcPolicyInfo": {
      "description": "Garbage collection policy information.\n\nRetention fields are policy-specific: ``keep_count`` for COUNT,\n``keep_years/months/weeks/days`` for AGE, and\n``keep_yearly/quarterly/monthly/weekly/daily/hourly`` for TIMELINE.",
      "properties": {
        "policy_name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Policy Name"
        },
        "policy_type": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Policy Type"
        },
        "keep_count": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Count"
        },
        "keep_years": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Years"
        },
        "keep_months": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Months"
        },
        "keep_weeks": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Weeks"
        },
        "keep_days": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Days"
        },
        "keep_yearly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Yearly"
        },
        "keep_quarterly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Quarterly"
        },
        "keep_monthly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Monthly"
        },
        "keep_weekly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Weekly"
        },
        "keep_daily": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Daily"
        },
        "keep_hourly": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Keep Hourly"
        }
      },
      "title": "GcPolicyInfo",
      "type": "object"
    }
  },
  "description": "Schedule information parsed from ``snapm schedule show`` output.",
  "properties": {
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "sources": {
      "items": {
        "type": "string"
      },
      "title": "Sources",
      "type": "array"
    },
    "default_size_policy": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Default Size Policy"
    },
    "autoindex": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Autoindex"
    },
    "calendarspec": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Calendarspec"
    },
    "boot": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Boot"
    },
    "revert": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Revert"
    },
    "gc_policy": {
      "anyOf": [
        {
          "$ref": "#/$defs/GcPolicyInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "enabled": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Enabled"
    },
    "running": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Running"
    },
    "next_elapse": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Next Elapse"
    }
  },
  "title": "ScheduleInfo",
  "type": "object"
}

Fields:

  • name (str | None)
  • sources (list[str])
  • default_size_policy (str | None)
  • autoindex (bool | None)
  • calendarspec (str | None)
  • boot (bool | None)
  • revert (bool | None)
  • gc_policy (GcPolicyInfo | None)
  • enabled (bool | None)
  • running (bool | None)
  • next_elapse (str | None)
Source code in sts_libs/src/sts/snapm/schedule.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class ScheduleInfo(ReportModel):
    """Schedule information parsed from ``snapm schedule show`` output."""

    name: str | None = None
    sources: list[str] = Field(default_factory=list)
    default_size_policy: str | None = None
    autoindex: bool | None = None
    calendarspec: str | None = None
    boot: bool | None = None
    revert: bool | None = None
    gc_policy: GcPolicyInfo | None = None
    enabled: bool | None = None
    running: bool | None = None
    next_elapse: str | None = None

sts.snapm.snapset

Snapshot set management.

Snapset pydantic-model

Bases: SnapmBase

Snapset management.

A Snapset is a collection of snapshots across multiple filesystems.

Show JSON schema:
{
  "$defs": {
    "SnapsetInfo": {
      "description": "Snapset information parsed from ``snapm snapset show`` output.",
      "properties": {
        "SnapsetName": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Snapsetname"
        },
        "Sources": {
          "items": {
            "type": "string"
          },
          "title": "Sources",
          "type": "array"
        },
        "MountPoints": {
          "items": {
            "type": "string"
          },
          "title": "Mountpoints",
          "type": "array"
        },
        "Devices": {
          "items": {
            "type": "string"
          },
          "title": "Devices",
          "type": "array"
        },
        "NrSnapshots": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Nrsnapshots"
        },
        "Timestamp": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Timestamp"
        },
        "Time": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Time"
        },
        "UUID": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uuid"
        },
        "Status": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Status"
        },
        "Autoactivate": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Autoactivate"
        },
        "Bootable": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Bootable"
        },
        "BootEntries": {
          "additionalProperties": {
            "type": "string"
          },
          "title": "Bootentries",
          "type": "object"
        },
        "Snapshots": {
          "items": {
            "$ref": "#/$defs/SnapshotInfo"
          },
          "title": "Snapshots",
          "type": "array"
        }
      },
      "title": "SnapsetInfo",
      "type": "object"
    },
    "SnapshotInfo": {
      "description": "Snapshot information parsed from ``snapm snapshot show`` output.",
      "properties": {
        "Name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "SnapsetName": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Snapsetname"
        },
        "Origin": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Origin"
        },
        "Timestamp": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Timestamp"
        },
        "Time": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Time"
        },
        "Source": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Source"
        },
        "MountPoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "Provider": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Provider"
        },
        "UUID": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uuid"
        },
        "Status": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Status"
        },
        "Size": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "Free": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Free"
        },
        "SizeBytes": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Sizebytes"
        },
        "FreeBytes": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Freebytes"
        },
        "Autoactivate": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Autoactivate"
        },
        "DevicePath": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Devicepath"
        }
      },
      "title": "SnapshotInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Snapset management.\n\nA Snapset is a collection of snapshots across multiple filesystems.",
  "properties": {
    "debugopts": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Debugopts"
    },
    "verbose": {
      "default": false,
      "title": "Verbose",
      "type": "boolean"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "report": {
      "anyOf": [
        {
          "$ref": "#/$defs/SnapsetInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "title": "Snapset",
  "type": "object"
}

Fields:

  • debugopts (list[str] | None)
  • verbose (bool)
  • name (str | None)
  • uuid (str | None)
  • report (SnapsetInfo | None)
Source code in sts_libs/src/sts/snapm/snapset.py
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
class Snapset(SnapmBase):
    """Snapset management.

    A Snapset is a collection of snapshots across multiple filesystems.
    """

    SUBCOMMAND: ClassVar[str] = 'snapset'

    name: str | None = None
    uuid: str | None = None
    report: SnapsetInfo | None = Field(default=None, repr=False)

    def refresh_report(self) -> bool:
        """Refresh snapset information from show command."""
        result = self.show()

        if result.failed:
            logger.error(f'Failed to get snapset info: {result.stderr}')
            return False

        if not result.stdout:
            return False

        try:
            data_raw: list[dict[str, Any]] = json.loads(result.stdout)

            data: dict[str, Any]
            if not data_raw:
                logger.error(f'Failed to get snapshot info: {result.stderr}')
                return False
            data = data_raw[0]

            # Create or update report
            self.report = SnapsetInfo.model_validate(data)

        except json.JSONDecodeError as e:
            logger.warning(f'Failed to parse snapset info (invalid JSON): {e}')
            return False
        except (KeyError, TypeError) as e:
            logger.warning(f'Failed to process snapset info (unexpected format): {e}')
            return False

        return True

    def _common_operation(
        self, action: str, options: SnapmOptions | None = None, positional_args: list[str] | None = None
    ) -> CommandResult:
        """Run a snapset action using stored name/UUID, refreshing report on success.

        Raises:
            SnapmError: If no snapset identifier is available.
        """
        if not options:
            options = {}
        if not positional_args:
            positional_args = []

        # Use stored attributes
        if self.name:
            options['--name'] = self.name
        elif self.uuid:
            options['--uuid'] = self.uuid
        else:
            raise SnapmError('No snapset identifier available')

        result = self.run_command(
            subcommand=self.SUBCOMMAND, action=action, options=options, positional_args=positional_args
        ).assert_ok()

        # Refresh information
        if action != 'delete':
            self.refresh_report()

        return result

    def create(
        self,
        snapset_name: str | None = None,
        sources: list[str] | None = None,
        size_policy: str | None = None,
        *,
        bootable: bool = False,
        revert: bool = False,
        autoindex: bool = False,
    ) -> CommandResult:
        """Create a snapset.

        Args:
            snapset_name: Snapset name (uses self.name if not provided)
            sources: Source paths, optionally with size policy (e.g. ``'/mnt/data:2G'``)
            size_policy: Default size policy for all sources (e.g. "50%FREE", "2G")
            bootable: Whether to create a boot entry for this snapset
            revert: Whether to create a revert entry for this snapset
            autoindex: Treat name as basename and append a unique index

        Raises:
            SnapmError: If the snapset name or sources are missing.
        """
        # Use instance name if not provided
        name = snapset_name or self.name
        if not name:
            raise SnapmError('Snapset name is required')

        # Ensure sources is a list
        if sources is None:
            raise SnapmError('Snapset sources are required')

        options: SnapmOptions = {}
        if bootable:
            options['--bootable'] = None
        if revert:
            options['--revert'] = None
        if size_policy:
            options['--size-policy'] = size_policy
        if autoindex:
            options['--autoindex'] = None

        # Add sources to positional args
        positional_args = [name]
        positional_args.extend(sources)

        result = self.run_command(
            subcommand=self.SUBCOMMAND, action='create', options=options, positional_args=positional_args
        ).assert_ok()

        if autoindex:
            self.name = self._resolve_autoindex_name(name)
        else:
            self.name = name
        self.refresh_report()

        return result

    def _resolve_autoindex_name(self, basename: str) -> str:
        """Find the actual snapset name after autoindex creation.

        With ``--autoindex``, snapm appends an index (e.g. ``mysnap`` becomes
        ``mysnap.0``). Returns the most recently created match, or *basename*
        as fallback.
        """
        list_result = self.run_command(
            subcommand=self.SUBCOMMAND,
            action='list',
            options={'--json': None},
        )
        if list_result.failed or not list_result.stdout:
            return basename

        try:
            data: dict[str, Any] | list[Any] = json.loads(list_result.stdout)
        except (json.JSONDecodeError, TypeError):
            return basename

        entries: list[dict[str, Any]] = data.get('Snapsets', []) if isinstance(data, dict) else data

        candidates: list[dict[str, Any]] = [
            e for e in entries if isinstance(e.get('snapset_name'), str) and e['snapset_name'].startswith(basename)
        ]
        if not candidates:
            return basename

        candidates.sort(key=lambda e: str(e.get('snapset_time', '')), reverse=True)
        resolved: str = candidates[0]['snapset_name']
        return resolved

    def create_scheduled(self, schedule_name: str) -> CommandResult:
        """Create snapset from a schedule configuration.

        Does not call ``.assert_ok()`` internally -- callers should check
        ``result.failed`` or call ``.assert_ok()`` themselves.

        Raises:
            SnapmError: If schedule_name is missing.
        """
        if not schedule_name:
            raise SnapmError('Schedule name is required')

        options: SnapmOptions = {'--config': schedule_name}

        return self.run_command(subcommand=self.SUBCOMMAND, action='create-scheduled', options=options)

    def delete(self) -> CommandResult:
        """Delete a snapset and all associated snapshots."""
        return self._common_operation('delete')

    def rename(self, new_name: str) -> CommandResult:
        """Rename a snapset.

        Raises:
            SnapmError: If the current snapset name is missing.
        """
        if not self.name:
            raise SnapmError('Current snapset name is required')

        result = self.run_command(
            subcommand=self.SUBCOMMAND, action='rename', positional_args=[self.name, new_name]
        ).assert_ok()

        # Update instance name
        self.name = new_name
        self.refresh_report()

        return result

    def revert(self) -> CommandResult:
        """Revert to the state captured in the snapset. May require reboot."""
        return self._common_operation('revert')

    def resize(self, sources: list[str] | None = None, size_policy: str | None = None) -> CommandResult:
        """Resize snapset.

        Args:
            sources: Sources with optional size policies (e.g. ``['/path:2G']``)
            size_policy: Default size policy to apply (e.g. "25%FREE")
        """
        options: SnapmOptions = {}
        positional_args: list[str] = []

        if size_policy:
            options['--size-policy'] = size_policy

        # Add sources if provided
        if sources:
            positional_args.extend(sources)

        return self._common_operation('resize', options=options, positional_args=positional_args)

    def activate(self) -> CommandResult:
        """Activate a snapset."""
        return self._common_operation('activate')

    def deactivate(self) -> CommandResult:
        """Deactivate a snapset."""
        return self._common_operation('deactivate')

    def autoactivate(self, *, enable: bool = True) -> CommandResult:
        """Enable or disable auto-activation for a snapset."""
        options: SnapmOptions = {}

        # Add yes/no option
        if enable:
            options['--yes'] = None
        else:
            options['--no'] = None

        return self._common_operation('autoactivate', options)

    def list_items(self, *, fields: str | None = None, json_output: bool = False) -> CommandResult:
        """List snapsets, optionally filtered by stored name/UUID.

        Args:
            fields: Comma-separated list of fields to display
            json_output: Whether to output in JSON format
        """
        options: SnapmOptions = {}

        # Add name/uuid if set in this instance
        if self.name:
            options['--name'] = self.name
        elif self.uuid:
            options['--uuid'] = self.uuid

        # Add fields if provided
        if fields:
            options['--options'] = fields

        # Add JSON output if requested
        if json_output:
            options['--json'] = None

        return self.run_command(subcommand=self.SUBCOMMAND, action='list', options=options)

    def show(self, *, include_members: bool = True, json_output: bool = True) -> CommandResult:
        """Show detailed snapset information."""
        options: SnapmOptions = {}

        # Use stored attributes
        if self.name:
            options['--name'] = self.name
        elif self.uuid:
            options['--uuid'] = self.uuid
        else:
            logger.error('No snapset identifier available')
            return self.run_command(subcommand=self.SUBCOMMAND, action='show')

        # Add member and json options
        if include_members:
            options['--members'] = None

        if json_output:
            options['--json'] = None

        return self.run_command(subcommand=self.SUBCOMMAND, action='show', options=options)

    def prune(self, sources: list[str] | None = None) -> CommandResult:
        """Remove sources (and their snapshots) from the snapset. Cannot be undone."""
        positional_args: list[str] = []

        if self.name:
            positional_args.append(self.name)
        if sources:
            positional_args.extend(sources)

        result = self.run_command(
            subcommand=self.SUBCOMMAND, action='prune', positional_args=positional_args
        ).assert_ok()
        self.refresh_report()

        return result

    def split(self, new_name: str | None = None, sources: list[str] | None = None) -> Snapset:
        """Move specified sources into a new snapset, keeping the rest here."""
        positional_args: list[str] = [self.name or '']

        # Add sources if provided
        if new_name:
            positional_args.append(new_name)
        if sources:
            positional_args.extend(sources)

        result = self.run_command(subcommand=self.SUBCOMMAND, action='split', positional_args=positional_args)

        if result.failed:
            logger.error('Failed to split snapset')
            return Snapset()
        self.refresh_report()
        new_snapset = Snapset(name=new_name)
        new_snapset.refresh_report()
        return new_snapset

    @classmethod
    def get_all(cls) -> list[Snapset]:
        """Get all snapsets with their reports populated."""
        snapsets: list[Snapset] = []

        # Create base instance to run command
        base = cls()

        # Get snapset with members in JSON format for easier parsing
        options: SnapmOptions = {'--json': None, '--members': None}
        result = base.run_command(subcommand=cls.SUBCOMMAND, action='show', options=options)

        if result.failed or not result.stdout:
            return snapsets

        try:
            # Parse JSON output
            data: dict[str, Any] | list[Any] = json.loads(result.stdout)

            # Handle different possible JSON structures
            items: list[dict[str, Any]]
            if isinstance(data, list):
                items = data
            elif 'snapsets' in data:
                items = data['snapsets']
            else:
                items = [data]

            # Create snapset instances from each entry
            for item in items:
                snapset = cls()

                # Set basic attributes
                if 'SnapsetName' in item:
                    snapset.name = item['SnapsetName']
                if 'UUID' in item:
                    snapset.uuid = item['UUID']

                # Create report object
                snapset.report = SnapsetInfo.model_validate(item)

                snapsets.append(snapset)

        except (json.JSONDecodeError, TypeError) as e:
            logger.warning(f'Failed to parse snapsets: {e}')

        return snapsets

    def get_snapshots(self) -> list[SnapshotInfo]:
        """Get all snapshots in this snapset, refreshing the report first."""
        # Refresh report to ensure we have the latest data including snapshots
        if not self.refresh_report():
            return []

        # Return snapshots from report if available
        if self.report and self.report.snapshots:
            return self.report.snapshots

        return []

    def mount(self, *args: str, **kwargs: str | None) -> CommandResult:
        """Mount the members of a snapshot set at ``/run/snapm/mounts/<name>``."""
        options = self._kwargs_to_options(kwargs)
        positional_args = list(args)

        # Use instance name if no positional args provided
        if not positional_args and self.name:
            positional_args.append(self.name)

        return self.run_command(
            subcommand=self.SUBCOMMAND,
            action='mount',
            options=options,
            positional_args=positional_args,
        )

    def umount(self, *args: str, **kwargs: str | None) -> CommandResult:
        """Unmount the members of a snapshot set."""
        options = self._kwargs_to_options(kwargs)
        positional_args = list(args)

        # Use instance name if no positional args provided
        if not positional_args and self.name:
            positional_args.append(self.name)

        return self.run_command(
            subcommand=self.SUBCOMMAND,
            action='umount',
            options=options,
            positional_args=positional_args,
        )

    def exec_cmd(self, *args: str, **kwargs: str | None) -> CommandResult:
        """Execute a command inside the mounted snapshot set.

        Args:
            *args: Snapset name (if not set on instance) followed by command and its args.
        """
        options = self._kwargs_to_options(kwargs)
        positional_args = list(args)

        if self.name and positional_args:
            positional_args = [self.name, *positional_args]
        elif self.name and not positional_args:
            positional_args = [self.name]

        return self.run_command(
            subcommand=self.SUBCOMMAND,
            action='exec',
            options=options,
            positional_args=positional_args,
        )

    def shell(self, *args: str, **kwargs: str | None) -> CommandResult:
        """Start an interactive shell in the mounted snapshot set."""
        options = self._kwargs_to_options(kwargs)
        positional_args = list(args)

        # Use instance name if no positional args provided
        if not positional_args and self.name:
            positional_args.append(self.name)

        return self.run_command(
            subcommand=self.SUBCOMMAND,
            action='shell',
            options=options,
            positional_args=positional_args,
        )

    def diff(self, *args: str, **kwargs: str | None) -> CommandResult:
        """Compare two snapshot sets or a snapshot set and the running system.

        Args:
            *args: 'from' and 'to' targets (snapset name or '.' for running system)
            **kwargs: CLI options (underscores become dashes). Key options:
                ``output_format`` (paths/full/short/json/diff/summary/tree),
                ``ignore_timestamps``, ``ignore_permissions``, ``content_only``,
                ``include_pattern``, ``exclude_pattern``, ``start_path``

        Example:
            ```python
            result = snapset.diff('before-upgrade', '.', output_format='tree')
            ```
        """
        options = self._kwargs_to_options(kwargs)
        positional_args = list(args)

        return self.run_command(
            subcommand=self.SUBCOMMAND,
            action='diff',
            options=options,
            positional_args=positional_args,
        )

    def diffreport(self, *args: str, **kwargs: str | None) -> CommandResult:
        """Compare snapshot sets and output results in a tabular report.

        Same comparison as ``diff`` but formatted as a report with standard columns.

        Args:
            *args: 'from' and 'to' targets
            **kwargs: All diff options plus report options
                (``options``, ``sort``, ``json``, ``separator``, etc.)
        """
        options = self._kwargs_to_options(kwargs)
        positional_args = list(args)

        return self.run_command(
            subcommand=self.SUBCOMMAND,
            action='diffreport',
            options=options,
            positional_args=positional_args,
        )

    @staticmethod
    def _kwargs_to_options(kwargs: dict[str, str | None]) -> SnapmOptions:
        """Convert kwargs to CLI options (underscores to dashes, ``--`` prefix).

        ``None`` values become flags without arguments.
        """
        options: SnapmOptions = {}
        for key, value in kwargs.items():
            # Convert underscore to dash and add -- prefix
            cli_key = f'--{key.replace("_", "-")}'
            options[cli_key] = value
        return options

activate()

Activate a snapset.

Source code in sts_libs/src/sts/snapm/snapset.py
272
273
274
def activate(self) -> CommandResult:
    """Activate a snapset."""
    return self._common_operation('activate')

autoactivate(*, enable=True)

Enable or disable auto-activation for a snapset.

Source code in sts_libs/src/sts/snapm/snapset.py
280
281
282
283
284
285
286
287
288
289
290
def autoactivate(self, *, enable: bool = True) -> CommandResult:
    """Enable or disable auto-activation for a snapset."""
    options: SnapmOptions = {}

    # Add yes/no option
    if enable:
        options['--yes'] = None
    else:
        options['--no'] = None

    return self._common_operation('autoactivate', options)

create(snapset_name=None, sources=None, size_policy=None, *, bootable=False, revert=False, autoindex=False)

Create a snapset.

Parameters:

Name Type Description Default
snapset_name str | None

Snapset name (uses self.name if not provided)

None
sources list[str] | None

Source paths, optionally with size policy (e.g. '/mnt/data:2G')

None
size_policy str | None

Default size policy for all sources (e.g. "50%FREE", "2G")

None
bootable bool

Whether to create a boot entry for this snapset

False
revert bool

Whether to create a revert entry for this snapset

False
autoindex bool

Treat name as basename and append a unique index

False

Raises:

Type Description
SnapmError

If the snapset name or sources are missing.

Source code in sts_libs/src/sts/snapm/snapset.py
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def create(
    self,
    snapset_name: str | None = None,
    sources: list[str] | None = None,
    size_policy: str | None = None,
    *,
    bootable: bool = False,
    revert: bool = False,
    autoindex: bool = False,
) -> CommandResult:
    """Create a snapset.

    Args:
        snapset_name: Snapset name (uses self.name if not provided)
        sources: Source paths, optionally with size policy (e.g. ``'/mnt/data:2G'``)
        size_policy: Default size policy for all sources (e.g. "50%FREE", "2G")
        bootable: Whether to create a boot entry for this snapset
        revert: Whether to create a revert entry for this snapset
        autoindex: Treat name as basename and append a unique index

    Raises:
        SnapmError: If the snapset name or sources are missing.
    """
    # Use instance name if not provided
    name = snapset_name or self.name
    if not name:
        raise SnapmError('Snapset name is required')

    # Ensure sources is a list
    if sources is None:
        raise SnapmError('Snapset sources are required')

    options: SnapmOptions = {}
    if bootable:
        options['--bootable'] = None
    if revert:
        options['--revert'] = None
    if size_policy:
        options['--size-policy'] = size_policy
    if autoindex:
        options['--autoindex'] = None

    # Add sources to positional args
    positional_args = [name]
    positional_args.extend(sources)

    result = self.run_command(
        subcommand=self.SUBCOMMAND, action='create', options=options, positional_args=positional_args
    ).assert_ok()

    if autoindex:
        self.name = self._resolve_autoindex_name(name)
    else:
        self.name = name
    self.refresh_report()

    return result

create_scheduled(schedule_name)

Create snapset from a schedule configuration.

Does not call .assert_ok() internally -- callers should check result.failed or call .assert_ok() themselves.

Raises:

Type Description
SnapmError

If schedule_name is missing.

Source code in sts_libs/src/sts/snapm/snapset.py
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def create_scheduled(self, schedule_name: str) -> CommandResult:
    """Create snapset from a schedule configuration.

    Does not call ``.assert_ok()`` internally -- callers should check
    ``result.failed`` or call ``.assert_ok()`` themselves.

    Raises:
        SnapmError: If schedule_name is missing.
    """
    if not schedule_name:
        raise SnapmError('Schedule name is required')

    options: SnapmOptions = {'--config': schedule_name}

    return self.run_command(subcommand=self.SUBCOMMAND, action='create-scheduled', options=options)

deactivate()

Deactivate a snapset.

Source code in sts_libs/src/sts/snapm/snapset.py
276
277
278
def deactivate(self) -> CommandResult:
    """Deactivate a snapset."""
    return self._common_operation('deactivate')

delete()

Delete a snapset and all associated snapshots.

Source code in sts_libs/src/sts/snapm/snapset.py
226
227
228
def delete(self) -> CommandResult:
    """Delete a snapset and all associated snapshots."""
    return self._common_operation('delete')

diff(*args, **kwargs)

Compare two snapshot sets or a snapshot set and the running system.

Parameters:

Name Type Description Default
*args str

'from' and 'to' targets (snapset name or '.' for running system)

()
**kwargs str | None

CLI options (underscores become dashes). Key options: output_format (paths/full/short/json/diff/summary/tree), ignore_timestamps, ignore_permissions, content_only, include_pattern, exclude_pattern, start_path

{}
Example
result = snapset.diff('before-upgrade', '.', output_format='tree')
Source code in sts_libs/src/sts/snapm/snapset.py
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
def diff(self, *args: str, **kwargs: str | None) -> CommandResult:
    """Compare two snapshot sets or a snapshot set and the running system.

    Args:
        *args: 'from' and 'to' targets (snapset name or '.' for running system)
        **kwargs: CLI options (underscores become dashes). Key options:
            ``output_format`` (paths/full/short/json/diff/summary/tree),
            ``ignore_timestamps``, ``ignore_permissions``, ``content_only``,
            ``include_pattern``, ``exclude_pattern``, ``start_path``

    Example:
        ```python
        result = snapset.diff('before-upgrade', '.', output_format='tree')
        ```
    """
    options = self._kwargs_to_options(kwargs)
    positional_args = list(args)

    return self.run_command(
        subcommand=self.SUBCOMMAND,
        action='diff',
        options=options,
        positional_args=positional_args,
    )

diffreport(*args, **kwargs)

Compare snapshot sets and output results in a tabular report.

Same comparison as diff but formatted as a report with standard columns.

Parameters:

Name Type Description Default
*args str

'from' and 'to' targets

()
**kwargs str | None

All diff options plus report options (options, sort, json, separator, etc.)

{}
Source code in sts_libs/src/sts/snapm/snapset.py
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
def diffreport(self, *args: str, **kwargs: str | None) -> CommandResult:
    """Compare snapshot sets and output results in a tabular report.

    Same comparison as ``diff`` but formatted as a report with standard columns.

    Args:
        *args: 'from' and 'to' targets
        **kwargs: All diff options plus report options
            (``options``, ``sort``, ``json``, ``separator``, etc.)
    """
    options = self._kwargs_to_options(kwargs)
    positional_args = list(args)

    return self.run_command(
        subcommand=self.SUBCOMMAND,
        action='diffreport',
        options=options,
        positional_args=positional_args,
    )

exec_cmd(*args, **kwargs)

Execute a command inside the mounted snapshot set.

Parameters:

Name Type Description Default
*args str

Snapset name (if not set on instance) followed by command and its args.

()
Source code in sts_libs/src/sts/snapm/snapset.py
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
def exec_cmd(self, *args: str, **kwargs: str | None) -> CommandResult:
    """Execute a command inside the mounted snapshot set.

    Args:
        *args: Snapset name (if not set on instance) followed by command and its args.
    """
    options = self._kwargs_to_options(kwargs)
    positional_args = list(args)

    if self.name and positional_args:
        positional_args = [self.name, *positional_args]
    elif self.name and not positional_args:
        positional_args = [self.name]

    return self.run_command(
        subcommand=self.SUBCOMMAND,
        action='exec',
        options=options,
        positional_args=positional_args,
    )

get_all() classmethod

Get all snapsets with their reports populated.

Source code in sts_libs/src/sts/snapm/snapset.py
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
@classmethod
def get_all(cls) -> list[Snapset]:
    """Get all snapsets with their reports populated."""
    snapsets: list[Snapset] = []

    # Create base instance to run command
    base = cls()

    # Get snapset with members in JSON format for easier parsing
    options: SnapmOptions = {'--json': None, '--members': None}
    result = base.run_command(subcommand=cls.SUBCOMMAND, action='show', options=options)

    if result.failed or not result.stdout:
        return snapsets

    try:
        # Parse JSON output
        data: dict[str, Any] | list[Any] = json.loads(result.stdout)

        # Handle different possible JSON structures
        items: list[dict[str, Any]]
        if isinstance(data, list):
            items = data
        elif 'snapsets' in data:
            items = data['snapsets']
        else:
            items = [data]

        # Create snapset instances from each entry
        for item in items:
            snapset = cls()

            # Set basic attributes
            if 'SnapsetName' in item:
                snapset.name = item['SnapsetName']
            if 'UUID' in item:
                snapset.uuid = item['UUID']

            # Create report object
            snapset.report = SnapsetInfo.model_validate(item)

            snapsets.append(snapset)

    except (json.JSONDecodeError, TypeError) as e:
        logger.warning(f'Failed to parse snapsets: {e}')

    return snapsets

get_snapshots()

Get all snapshots in this snapset, refreshing the report first.

Source code in sts_libs/src/sts/snapm/snapset.py
423
424
425
426
427
428
429
430
431
432
433
def get_snapshots(self) -> list[SnapshotInfo]:
    """Get all snapshots in this snapset, refreshing the report first."""
    # Refresh report to ensure we have the latest data including snapshots
    if not self.refresh_report():
        return []

    # Return snapshots from report if available
    if self.report and self.report.snapshots:
        return self.report.snapshots

    return []

list_items(*, fields=None, json_output=False)

List snapsets, optionally filtered by stored name/UUID.

Parameters:

Name Type Description Default
fields str | None

Comma-separated list of fields to display

None
json_output bool

Whether to output in JSON format

False
Source code in sts_libs/src/sts/snapm/snapset.py
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def list_items(self, *, fields: str | None = None, json_output: bool = False) -> CommandResult:
    """List snapsets, optionally filtered by stored name/UUID.

    Args:
        fields: Comma-separated list of fields to display
        json_output: Whether to output in JSON format
    """
    options: SnapmOptions = {}

    # Add name/uuid if set in this instance
    if self.name:
        options['--name'] = self.name
    elif self.uuid:
        options['--uuid'] = self.uuid

    # Add fields if provided
    if fields:
        options['--options'] = fields

    # Add JSON output if requested
    if json_output:
        options['--json'] = None

    return self.run_command(subcommand=self.SUBCOMMAND, action='list', options=options)

mount(*args, **kwargs)

Mount the members of a snapshot set at /run/snapm/mounts/<name>.

Source code in sts_libs/src/sts/snapm/snapset.py
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def mount(self, *args: str, **kwargs: str | None) -> CommandResult:
    """Mount the members of a snapshot set at ``/run/snapm/mounts/<name>``."""
    options = self._kwargs_to_options(kwargs)
    positional_args = list(args)

    # Use instance name if no positional args provided
    if not positional_args and self.name:
        positional_args.append(self.name)

    return self.run_command(
        subcommand=self.SUBCOMMAND,
        action='mount',
        options=options,
        positional_args=positional_args,
    )

prune(sources=None)

Remove sources (and their snapshots) from the snapset. Cannot be undone.

Source code in sts_libs/src/sts/snapm/snapset.py
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
def prune(self, sources: list[str] | None = None) -> CommandResult:
    """Remove sources (and their snapshots) from the snapset. Cannot be undone."""
    positional_args: list[str] = []

    if self.name:
        positional_args.append(self.name)
    if sources:
        positional_args.extend(sources)

    result = self.run_command(
        subcommand=self.SUBCOMMAND, action='prune', positional_args=positional_args
    ).assert_ok()
    self.refresh_report()

    return result

refresh_report()

Refresh snapset information from show command.

Source code in sts_libs/src/sts/snapm/snapset.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def refresh_report(self) -> bool:
    """Refresh snapset information from show command."""
    result = self.show()

    if result.failed:
        logger.error(f'Failed to get snapset info: {result.stderr}')
        return False

    if not result.stdout:
        return False

    try:
        data_raw: list[dict[str, Any]] = json.loads(result.stdout)

        data: dict[str, Any]
        if not data_raw:
            logger.error(f'Failed to get snapshot info: {result.stderr}')
            return False
        data = data_raw[0]

        # Create or update report
        self.report = SnapsetInfo.model_validate(data)

    except json.JSONDecodeError as e:
        logger.warning(f'Failed to parse snapset info (invalid JSON): {e}')
        return False
    except (KeyError, TypeError) as e:
        logger.warning(f'Failed to process snapset info (unexpected format): {e}')
        return False

    return True

rename(new_name)

Rename a snapset.

Raises:

Type Description
SnapmError

If the current snapset name is missing.

Source code in sts_libs/src/sts/snapm/snapset.py
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def rename(self, new_name: str) -> CommandResult:
    """Rename a snapset.

    Raises:
        SnapmError: If the current snapset name is missing.
    """
    if not self.name:
        raise SnapmError('Current snapset name is required')

    result = self.run_command(
        subcommand=self.SUBCOMMAND, action='rename', positional_args=[self.name, new_name]
    ).assert_ok()

    # Update instance name
    self.name = new_name
    self.refresh_report()

    return result

resize(sources=None, size_policy=None)

Resize snapset.

Parameters:

Name Type Description Default
sources list[str] | None

Sources with optional size policies (e.g. ['/path:2G'])

None
size_policy str | None

Default size policy to apply (e.g. "25%FREE")

None
Source code in sts_libs/src/sts/snapm/snapset.py
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def resize(self, sources: list[str] | None = None, size_policy: str | None = None) -> CommandResult:
    """Resize snapset.

    Args:
        sources: Sources with optional size policies (e.g. ``['/path:2G']``)
        size_policy: Default size policy to apply (e.g. "25%FREE")
    """
    options: SnapmOptions = {}
    positional_args: list[str] = []

    if size_policy:
        options['--size-policy'] = size_policy

    # Add sources if provided
    if sources:
        positional_args.extend(sources)

    return self._common_operation('resize', options=options, positional_args=positional_args)

revert()

Revert to the state captured in the snapset. May require reboot.

Source code in sts_libs/src/sts/snapm/snapset.py
249
250
251
def revert(self) -> CommandResult:
    """Revert to the state captured in the snapset. May require reboot."""
    return self._common_operation('revert')

shell(*args, **kwargs)

Start an interactive shell in the mounted snapshot set.

Source code in sts_libs/src/sts/snapm/snapset.py
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
def shell(self, *args: str, **kwargs: str | None) -> CommandResult:
    """Start an interactive shell in the mounted snapshot set."""
    options = self._kwargs_to_options(kwargs)
    positional_args = list(args)

    # Use instance name if no positional args provided
    if not positional_args and self.name:
        positional_args.append(self.name)

    return self.run_command(
        subcommand=self.SUBCOMMAND,
        action='shell',
        options=options,
        positional_args=positional_args,
    )

show(*, include_members=True, json_output=True)

Show detailed snapset information.

Source code in sts_libs/src/sts/snapm/snapset.py
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def show(self, *, include_members: bool = True, json_output: bool = True) -> CommandResult:
    """Show detailed snapset information."""
    options: SnapmOptions = {}

    # Use stored attributes
    if self.name:
        options['--name'] = self.name
    elif self.uuid:
        options['--uuid'] = self.uuid
    else:
        logger.error('No snapset identifier available')
        return self.run_command(subcommand=self.SUBCOMMAND, action='show')

    # Add member and json options
    if include_members:
        options['--members'] = None

    if json_output:
        options['--json'] = None

    return self.run_command(subcommand=self.SUBCOMMAND, action='show', options=options)

split(new_name=None, sources=None)

Move specified sources into a new snapset, keeping the rest here.

Source code in sts_libs/src/sts/snapm/snapset.py
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
def split(self, new_name: str | None = None, sources: list[str] | None = None) -> Snapset:
    """Move specified sources into a new snapset, keeping the rest here."""
    positional_args: list[str] = [self.name or '']

    # Add sources if provided
    if new_name:
        positional_args.append(new_name)
    if sources:
        positional_args.extend(sources)

    result = self.run_command(subcommand=self.SUBCOMMAND, action='split', positional_args=positional_args)

    if result.failed:
        logger.error('Failed to split snapset')
        return Snapset()
    self.refresh_report()
    new_snapset = Snapset(name=new_name)
    new_snapset.refresh_report()
    return new_snapset

umount(*args, **kwargs)

Unmount the members of a snapshot set.

Source code in sts_libs/src/sts/snapm/snapset.py
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
def umount(self, *args: str, **kwargs: str | None) -> CommandResult:
    """Unmount the members of a snapshot set."""
    options = self._kwargs_to_options(kwargs)
    positional_args = list(args)

    # Use instance name if no positional args provided
    if not positional_args and self.name:
        positional_args.append(self.name)

    return self.run_command(
        subcommand=self.SUBCOMMAND,
        action='umount',
        options=options,
        positional_args=positional_args,
    )

SnapsetInfo pydantic-model

Bases: ReportModel

Snapset information parsed from snapm snapset show output.

Show JSON schema:
{
  "$defs": {
    "SnapshotInfo": {
      "description": "Snapshot information parsed from ``snapm snapshot show`` output.",
      "properties": {
        "Name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "SnapsetName": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Snapsetname"
        },
        "Origin": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Origin"
        },
        "Timestamp": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Timestamp"
        },
        "Time": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Time"
        },
        "Source": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Source"
        },
        "MountPoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "Provider": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Provider"
        },
        "UUID": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uuid"
        },
        "Status": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Status"
        },
        "Size": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "Free": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Free"
        },
        "SizeBytes": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Sizebytes"
        },
        "FreeBytes": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Freebytes"
        },
        "Autoactivate": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Autoactivate"
        },
        "DevicePath": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Devicepath"
        }
      },
      "title": "SnapshotInfo",
      "type": "object"
    }
  },
  "description": "Snapset information parsed from ``snapm snapset show`` output.",
  "properties": {
    "SnapsetName": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Snapsetname"
    },
    "Sources": {
      "items": {
        "type": "string"
      },
      "title": "Sources",
      "type": "array"
    },
    "MountPoints": {
      "items": {
        "type": "string"
      },
      "title": "Mountpoints",
      "type": "array"
    },
    "Devices": {
      "items": {
        "type": "string"
      },
      "title": "Devices",
      "type": "array"
    },
    "NrSnapshots": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Nrsnapshots"
    },
    "Timestamp": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Timestamp"
    },
    "Time": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Time"
    },
    "UUID": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "Status": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Status"
    },
    "Autoactivate": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Autoactivate"
    },
    "Bootable": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Bootable"
    },
    "BootEntries": {
      "additionalProperties": {
        "type": "string"
      },
      "title": "Bootentries",
      "type": "object"
    },
    "Snapshots": {
      "items": {
        "$ref": "#/$defs/SnapshotInfo"
      },
      "title": "Snapshots",
      "type": "array"
    }
  },
  "title": "SnapsetInfo",
  "type": "object"
}

Config:

  • populate_by_name: True

Fields:

  • name (str | None)
  • sources (list[str])
  • mount_points (list[str])
  • devices (list[str])
  • snapshot_count (int | None)
  • timestamp (int | None)
  • time (str | None)
  • uuid (str | None)
  • status (str | None)
  • autoactivate (bool | None)
  • bootable (bool | None)
  • boot_entries (dict[str, str])
  • snapshots (list[SnapshotInfo])
Source code in sts_libs/src/sts/snapm/snapset.py
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class SnapsetInfo(ReportModel):
    """Snapset information parsed from ``snapm snapset show`` output."""

    model_config = ConfigDict(populate_by_name=True)

    name: str | None = Field(default=None, alias='SnapsetName')
    sources: list[str] = Field(default_factory=list, alias='Sources')
    mount_points: list[str] = Field(default_factory=list, alias='MountPoints')
    devices: list[str] = Field(default_factory=list, alias='Devices')
    snapshot_count: int | None = Field(default=None, alias='NrSnapshots')
    timestamp: int | None = Field(default=None, alias='Timestamp')
    time: str | None = Field(default=None, alias='Time')
    uuid: str | None = Field(default=None, alias='UUID')
    status: str | None = Field(default=None, alias='Status')
    autoactivate: bool | None = Field(default=None, alias='Autoactivate')
    bootable: bool | None = Field(default=None, alias='Bootable')
    boot_entries: dict[str, str] = Field(default_factory=dict, alias='BootEntries')
    snapshots: list[SnapshotInfo] = Field(default_factory=list, alias='Snapshots')

sts.snapm.snapshot

Individual snapshot management.

Snapshot pydantic-model

Bases: SnapmBase

Snapshot management.

A Snapshot is a point-in-time copy of a filesystem.

Show JSON schema:
{
  "$defs": {
    "SnapshotInfo": {
      "description": "Snapshot information parsed from ``snapm snapshot show`` output.",
      "properties": {
        "Name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "SnapsetName": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Snapsetname"
        },
        "Origin": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Origin"
        },
        "Timestamp": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Timestamp"
        },
        "Time": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Time"
        },
        "Source": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Source"
        },
        "MountPoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "Provider": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Provider"
        },
        "UUID": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uuid"
        },
        "Status": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Status"
        },
        "Size": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "Free": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Free"
        },
        "SizeBytes": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Sizebytes"
        },
        "FreeBytes": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Freebytes"
        },
        "Autoactivate": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Autoactivate"
        },
        "DevicePath": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Devicepath"
        }
      },
      "title": "SnapshotInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Snapshot management.\n\nA Snapshot is a point-in-time copy of a filesystem.",
  "properties": {
    "debugopts": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Debugopts"
    },
    "verbose": {
      "default": false,
      "title": "Verbose",
      "type": "boolean"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "snapset_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Snapset Name"
    },
    "snapset_uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Snapset Uuid"
    },
    "report": {
      "anyOf": [
        {
          "$ref": "#/$defs/SnapshotInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "title": "Snapshot",
  "type": "object"
}

Fields:

  • debugopts (list[str] | None)
  • verbose (bool)
  • name (str | None)
  • uuid (str | None)
  • snapset_name (str | None)
  • snapset_uuid (str | None)
  • report (SnapshotInfo | None)
Source code in sts_libs/src/sts/snapm/snapshot.py
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
class Snapshot(SnapmBase):
    """Snapshot management.

    A Snapshot is a point-in-time copy of a filesystem.
    """

    # Class-level constants
    SUBCOMMAND: ClassVar[str] = 'snapshot'

    # Instance attributes
    name: str | None = None
    uuid: str | None = None
    snapset_name: str | None = None
    snapset_uuid: str | None = None
    report: SnapshotInfo | None = Field(default=None, repr=False)

    def _build_identifier_options(self) -> SnapmOptions:
        """Build options dictionary with available identifiers."""
        options: SnapmOptions = {}

        if self.name:
            options['--snapshot-name'] = self.name
        elif self.uuid:
            options['--snapshot-uuid'] = self.uuid
        elif self.snapset_name:
            options['--name'] = self.snapset_name
        elif self.snapset_uuid:
            options['--uuid'] = self.snapset_uuid
        else:
            logger.warning('No snapshot identifier (ID, snapset name, or snapset UUID) available')

        return options

    def _common_operation(self, action: str, options: SnapmOptions | None = None) -> CommandResult:
        """Run a snapshot action using stored identifiers, refreshing report on success.

        Raises:
            SnapmError: If no snapshot identifier is available.
        """
        if not options:
            options = {}
        positional_args: list[str] = []

        id_options = self._build_identifier_options()
        if not id_options:
            raise SnapmError('No snapshot identifier (ID, snapset name, or snapset UUID) available')

        options.update(id_options)

        result = self.run_command(
            subcommand=self.SUBCOMMAND, action=action, options=options, positional_args=positional_args
        ).assert_ok()

        # Refresh information on success
        self.refresh_report()

        return result

    def refresh_report(self) -> bool:
        """Refresh snapshot information from show command."""
        # Get report
        options: SnapmOptions = {}

        # Add JSON output option for easier parsing
        options['--json'] = None

        # Ensure an identifier is available to refresh instance specific information
        if not self._build_identifier_options():
            return False

        result = self.show()

        if result.failed:
            logger.error(f'Failed to get snapshot info: {result.stderr}')
            return False

        if not result.stdout:
            return False

        try:
            data_raw: list[dict[str, Any]] = json.loads(result.stdout)
            if not data_raw:
                logger.error(f'Failed to get snapshot info: {result.stderr}')
                return False
            # Getting data for specific name or uuid, so we only need first index
            data: dict[str, Any] = data_raw[0]

            # Create or update report
            self.report = SnapshotInfo.model_validate(data)

        except json.JSONDecodeError as e:
            logger.warning(f'Failed to parse snapshot info (invalid JSON): {e}')
            return False
        except (KeyError, TypeError) as e:
            logger.warning(f'Failed to process snapshot info (unexpected format): {e}')
            return False

        return True

    def activate(self) -> CommandResult:
        """Activate a snapshot."""
        return self._common_operation('activate')

    def deactivate(self) -> CommandResult:
        """Deactivate a snapshot."""
        return self._common_operation('deactivate')

    def autoactivate(self, *, enable: bool = True) -> CommandResult:
        """Enable or disable auto-activation for a snapshot."""
        options: SnapmOptions = {}

        # Add yes/no option
        if enable:
            options['--yes'] = None
        else:
            options['--no'] = None

        return self._common_operation('autoactivate', options)

    def list_items(self, fields: str | None = None, *, json_output: bool = False) -> CommandResult:
        """List snapshots, optionally filtered by stored identifiers.

        Args:
            fields: Comma-separated list of fields to display
            json_output: Whether to output in JSON format
        """
        options: SnapmOptions = {}

        # No need to fail here. If ID is not available, empty dict is returned
        id_options = self._build_identifier_options()
        options.update(id_options)

        # Add fields if provided
        if fields:
            options['--options'] = fields

        # Add JSON output if requested
        if json_output:
            options['--json'] = None

        return self.run_command(subcommand=self.SUBCOMMAND, action='list', options=options)

    def show(self, *, json_output: bool = True) -> CommandResult:
        """Show detailed snapshot information."""
        options: SnapmOptions = {}

        # No need to fail here. If ID is not available, empty dict is returned
        id_options = self._build_identifier_options()
        options.update(id_options)

        # Add json options

        if json_output:
            options['--json'] = None

        return self.run_command(subcommand=self.SUBCOMMAND, action='show', options=options)

activate()

Activate a snapshot.

Source code in sts_libs/src/sts/snapm/snapshot.py
146
147
148
def activate(self) -> CommandResult:
    """Activate a snapshot."""
    return self._common_operation('activate')

autoactivate(*, enable=True)

Enable or disable auto-activation for a snapshot.

Source code in sts_libs/src/sts/snapm/snapshot.py
154
155
156
157
158
159
160
161
162
163
164
def autoactivate(self, *, enable: bool = True) -> CommandResult:
    """Enable or disable auto-activation for a snapshot."""
    options: SnapmOptions = {}

    # Add yes/no option
    if enable:
        options['--yes'] = None
    else:
        options['--no'] = None

    return self._common_operation('autoactivate', options)

deactivate()

Deactivate a snapshot.

Source code in sts_libs/src/sts/snapm/snapshot.py
150
151
152
def deactivate(self) -> CommandResult:
    """Deactivate a snapshot."""
    return self._common_operation('deactivate')

list_items(fields=None, *, json_output=False)

List snapshots, optionally filtered by stored identifiers.

Parameters:

Name Type Description Default
fields str | None

Comma-separated list of fields to display

None
json_output bool

Whether to output in JSON format

False
Source code in sts_libs/src/sts/snapm/snapshot.py
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def list_items(self, fields: str | None = None, *, json_output: bool = False) -> CommandResult:
    """List snapshots, optionally filtered by stored identifiers.

    Args:
        fields: Comma-separated list of fields to display
        json_output: Whether to output in JSON format
    """
    options: SnapmOptions = {}

    # No need to fail here. If ID is not available, empty dict is returned
    id_options = self._build_identifier_options()
    options.update(id_options)

    # Add fields if provided
    if fields:
        options['--options'] = fields

    # Add JSON output if requested
    if json_output:
        options['--json'] = None

    return self.run_command(subcommand=self.SUBCOMMAND, action='list', options=options)

refresh_report()

Refresh snapshot information from show command.

Source code in sts_libs/src/sts/snapm/snapshot.py
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def refresh_report(self) -> bool:
    """Refresh snapshot information from show command."""
    # Get report
    options: SnapmOptions = {}

    # Add JSON output option for easier parsing
    options['--json'] = None

    # Ensure an identifier is available to refresh instance specific information
    if not self._build_identifier_options():
        return False

    result = self.show()

    if result.failed:
        logger.error(f'Failed to get snapshot info: {result.stderr}')
        return False

    if not result.stdout:
        return False

    try:
        data_raw: list[dict[str, Any]] = json.loads(result.stdout)
        if not data_raw:
            logger.error(f'Failed to get snapshot info: {result.stderr}')
            return False
        # Getting data for specific name or uuid, so we only need first index
        data: dict[str, Any] = data_raw[0]

        # Create or update report
        self.report = SnapshotInfo.model_validate(data)

    except json.JSONDecodeError as e:
        logger.warning(f'Failed to parse snapshot info (invalid JSON): {e}')
        return False
    except (KeyError, TypeError) as e:
        logger.warning(f'Failed to process snapshot info (unexpected format): {e}')
        return False

    return True

show(*, json_output=True)

Show detailed snapshot information.

Source code in sts_libs/src/sts/snapm/snapshot.py
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def show(self, *, json_output: bool = True) -> CommandResult:
    """Show detailed snapshot information."""
    options: SnapmOptions = {}

    # No need to fail here. If ID is not available, empty dict is returned
    id_options = self._build_identifier_options()
    options.update(id_options)

    # Add json options

    if json_output:
        options['--json'] = None

    return self.run_command(subcommand=self.SUBCOMMAND, action='show', options=options)

SnapshotInfo pydantic-model

Bases: ReportModel

Snapshot information parsed from snapm snapshot show output.

Show JSON schema:
{
  "description": "Snapshot information parsed from ``snapm snapshot show`` output.",
  "properties": {
    "Name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "SnapsetName": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Snapsetname"
    },
    "Origin": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Origin"
    },
    "Timestamp": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Timestamp"
    },
    "Time": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Time"
    },
    "Source": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Source"
    },
    "MountPoint": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Mountpoint"
    },
    "Provider": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Provider"
    },
    "UUID": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "Status": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Status"
    },
    "Size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "Free": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Free"
    },
    "SizeBytes": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Sizebytes"
    },
    "FreeBytes": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Freebytes"
    },
    "Autoactivate": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Autoactivate"
    },
    "DevicePath": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Devicepath"
    }
  },
  "title": "SnapshotInfo",
  "type": "object"
}

Config:

  • populate_by_name: True

Fields:

  • name (str | None)
  • snapset_name (str | None)
  • origin (str | None)
  • timestamp (int | None)
  • time (str | None)
  • source (str | None)
  • mount_point (str | None)
  • provider (str | None)
  • uuid (str | None)
  • status (str | None)
  • size (str | None)
  • free (str | None)
  • size_bytes (int | None)
  • free_bytes (int | None)
  • autoactivate (bool | None)
  • device_path (str | None)
Source code in sts_libs/src/sts/snapm/snapshot.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class SnapshotInfo(ReportModel):
    """Snapshot information parsed from ``snapm snapshot show`` output."""

    model_config = ConfigDict(populate_by_name=True)

    name: str | None = Field(default=None, alias='Name')
    snapset_name: str | None = Field(default=None, alias='SnapsetName')
    origin: str | None = Field(default=None, alias='Origin')
    timestamp: int | None = Field(default=None, alias='Timestamp')
    time: str | None = Field(default=None, alias='Time')
    source: str | None = Field(default=None, alias='Source')
    mount_point: str | None = Field(default=None, alias='MountPoint')
    provider: str | None = Field(default=None, alias='Provider')
    uuid: str | None = Field(default=None, alias='UUID')
    status: str | None = Field(default=None, alias='Status')
    size: str | None = Field(default=None, alias='Size')
    free: str | None = Field(default=None, alias='Free')
    size_bytes: int | None = Field(default=None, alias='SizeBytes')
    free_bytes: int | None = Field(default=None, alias='FreeBytes')
    autoactivate: bool | None = Field(default=None, alias='Autoactivate')
    device_path: str | None = Field(default=None, alias='DevicePath')