Skip to content

Stratis

Stratis is a local storage management solution built on device-mapper thin provisioning. It organizes block devices into pools, which host XFS filesystems with automatic thin provisioning, snapshots, and optional encryption (LUKS2 via kernel keyring or Clevis/Tang NBDE). Managed by the stratisd daemon via the stratis CLI and D-Bus.

Base Functionality

sts.stratis.base

Base Stratis CLI wrapper, configuration, and cross-cutting patterns.

Patterns used across Stratis modules:

D-Bus Maybe-type [flag, value] pattern: stratisd encodes optional D-Bus properties as two-element lists. [0, default] means the value is not set; [1, actual_value] means the value is set. This pattern appears in TotalPhysicalUsed, LastReencryptedTimestamp, Used, SizeLimit, and others across pool.py and filesystem.py.

StratisOptions None-value convention: In the StratisOptions type alias (dict[str, str | None]), a None value means the key is a boolean CLI flag with no argument (e.g., '--trust-url': None produces just --trust-url). A string value means the flag takes that argument (e.g., '--key-desc': 'mykey' produces --key-desc mykey).

Two-tier data fetching

stratis report provides basic pool/filesystem metadata (names, UUIDs, blockdevs). stratis report managed_objects_report queries the D-Bus managed-objects interface for additional data (sizes in bytes, encryption status, D-Bus variant fields). Code in pool.py and filesystem.py typically calls both reports and merges results.

last_revision pattern: The managed-objects report organizes data by D-Bus object path, then by interface revision (e.g., org.storage.stratis3.pool.r5, ...r6, ...r7). Code uses list(interfaces.keys())[-1] to select the highest (latest) revision, which carries the most complete set of properties.

Key pydantic-model

Bases: StratisBase

Stratis encryption key management (kernel keyring).

Show JSON schema: 
{
  "$defs": {
    "StratisConfig": {
      "additionalProperties": false,
      "description": "Stratis configuration controlling global CLI options.",
      "properties": {
        "unhyphenated_uuids": {
          "default": false,
          "title": "Unhyphenated Uuids",
          "type": "boolean"
        }
      },
      "title": "StratisConfig",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Stratis encryption key management (kernel keyring).",
  "properties": {
    "config": {
      "$ref": "#/$defs/StratisConfig"
    }
  },
  "title": "Key",
  "type": "object"
}

Fields:

Source code in sts_libs/src/sts/stratis/base.py 
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
class Key(StratisBase):
    """Stratis encryption key management (kernel keyring)."""

    def set(self, keydesc: str, keyfile_path: str) -> CommandResult:
        """Register a key file under the given description."""
        return self.run_command(
            'key',
            'set',
            options={'--keyfile-path': keyfile_path},
            positional_args=[keydesc],
        )

    def reset(self, keydesc: str, keyfile_path: str) -> CommandResult:
        """Reset key data for an existing key in the kernel keyring."""
        return self.run_command(
            'key',
            'reset',
            options={'--keyfile-path': keyfile_path},
            positional_args=[keydesc],
        )

    def unset(self, keydesc: str) -> CommandResult:
        """Remove key registration (does not delete the key file)."""
        return self.run_command('key', 'unset', positional_args=[keydesc])

    def list(self) -> CommandResult:
        """List registered key descriptions."""
        return self.run_command('key', 'list')

    def exists(self, keydesc: str) -> bool:
        """Check if a key with the given description is registered."""
        result = self.list()
        return bool(result.succeeded and keydesc in result.stdout)

exists(keydesc)

Check if a key with the given description is registered.

Source code in sts_libs/src/sts/stratis/base.py 
137
138
139
140
def exists(self, keydesc: str) -> bool:
    """Check if a key with the given description is registered."""
    result = self.list()
    return bool(result.succeeded and keydesc in result.stdout)

list()

List registered key descriptions.

Source code in sts_libs/src/sts/stratis/base.py 
133
134
135
def list(self) -> CommandResult:
    """List registered key descriptions."""
    return self.run_command('key', 'list')

reset(keydesc, keyfile_path)

Reset key data for an existing key in the kernel keyring.

Source code in sts_libs/src/sts/stratis/base.py 
120
121
122
123
124
125
126
127
def reset(self, keydesc: str, keyfile_path: str) -> CommandResult:
    """Reset key data for an existing key in the kernel keyring."""
    return self.run_command(
        'key',
        'reset',
        options={'--keyfile-path': keyfile_path},
        positional_args=[keydesc],
    )

set(keydesc, keyfile_path)

Register a key file under the given description.

Source code in sts_libs/src/sts/stratis/base.py 
111
112
113
114
115
116
117
118
def set(self, keydesc: str, keyfile_path: str) -> CommandResult:
    """Register a key file under the given description."""
    return self.run_command(
        'key',
        'set',
        options={'--keyfile-path': keyfile_path},
        positional_args=[keydesc],
    )

unset(keydesc)

Remove key registration (does not delete the key file).

Source code in sts_libs/src/sts/stratis/base.py 
129
130
131
def unset(self, keydesc: str) -> CommandResult:
    """Remove key registration (does not delete the key file)."""
    return self.run_command('key', 'unset', positional_args=[keydesc])

StratisBase pydantic-model

Bases: CliTool

Base class for Stratis operations.

Show JSON schema: 
{
  "$defs": {
    "StratisConfig": {
      "additionalProperties": false,
      "description": "Stratis configuration controlling global CLI options.",
      "properties": {
        "unhyphenated_uuids": {
          "default": false,
          "title": "Unhyphenated Uuids",
          "type": "boolean"
        }
      },
      "title": "StratisConfig",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Base class for Stratis operations.",
  "properties": {
    "config": {
      "$ref": "#/$defs/StratisConfig"
    }
  },
  "title": "StratisBase",
  "type": "object"
}

Fields:

Source code in sts_libs/src/sts/stratis/base.py 
 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
class StratisBase(CliTool):
    """Base class for Stratis operations."""

    CLI_NAME: ClassVar[str] = CLI_NAME

    config: StratisConfig = Field(default_factory=StratisConfig, repr=False)

    def _global_args(self) -> list[str]:
        """Return global CLI arguments derived from ``self.config``."""
        return self.config.to_args()

    @cached_property
    def version(self) -> VersionInfo:
        """Get Stratis CLI version."""
        return VersionInfo.from_string(self.run_command(action='--version').stdout.strip())

    def get_report(self) -> StratisReportData | None:
        """Get system-wide Stratis report (pools, filesystems, block devices)."""
        result = self.run_command('report')
        if result.failed or not result.stdout:
            return None

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            return None

version cached property

Get Stratis CLI version.

get_report()

Get system-wide Stratis report (pools, filesystems, block devices).

Source code in sts_libs/src/sts/stratis/base.py 
 96
 97
 98
 99
100
101
102
103
104
105
def get_report(self) -> StratisReportData | None:
    """Get system-wide Stratis report (pools, filesystems, block devices)."""
    result = self.run_command('report')
    if result.failed or not result.stdout:
        return None

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return None

StratisConfig pydantic-model

Bases: StsBaseModel

Stratis configuration controlling global CLI options.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Stratis configuration controlling global CLI options.",
  "properties": {
    "unhyphenated_uuids": {
      "default": false,
      "title": "Unhyphenated Uuids",
      "type": "boolean"
    }
  },
  "title": "StratisConfig",
  "type": "object"
}

Fields:

  • unhyphenated_uuids (bool)
Source code in sts_libs/src/sts/stratis/base.py 
67
68
69
70
71
72
73
74
75
76
77
class StratisConfig(StsBaseModel):
    """Stratis configuration controlling global CLI options."""

    unhyphenated_uuids: bool = False  # Use UUIDs without hyphens

    def to_args(self) -> list[str]:
        """Convert configuration to CLI arguments."""
        args: list[str] = []
        if self.unhyphenated_uuids:
            args.append('--unhyphenated_uuids')
        return args

to_args()

Convert configuration to CLI arguments.

Source code in sts_libs/src/sts/stratis/base.py 
72
73
74
75
76
77
def to_args(self) -> list[str]:
    """Convert configuration to CLI arguments."""
    args: list[str] = []
    if self.unhyphenated_uuids:
        args.append('--unhyphenated_uuids')
    return args

dbus_maybe_value(data, key)

Extract value from a D-Bus Maybe-type [flag, value] field.

stratisd encodes optional D-Bus properties as [0, default] (not set) or [1, actual_value] (set).

Source code in sts_libs/src/sts/stratis/base.py 
55
56
57
58
59
60
61
62
63
64
def dbus_maybe_value(data: dict[str, Any], key: str) -> Any | None:
    """Extract value from a D-Bus Maybe-type ``[flag, value]`` field.

    stratisd encodes optional D-Bus properties as ``[0, default]``
    (not set) or ``[1, actual_value]`` (set).
    """
    raw: list[Any] | None = data.get(key)
    if isinstance(raw, list) and len(raw) > 1 and raw[0] == 1:
        return raw[1]
    return None

Pool Management

sts.stratis.pool

Stratis pool management.

Two main classes:

  • PoolReport — fetches and holds pool metadata from stratisd. Uses the two-tier fetching pattern: first stratis report for filesystem/blockdev details, then the D-Bus managed-objects report for sizes and encryption state not exposed by the basic report.
  • StratisPool — high-level pool management interface. Wraps a PoolReport and exposes pool operations (create, destroy, encryption, cache, etc.).

The managed-objects interface provides D-Bus-level details (total/used sizes, encryption state, key descriptors, Clevis info) that are absent from the standard stratis report output.

See stratis.base module docstring for cross-cutting patterns (D-Bus Maybe-type, StratisOptions convention, two-tier data fetching, last_revision).

BlockDevInfo pydantic-model

Bases: ReportModel

Block device information from stratis report.

Show JSON schema: 
{
  "description": "Block device information from stratis report.",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "in_use": {
      "default": false,
      "title": "In Use",
      "type": "boolean"
    },
    "blksizes": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Blksizes"
    },
    "clevis_config": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Clevis Config"
    },
    "clevis_pin": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Clevis Pin"
    },
    "key_description": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Key Description"
    }
  },
  "title": "BlockDevInfo",
  "type": "object"
}

Fields:

  • path (str | None)
  • size (str | None)
  • uuid (str | None)
  • in_use (CoerceBool)
  • blksizes (str | None)
  • clevis_config (dict[str, Any] | None)
  • clevis_pin (str | None)
  • key_description (str | None)
Source code in sts_libs/src/sts/stratis/pool.py 
52
53
54
55
56
57
58
59
60
61
62
63
class BlockDevInfo(ReportModel):
    """Block device information from stratis report."""

    path: str | None = None
    size: str | None = None
    uuid: str | None = None
    # stratisd returns in_use as string, int, or bool depending on version — normalize here
    in_use: CoerceBool = False
    blksizes: str | None = None
    clevis_config: dict[str, Any] | None = None
    clevis_pin: str | None = None
    key_description: str | None = None

BlockDevs pydantic-model

Bases: ReportModel

Block devices in a Stratis pool.

Pools have two storage tiers: datadevs for primary data storage and cachedevs for an optional fast cache layer (typically SSDs accelerating HDDs).

Show JSON schema: 
{
  "$defs": {
    "BlockDevInfo": {
      "description": "Block device information from stratis report.",
      "properties": {
        "path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Path"
        },
        "size": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "uuid": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uuid"
        },
        "in_use": {
          "default": false,
          "title": "In Use",
          "type": "boolean"
        },
        "blksizes": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Blksizes"
        },
        "clevis_config": {
          "anyOf": [
            {
              "additionalProperties": true,
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Clevis Config"
        },
        "clevis_pin": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Clevis Pin"
        },
        "key_description": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Key Description"
        }
      },
      "title": "BlockDevInfo",
      "type": "object"
    }
  },
  "description": "Block devices in a Stratis pool.\n\nPools have two storage tiers: ``datadevs`` for primary data storage and ``cachedevs`` for an\noptional fast cache layer (typically SSDs accelerating HDDs).",
  "properties": {
    "datadevs": {
      "items": {
        "$ref": "#/$defs/BlockDevInfo"
      },
      "title": "Datadevs",
      "type": "array"
    },
    "cachedevs": {
      "items": {
        "$ref": "#/$defs/BlockDevInfo"
      },
      "title": "Cachedevs",
      "type": "array"
    }
  },
  "title": "BlockDevs",
  "type": "object"
}

Fields:

Source code in sts_libs/src/sts/stratis/pool.py 
101
102
103
104
105
106
107
108
109
class BlockDevs(ReportModel):
    """Block devices in a Stratis pool.

    Pools have two storage tiers: ``datadevs`` for primary data storage and ``cachedevs`` for an
    optional fast cache layer (typically SSDs accelerating HDDs).
    """

    datadevs: list[BlockDevInfo] = Field(default_factory=list)
    cachedevs: list[BlockDevInfo] = Field(default_factory=list)

EncryptionInfo pydantic-model

Bases: StsBaseModel

Encryption metadata for a pool (key descriptions and Clevis bindings).

Uses StsBaseModel rather than ReportModel: PoolReport and StratisPool are mutable and reassign self.encryption wholesale (see _parse_pool_interface), which a frozen ReportModel would reject.

v1 pools have at most one binding (singular KeyDescription/ClevisInfo in the D-Bus interface); v2 pools support multiple (plural KeyDescriptions/ClevisInfos). Both are normalized to the plural field names here — see PoolReport._parse_pool_interface.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Encryption metadata for a pool (key descriptions and Clevis bindings).\n\nUses `StsBaseModel` rather than `ReportModel`: `PoolReport` and `StratisPool`\nare mutable and reassign `self.encryption` wholesale (see\n`_parse_pool_interface`), which a frozen `ReportModel` would reject.\n\nv1 pools have at most one binding (singular ``KeyDescription``/``ClevisInfo``\nin the D-Bus interface); v2 pools support multiple (plural\n``KeyDescriptions``/``ClevisInfos``). Both are normalized to the plural\nfield names here \u2014 see `PoolReport._parse_pool_interface`.",
  "properties": {
    "key_descriptions": {
      "default": null,
      "title": "Key Descriptions"
    },
    "clevis_infos": {
      "default": null,
      "title": "Clevis Infos"
    }
  },
  "title": "EncryptionInfo",
  "type": "object"
}

Fields:

  • key_descriptions (Any)
  • clevis_infos (Any)
Source code in sts_libs/src/sts/stratis/pool.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class EncryptionInfo(StsBaseModel):
    """Encryption metadata for a pool (key descriptions and Clevis bindings).

    Uses `StsBaseModel` rather than `ReportModel`: `PoolReport` and `StratisPool`
    are mutable and reassign `self.encryption` wholesale (see
    `_parse_pool_interface`), which a frozen `ReportModel` would reject.

    v1 pools have at most one binding (singular ``KeyDescription``/``ClevisInfo``
    in the D-Bus interface); v2 pools support multiple (plural
    ``KeyDescriptions``/``ClevisInfos``). Both are normalized to the plural
    field names here — see `PoolReport._parse_pool_interface`.
    """

    key_descriptions: Any = None
    clevis_infos: Any = None

    def __bool__(self) -> bool:
        """True if any encryption binding is present (dict-like truthiness).

        Callers use ``if pool.encryption:`` to mean "is this pool encrypted",
        matching the raw-dict behavior this model replaced.
        """
        return bool(self.key_descriptions or self.clevis_infos)

__bool__()

True if any encryption binding is present (dict-like truthiness).

Callers use if pool.encryption: to mean "is this pool encrypted", matching the raw-dict behavior this model replaced.

Source code in sts_libs/src/sts/stratis/pool.py 
82
83
84
85
86
87
88
def __bool__(self) -> bool:
    """True if any encryption binding is present (dict-like truthiness).

    Callers use ``if pool.encryption:`` to mean "is this pool encrypted",
    matching the raw-dict behavior this model replaced.
    """
    return bool(self.key_descriptions or self.clevis_infos)

PoolCreateConfig pydantic-model

Bases: StsBaseModel

Pool creation configuration (encryption and integrity options).

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Pool creation configuration (encryption and integrity options).",
  "properties": {
    "key_desc": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Key Desc"
    },
    "tang_url": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Tang Url"
    },
    "thumbprint": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Thumbprint"
    },
    "clevis": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Clevis"
    },
    "trust_url": {
      "default": false,
      "title": "Trust Url",
      "type": "boolean"
    },
    "no_overprovision": {
      "default": false,
      "title": "No Overprovision",
      "type": "boolean"
    },
    "integrity": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Integrity"
    },
    "journal_size": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Journal Size"
    },
    "tag_spec": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Tag Spec"
    }
  },
  "title": "PoolCreateConfig",
  "type": "object"
}

Fields:

  • key_desc (str | None)
  • tang_url (str | None)
  • thumbprint (str | None)
  • clevis (str | None)
  • trust_url (bool)
  • no_overprovision (bool)
  • integrity (str | None)
  • journal_size (int | None)
  • tag_spec (str | None)
Source code in sts_libs/src/sts/stratis/pool.py 
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
class PoolCreateConfig(StsBaseModel):
    """Pool creation configuration (encryption and integrity options)."""

    # Encryption options
    key_desc: str | None = None
    tang_url: str | None = None
    thumbprint: str | None = None
    clevis: str | None = None
    trust_url: bool = False
    no_overprovision: bool = False

    # Integrity options
    integrity: str | None = None
    journal_size: int | None = None
    tag_spec: str | None = None

PoolReport pydantic-model

Bases: StratisBase

Pool report data.

Mutable model that fetches and holds pool metadata from stratisd. Call refresh() after construction to populate from the system.

Show JSON schema: 
{
  "$defs": {
    "BlockDevInfo": {
      "description": "Block device information from stratis report.",
      "properties": {
        "path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Path"
        },
        "size": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "uuid": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uuid"
        },
        "in_use": {
          "default": false,
          "title": "In Use",
          "type": "boolean"
        },
        "blksizes": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Blksizes"
        },
        "clevis_config": {
          "anyOf": [
            {
              "additionalProperties": true,
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Clevis Config"
        },
        "clevis_pin": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Clevis Pin"
        },
        "key_description": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Key Description"
        }
      },
      "title": "BlockDevInfo",
      "type": "object"
    },
    "BlockDevs": {
      "description": "Block devices in a Stratis pool.\n\nPools have two storage tiers: ``datadevs`` for primary data storage and ``cachedevs`` for an\noptional fast cache layer (typically SSDs accelerating HDDs).",
      "properties": {
        "datadevs": {
          "items": {
            "$ref": "#/$defs/BlockDevInfo"
          },
          "title": "Datadevs",
          "type": "array"
        },
        "cachedevs": {
          "items": {
            "$ref": "#/$defs/BlockDevInfo"
          },
          "title": "Cachedevs",
          "type": "array"
        }
      },
      "title": "BlockDevs",
      "type": "object"
    },
    "EncryptionInfo": {
      "additionalProperties": false,
      "description": "Encryption metadata for a pool (key descriptions and Clevis bindings).\n\nUses `StsBaseModel` rather than `ReportModel`: `PoolReport` and `StratisPool`\nare mutable and reassign `self.encryption` wholesale (see\n`_parse_pool_interface`), which a frozen `ReportModel` would reject.\n\nv1 pools have at most one binding (singular ``KeyDescription``/``ClevisInfo``\nin the D-Bus interface); v2 pools support multiple (plural\n``KeyDescriptions``/``ClevisInfos``). Both are normalized to the plural\nfield names here \u2014 see `PoolReport._parse_pool_interface`.",
      "properties": {
        "key_descriptions": {
          "default": null,
          "title": "Key Descriptions"
        },
        "clevis_infos": {
          "default": null,
          "title": "Clevis Infos"
        }
      },
      "title": "EncryptionInfo",
      "type": "object"
    },
    "StratisConfig": {
      "additionalProperties": false,
      "description": "Stratis configuration controlling global CLI options.",
      "properties": {
        "unhyphenated_uuids": {
          "default": false,
          "title": "Unhyphenated Uuids",
          "type": "boolean"
        }
      },
      "title": "StratisConfig",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Pool report data.\n\nMutable model that fetches and holds pool metadata from stratisd.\nCall ``refresh()`` after construction to populate from the system.",
  "properties": {
    "config": {
      "$ref": "#/$defs/StratisConfig"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "blockdevs": {
      "$ref": "#/$defs/BlockDevs"
    },
    "uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "encryption": {
      "$ref": "#/$defs/EncryptionInfo"
    },
    "encrypted": {
      "default": false,
      "title": "Encrypted",
      "type": "boolean"
    },
    "last_reencrypted_timestamp": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Last Reencrypted Timestamp"
    },
    "fs_limit": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Fs Limit"
    },
    "available_actions": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Available Actions"
    },
    "filesystems": {
      "items": {
        "type": "string"
      },
      "title": "Filesystems",
      "type": "array"
    },
    "raw_data": {
      "additionalProperties": true,
      "title": "Raw Data",
      "type": "object"
    },
    "total_size": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Total Size"
    },
    "used_size": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Used Size"
    },
    "object_path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Object Path"
    },
    "prevent_update": {
      "default": false,
      "title": "Prevent Update",
      "type": "boolean"
    }
  },
  "title": "PoolReport",
  "type": "object"
}

Fields:

  • config (StratisConfig)
  • name (str | None)
  • blockdevs (BlockDevs)
  • uuid (str | None)
  • encryption (EncryptionInfo)
  • encrypted (bool)
  • last_reencrypted_timestamp (str | None)
  • fs_limit (int | None)
  • available_actions (str | None)
  • filesystems (list[str])
  • raw_data (dict[str, Any])
  • total_size (int | None)
  • used_size (int | None)
  • object_path (str | None)
  • prevent_update (bool)
Source code in sts_libs/src/sts/stratis/pool.py 
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
class PoolReport(StratisBase):
    """Pool report data.

    Mutable model that fetches and holds pool metadata from stratisd.
    Call ``refresh()`` after construction to populate from the system.
    """

    name: str | None = None
    blockdevs: BlockDevs = Field(default_factory=BlockDevs)
    uuid: str | None = None
    encryption: EncryptionInfo = Field(default_factory=EncryptionInfo)
    encrypted: bool = False
    last_reencrypted_timestamp: str | None = None
    fs_limit: int | None = None
    available_actions: str | None = None
    filesystems: list[str] = Field(default_factory=list)
    raw_data: dict[str, Any] = Field(default_factory=dict, repr=False)
    total_size: int | None = None
    used_size: int | None = None
    object_path: str | None = None
    # When True, refresh/update methods become no-ops — used to avoid re-fetching already-populated data
    prevent_update: bool = False

    def refresh(self) -> bool:
        """Refresh pool report data from stratisd."""
        # If prevent_update is True, skip refresh
        if self.prevent_update:
            logger.debug('Refresh skipped due to prevent_update flag')
            return True

        # First get standard report data
        result = self.run_command('report')
        if result.failed or not result.stdout:
            logger.error('Failed to get report data')
            return False

        try:
            report_data = json.loads(result.stdout)
            if not self._update_from_report(report_data):
                return False

            # If we have a name, also fetch size information from managed objects
            if self.name:
                self.update_from_managed_objects()

        except json.JSONDecodeError:
            logger.exception('Failed to parse report JSON')
            return False
        else:
            return True

    def update_from_managed_objects(self) -> bool:
        """Fetch encryption info and total/used size from the managed objects interface."""
        if self.prevent_update:
            logger.debug('Size info update skipped due to prevent_update flag')
            return True

        if not self.name:
            return False

        # Get object path
        result = self.run_command(
            subcommand='pool', action=['debug', 'get-object-path'], positional_args=['--name', self.name]
        )

        if result.failed or not result.stdout:
            logger.debug(f'Failed to get object path for pool {self.name}')
            return False

        self.object_path = result.stdout.strip()

        # Get managed objects report
        result = self.run_command(subcommand='report', action='managed_objects_report')

        if result.failed or not result.stdout:
            logger.error('Failed to get managed objects report')
            return False

        try:
            report_data = json.loads(result.stdout)

            if self.object_path not in report_data:
                logger.error(f'Object path {self.object_path} not found in managed objects report')
                return False

            pool_interfaces = report_data[self.object_path]
            last_revision = list(pool_interfaces.keys())[-1]
            self._parse_pool_interface(pool_interfaces[last_revision])

        except (json.JSONDecodeError, KeyError, ValueError):
            logger.exception('Error parsing managed objects data')
            return False
        else:
            return True

    def _parse_pool_interface(self, pool_interface: dict[str, Any]) -> None:
        """Parse pool attributes from a D-Bus managed-objects interface revision dict.

        Handles the D-Bus ``[flag, value]`` Maybe-type pattern for optional fields and detects
        v1 vs v2 metadata (v2 uses plural ``KeyDescriptions``/``ClevisInfos`` for multi-binding
        support).
        """
        self.encrypted = pool_interface.get('Encrypted', 0) == 1

        ts = dbus_maybe_value(pool_interface, 'LastReencryptedTimestamp')
        self.last_reencrypted_timestamp = str(ts) if ts is not None else None

        # Encryption info — v2 uses plural keys, v1 uses singular
        if self.encrypted:
            is_v2 = pool_interface.get('MetadataVersion') == 2
            key_field = 'KeyDescriptions' if is_v2 else 'KeyDescription'
            clevis_field = 'ClevisInfos' if is_v2 else 'ClevisInfo'

            self.encryption = EncryptionInfo(
                key_descriptions=pool_interface.get(key_field) or self.encryption.key_descriptions,
                clevis_infos=pool_interface.get(clevis_field) or self.encryption.clevis_infos,
            )
        else:
            self.encryption = EncryptionInfo()

        # Size information
        if 'TotalPhysicalSize' in pool_interface:
            self.total_size = int(pool_interface['TotalPhysicalSize'])

        used_raw = dbus_maybe_value(pool_interface, 'TotalPhysicalUsed')
        if used_raw is not None:
            self.used_size = int(used_raw)

    def _update_from_report(self, report_data: ReportData) -> bool:
        """Update pool information from report data."""
        if self.prevent_update:
            logger.debug('Update from report skipped due to prevent_update flag')
            return True

        if 'pools' not in report_data:
            logger.error('Invalid report format')
            return False

        pools: list[Any] = report_data.get('pools', [])

        # Find the pool with matching name
        for pool in pools:
            if not isinstance(pool, dict):
                continue

            pd = cast('dict[str, Any]', pool)
            if not self.name or self.name == pd.get('name'):
                # Store raw data for access to fields not explicitly mapped
                self.raw_data = pd.copy()

                # Update explicit fields
                self.name = pd.get('name')
                self.uuid = pd.get('uuid')
                self.fs_limit = pd.get('fs_limit')
                self.available_actions = pd.get('available_actions')
                self.filesystems = _extract_filesystem_names(pd.get('filesystems', []))

                # Update blockdevs if present
                if 'blockdevs' in pd:
                    self.blockdevs = BlockDevs.model_validate(pd.get('blockdevs', {}))

                return True

        # If we get here and name was specified, pool wasn't found
        if self.name:
            logger.warning(f"Pool '{self.name}' not found in report")
            return False

        # If no name was specified and no pools exist
        if not pools:
            logger.warning('No pools found in report')
            return False

        return False

    def get_device_paths(self) -> list[str]:
        """Get all device paths from the pool.

        Returns:
            List of device paths for both data and cache devices
        """
        return [dev.path for dev in self.blockdevs.datadevs if dev.path] + [
            dev.path for dev in self.blockdevs.cachedevs if dev.path
        ]

    @classmethod
    def get_all(cls) -> list[PoolReport]:
        """Get reports for all pools.

        Returns:
            List of PoolReport instances
        """
        reports: list[PoolReport] = []

        base = cls()
        result = base.run_command('report')
        if result.failed or not result.stdout:
            return reports

        try:
            report_data = json.loads(result.stdout)

            if 'pools' in report_data and isinstance(report_data['pools'], list):
                for pool_data in report_data['pools']:
                    if not isinstance(pool_data, dict):
                        continue

                    pd = cast('dict[str, Any]', pool_data)
                    try:
                        report = cls(
                            name=pd.get('name'),
                            blockdevs=BlockDevs.model_validate(pd.get('blockdevs', {})),
                            uuid=pd.get('uuid'),
                            fs_limit=pd.get('fs_limit'),
                            available_actions=pd.get('available_actions'),
                            filesystems=_extract_filesystem_names(pd.get('filesystems', [])),
                            raw_data=pd.copy(),
                        )
                        if report.name:
                            report.update_from_managed_objects()
                        reports.append(report)
                    except (KeyError, TypeError) as e:
                        logger.warning(f'Invalid pool report data: {e}')

        except (json.JSONDecodeError, KeyError, ValueError) as e:
            logger.warning(f'Failed to parse pools: {e}')

        return reports

get_all() classmethod

Get reports for all pools.

Returns:

Type Description
list[PoolReport]

List of PoolReport instances
Source code in sts_libs/src/sts/stratis/pool.py 
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
@classmethod
def get_all(cls) -> list[PoolReport]:
    """Get reports for all pools.

    Returns:
        List of PoolReport instances
    """
    reports: list[PoolReport] = []

    base = cls()
    result = base.run_command('report')
    if result.failed or not result.stdout:
        return reports

    try:
        report_data = json.loads(result.stdout)

        if 'pools' in report_data and isinstance(report_data['pools'], list):
            for pool_data in report_data['pools']:
                if not isinstance(pool_data, dict):
                    continue

                pd = cast('dict[str, Any]', pool_data)
                try:
                    report = cls(
                        name=pd.get('name'),
                        blockdevs=BlockDevs.model_validate(pd.get('blockdevs', {})),
                        uuid=pd.get('uuid'),
                        fs_limit=pd.get('fs_limit'),
                        available_actions=pd.get('available_actions'),
                        filesystems=_extract_filesystem_names(pd.get('filesystems', [])),
                        raw_data=pd.copy(),
                    )
                    if report.name:
                        report.update_from_managed_objects()
                    reports.append(report)
                except (KeyError, TypeError) as e:
                    logger.warning(f'Invalid pool report data: {e}')

    except (json.JSONDecodeError, KeyError, ValueError) as e:
        logger.warning(f'Failed to parse pools: {e}')

    return reports

get_device_paths()

Get all device paths from the pool.

Returns:

Type Description
list[str]

List of device paths for both data and cache devices
Source code in sts_libs/src/sts/stratis/pool.py 
287
288
289
290
291
292
293
294
295
def get_device_paths(self) -> list[str]:
    """Get all device paths from the pool.

    Returns:
        List of device paths for both data and cache devices
    """
    return [dev.path for dev in self.blockdevs.datadevs if dev.path] + [
        dev.path for dev in self.blockdevs.cachedevs if dev.path
    ]

refresh()

Refresh pool report data from stratisd.

Source code in sts_libs/src/sts/stratis/pool.py 
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
def refresh(self) -> bool:
    """Refresh pool report data from stratisd."""
    # If prevent_update is True, skip refresh
    if self.prevent_update:
        logger.debug('Refresh skipped due to prevent_update flag')
        return True

    # First get standard report data
    result = self.run_command('report')
    if result.failed or not result.stdout:
        logger.error('Failed to get report data')
        return False

    try:
        report_data = json.loads(result.stdout)
        if not self._update_from_report(report_data):
            return False

        # If we have a name, also fetch size information from managed objects
        if self.name:
            self.update_from_managed_objects()

    except json.JSONDecodeError:
        logger.exception('Failed to parse report JSON')
        return False
    else:
        return True

update_from_managed_objects()

Fetch encryption info and total/used size from the managed objects interface.

Source code in sts_libs/src/sts/stratis/pool.py 
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
def update_from_managed_objects(self) -> bool:
    """Fetch encryption info and total/used size from the managed objects interface."""
    if self.prevent_update:
        logger.debug('Size info update skipped due to prevent_update flag')
        return True

    if not self.name:
        return False

    # Get object path
    result = self.run_command(
        subcommand='pool', action=['debug', 'get-object-path'], positional_args=['--name', self.name]
    )

    if result.failed or not result.stdout:
        logger.debug(f'Failed to get object path for pool {self.name}')
        return False

    self.object_path = result.stdout.strip()

    # Get managed objects report
    result = self.run_command(subcommand='report', action='managed_objects_report')

    if result.failed or not result.stdout:
        logger.error('Failed to get managed objects report')
        return False

    try:
        report_data = json.loads(result.stdout)

        if self.object_path not in report_data:
            logger.error(f'Object path {self.object_path} not found in managed objects report')
            return False

        pool_interfaces = report_data[self.object_path]
        last_revision = list(pool_interfaces.keys())[-1]
        self._parse_pool_interface(pool_interfaces[last_revision])

    except (json.JSONDecodeError, KeyError, ValueError):
        logger.exception('Error parsing managed objects data')
        return False
    else:
        return True

StratisPool pydantic-model

Bases: StratisBase

Stratis pool representation.

Manages Stratis pools including creation, encryption, and cache. Call refresh_report() after construction to populate report data.

Example 
pool = StratisPool(name='pool1', blockdevs=['/dev/sda'])
pool.create()  # create() calls refresh_report() internally
Show JSON schema: 
{
  "$defs": {
    "BlockDevInfo": {
      "description": "Block device information from stratis report.",
      "properties": {
        "path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Path"
        },
        "size": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "uuid": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uuid"
        },
        "in_use": {
          "default": false,
          "title": "In Use",
          "type": "boolean"
        },
        "blksizes": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Blksizes"
        },
        "clevis_config": {
          "anyOf": [
            {
              "additionalProperties": true,
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Clevis Config"
        },
        "clevis_pin": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Clevis Pin"
        },
        "key_description": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Key Description"
        }
      },
      "title": "BlockDevInfo",
      "type": "object"
    },
    "BlockDevs": {
      "description": "Block devices in a Stratis pool.\n\nPools have two storage tiers: ``datadevs`` for primary data storage and ``cachedevs`` for an\noptional fast cache layer (typically SSDs accelerating HDDs).",
      "properties": {
        "datadevs": {
          "items": {
            "$ref": "#/$defs/BlockDevInfo"
          },
          "title": "Datadevs",
          "type": "array"
        },
        "cachedevs": {
          "items": {
            "$ref": "#/$defs/BlockDevInfo"
          },
          "title": "Cachedevs",
          "type": "array"
        }
      },
      "title": "BlockDevs",
      "type": "object"
    },
    "EncryptionInfo": {
      "additionalProperties": false,
      "description": "Encryption metadata for a pool (key descriptions and Clevis bindings).\n\nUses `StsBaseModel` rather than `ReportModel`: `PoolReport` and `StratisPool`\nare mutable and reassign `self.encryption` wholesale (see\n`_parse_pool_interface`), which a frozen `ReportModel` would reject.\n\nv1 pools have at most one binding (singular ``KeyDescription``/``ClevisInfo``\nin the D-Bus interface); v2 pools support multiple (plural\n``KeyDescriptions``/``ClevisInfos``). Both are normalized to the plural\nfield names here \u2014 see `PoolReport._parse_pool_interface`.",
      "properties": {
        "key_descriptions": {
          "default": null,
          "title": "Key Descriptions"
        },
        "clevis_infos": {
          "default": null,
          "title": "Clevis Infos"
        }
      },
      "title": "EncryptionInfo",
      "type": "object"
    },
    "PoolReport": {
      "additionalProperties": false,
      "description": "Pool report data.\n\nMutable model that fetches and holds pool metadata from stratisd.\nCall ``refresh()`` after construction to populate from the system.",
      "properties": {
        "config": {
          "$ref": "#/$defs/StratisConfig"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "blockdevs": {
          "$ref": "#/$defs/BlockDevs"
        },
        "uuid": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uuid"
        },
        "encryption": {
          "$ref": "#/$defs/EncryptionInfo"
        },
        "encrypted": {
          "default": false,
          "title": "Encrypted",
          "type": "boolean"
        },
        "last_reencrypted_timestamp": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Last Reencrypted Timestamp"
        },
        "fs_limit": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fs Limit"
        },
        "available_actions": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Available Actions"
        },
        "filesystems": {
          "items": {
            "type": "string"
          },
          "title": "Filesystems",
          "type": "array"
        },
        "raw_data": {
          "additionalProperties": true,
          "title": "Raw Data",
          "type": "object"
        },
        "total_size": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Total Size"
        },
        "used_size": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Used Size"
        },
        "object_path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Object Path"
        },
        "prevent_update": {
          "default": false,
          "title": "Prevent Update",
          "type": "boolean"
        }
      },
      "title": "PoolReport",
      "type": "object"
    },
    "StratisConfig": {
      "additionalProperties": false,
      "description": "Stratis configuration controlling global CLI options.",
      "properties": {
        "unhyphenated_uuids": {
          "default": false,
          "title": "Unhyphenated Uuids",
          "type": "boolean"
        }
      },
      "title": "StratisConfig",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Stratis pool representation.\n\nManages Stratis pools including creation, encryption, and cache.\nCall ``refresh_report()`` after construction to populate report data.\n\nExample:\n    ```python\n    pool = StratisPool(name='pool1', blockdevs=['/dev/sda'])\n    pool.create()  # create() calls refresh_report() internally\n    ```",
  "properties": {
    "config": {
      "$ref": "#/$defs/StratisConfig"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "encryption": {
      "$ref": "#/$defs/EncryptionInfo"
    },
    "blockdevs": {
      "items": {
        "type": "string"
      },
      "title": "Blockdevs",
      "type": "array"
    },
    "report": {
      "anyOf": [
        {
          "$ref": "#/$defs/PoolReport"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "prevent_report_updates": {
      "default": false,
      "title": "Prevent Report Updates",
      "type": "boolean"
    }
  },
  "title": "StratisPool",
  "type": "object"
}

Fields:

Source code in sts_libs/src/sts/stratis/pool.py 
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
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
class StratisPool(StratisBase):
    """Stratis pool representation.

    Manages Stratis pools including creation, encryption, and cache.
    Call ``refresh_report()`` after construction to populate report data.

    Example:
        ```python
        pool = StratisPool(name='pool1', blockdevs=['/dev/sda'])
        pool.create()  # create() calls refresh_report() internally
        ```
    """

    name: str | None = None
    uuid: str | None = None
    encryption: EncryptionInfo = Field(default_factory=EncryptionInfo)
    blockdevs: list[str] = Field(default_factory=list)
    report: PoolReport | None = Field(default=None, repr=False)
    prevent_report_updates: bool = False  # Propagated to PoolReport.prevent_update when creating reports

    POOL_PATH: ClassVar[str] = '/stratis/pool'

    def _update_from_report(self) -> None:
        """Update pool attributes from report data.

        This centralizes all attribute updates from the report to avoid inconsistencies.
        """
        if self.prevent_report_updates:
            logger.debug('Update from report skipped due to prevent_report_updates flag')
            return

        if not self.report:
            return

        if not self.name and self.report.name:
            self.name = self.report.name

        if not self.uuid and self.report.uuid:
            self.uuid = self.report.uuid

        self.encryption = self.report.encryption
        self.blockdevs = self.report.get_device_paths()

    def refresh_report(self) -> bool:
        """Create or refresh the pool report with the latest stratisd data."""
        # Create new report if needed
        if not self.report:
            # Set name after init because we are explicitly calling refresh below
            # Setting name at init would result in calling .refresh() twice
            self.report = PoolReport(prevent_update=self.prevent_report_updates)
            self.report.name = self.name

        # Refresh the report data
        success = self.report.refresh()

        # Update pool fields from report if successful
        if success and not self.prevent_report_updates:
            self._update_from_report()

        return success

    def create(self, config: PoolCreateConfig | None = None) -> CommandResult:
        """Create pool."""
        options: StratisOptions = {}
        if config:
            if config.key_desc:
                options['--key-desc'] = config.key_desc
            if config.clevis:
                options['--clevis'] = config.clevis
            if config.tang_url:
                options['--tang-url'] = config.tang_url
            if config.thumbprint:
                options['--thumbprint'] = config.thumbprint
            if config.trust_url:
                options['--trust-url'] = None  # None = boolean flag, no argument (see StratisOptions in stratis.base)
            if config.no_overprovision:
                options['--no-overprovision'] = None

        result = self.run_command(
            subcommand='pool',
            action='create',
            options=options,
            positional_args=[self.name or '', *self.blockdevs],
        )

        if result.succeeded:
            self.refresh_report()

        return result

    def destroy(self) -> CommandResult:
        """Destroy pool."""
        return self.run_command(
            subcommand='pool',
            action='destroy',
            positional_args=[self.name or ''],
        )

    def list_pools(self, *, stopped: bool = False, current_pool: bool = True) -> str:
        """List information about pools.

        Args:
            stopped: Display information about stopped pools only
            current_pool: Filter to this pool (by UUID or name)
        """
        options: StratisOptions = {}
        if current_pool:
            if self.uuid:
                options['--uuid'] = self.uuid
            else:
                options['--name'] = self.name
        if stopped:
            options['--stopped'] = None

        result = self.run_command('pool', action='list', options=options)
        return result.stdout

    def start(self, unlock_method: UnlockMethod | None = None, token_slot: int | None = None) -> CommandResult:
        """Start pool.

        Args:
            unlock_method: Encryption unlock method
            token_slot: Token slot number for V2 pools
        """
        options: StratisOptions = {}
        if unlock_method:
            options['--unlock-method'] = unlock_method
        if token_slot is not None:
            options['--token-slot'] = str(token_slot)
        if self.uuid:
            options['--uuid'] = self.uuid
        else:
            options['--name'] = self.name or ''
        result = self.run_command('pool', action='start', options=options)

        if result.succeeded:
            self.refresh_report()

        return result

    def stop(self) -> CommandResult:
        """Stop pool."""
        options: StratisOptions = {}
        if self.uuid:
            options['--uuid'] = self.uuid
        else:
            options['--name'] = self.name or ''

        return self.run_command('pool', action='stop', options=options)

    def add_data(self, blockdevs: list[str]) -> CommandResult:
        """Add data devices to pool."""
        result = self.run_command(
            subcommand='pool',
            action='add-data',
            positional_args=[self.name or '', *blockdevs],
        )

        logger.debug(result)

        if result.succeeded:
            self.refresh_report()

        return result

    def extend_data(self) -> CommandResult:
        """Extend data devices in the pool to use the full device size."""
        result = self.run_command(
            subcommand='pool',
            action='extend-data',
            positional_args=[self.name or ''],
        )

        if result.succeeded:
            self.refresh_report()

        return result

    def init_cache(self, blockdevs: list[str]) -> CommandResult:
        """Initialize cache tier with the given block devices."""
        # Pass each device as a separate argument
        result = self.run_command(
            subcommand='pool',
            action='init-cache',
            positional_args=[self.name or '', *blockdevs],
        )

        if result.succeeded:
            self.refresh_report()
        else:
            logger.error(f'Failed to initialize cache: {result.stderr}')

        return result

    def add_cache(self, blockdevs: list[str]) -> CommandResult:
        """Add cache devices to pool."""
        result = self.run_command(
            subcommand='pool',
            action='add-cache',
            positional_args=[self.name or '', *blockdevs],
        )
        if result.succeeded:
            self.refresh_report()

        return result

    def bind_keyring(self, key_desc: str) -> CommandResult:
        """Bind pool encryption to a kernel keyring key."""
        result = self.run_command(
            subcommand='pool',
            action=['bind', 'keyring'],
            positional_args=[self.name or '', key_desc],
        )
        if result.succeeded:
            self.refresh_report()

        return result

    def bind_tang(self, config: TangConfig) -> CommandResult:
        """Bind pool encryption to a Tang server."""
        options: StratisOptions = {}
        if config.trust_url:
            options['--trust-url'] = None
        if config.thumbprint:
            options['--thumbprint'] = config.thumbprint

        result = self.run_command(
            subcommand='pool',
            action=['bind', 'tang'],
            options=options,
            positional_args=[self.name or '', config.url or ''],
        )
        if result.succeeded:
            self.refresh_report()

        return result

    def bind_tpm2(self) -> CommandResult:
        """Bind pool encryption to TPM2."""
        result = self.run_command(
            subcommand='pool',
            action=['bind', 'tpm2'],
            positional_args=[self.name or ''],
        )
        if result.succeeded:
            self.refresh_report()

        return result

    def rebind_keyring(self, key_desc: str) -> CommandResult:
        """Rebind pool encryption to a different keyring key."""
        result = self.run_command(
            subcommand='pool',
            action=['rebind', 'keyring'],
            positional_args=[self.name or '', key_desc],
        )

        if result.succeeded:
            self.refresh_report()

        return result

    def rebind_clevis(self) -> CommandResult:
        """Rebind pool to Clevis."""
        result = self.run_command(
            subcommand='pool',
            action=['rebind', 'clevis'],
            positional_args=[self.name or ''],
        )

        if result.succeeded:
            self.refresh_report()

        return result

    def unbind_keyring(self) -> CommandResult:
        """Unbind pool from keyring."""
        result = self.run_command(
            subcommand='pool',
            action=['unbind', 'keyring'],
            positional_args=[self.name or ''],
        )
        if result.succeeded:
            self.refresh_report()

        return result

    def unbind_clevis(self) -> CommandResult:
        """Unbind pool from Clevis."""
        result = self.run_command(
            subcommand='pool',
            action=['unbind', 'clevis'],
            positional_args=[self.name or ''],
        )
        if result.succeeded:
            self.refresh_report()

        return result

    def encryption_on(
        self,
        *,
        key_desc: str | None = None,
        clevis: str | None = None,
        tang_url: str | None = None,
        thumbprint: str | None = None,
        trust_url: bool = False,
        in_place: bool = False,
    ) -> CommandResult:
        """Enable encryption on an unencrypted pool.

        This is a long-running operation (stratis >= 3.9.0). The CLI prints
        "Operation initiated" after ~10 seconds and returns, but the actual
        encryption continues in the background within stratisd. The pool's
        "Encryption Enabled" status is updated upon completion.

        At least one of key_desc or clevis must be specified.

        Args:
            key_desc: Key description of key in kernel keyring
            clevis: Clevis encryption specification (nbde, tang, tpm2)
            tang_url: URL of Clevis tang server (requires clevis=tang or clevis=nbde)
            thumbprint: Thumbprint of tang server at specified URL
            trust_url: Trust tang server URL without verification
            in_place: Perform the operation in place without additional devices
        """
        options: StratisOptions = {}
        if self.uuid:
            options['--uuid'] = self.uuid
        else:
            options['--name'] = self.name or ''
        if key_desc:
            options['--key-desc'] = key_desc
        if clevis:
            options['--clevis'] = clevis
        if tang_url:
            options['--tang-url'] = tang_url
        if thumbprint:
            options['--thumbprint'] = thumbprint
        if trust_url:
            options['--trust-url'] = None
        if in_place:
            options['--in-place'] = None

        return self.run_command(
            subcommand='pool',
            action=['encryption', 'on'],
            options=options,
        )

    def encryption_off(self, *, in_place: bool = False) -> CommandResult:
        """Disable encryption on an encrypted pool.

        This is a long-running operation (stratis >= 3.9.0). The CLI prints
        "Operation initiated" after ~10 seconds and returns, but the actual
        decryption continues in the background within stratisd. The pool's
        "Encryption Enabled" status is updated upon completion.

        Args:
            in_place: Perform the operation in place without additional devices
        """
        options: StratisOptions = {}
        if self.uuid:
            options['--uuid'] = self.uuid
        else:
            options['--name'] = self.name or ''
        if in_place:
            options['--in-place'] = None

        return self.run_command(
            subcommand='pool',
            action=['encryption', 'off'],
            options=options,
        )

    def encryption_reencrypt(self, *, in_place: bool = False) -> CommandResult:
        """Reencrypt an encrypted pool with a new master key.

        This is a long-running operation (stratis >= 3.9.0). The CLI prints
        "Operation initiated" after ~10 seconds and returns, but the actual
        reencryption continues in the background within stratisd. The pool's
        "Last Time Reencrypted" counter is incremented upon completion.
        Turning encryption off and back on resets the counter to 0.

        Args:
            in_place: Perform the operation in place without additional devices
        """
        options: StratisOptions = {}
        if self.uuid:
            options['--uuid'] = self.uuid
        else:
            options['--name'] = self.name or ''
        if in_place:
            options['--in-place'] = None

        return self.run_command(
            subcommand='pool',
            action=['encryption', 'reencrypt'],
            options=options,
        )

    def rename(self, new_name: str) -> CommandResult:
        """Rename pool."""
        result = self.run_command(
            subcommand='pool',
            action='rename',
            positional_args=[self.name or '', new_name],
        )
        if result.succeeded:
            self.name = new_name
            if self.report:
                self.report.name = new_name
            self.refresh_report()

        return result

    def set_fs_limit(self, amount: str) -> CommandResult:
        """Set the limit on the number of filesystems allowed per-pool.

        This number may only be increased from its current value.
        """
        result = self.run_command(
            subcommand='pool',
            action='set-fs-limit',
            positional_args=[self.name or '', amount],
        )
        if result.succeeded:
            self.refresh_report()

        return result

    def overprovision(self, enable: str) -> CommandResult:
        """Set overprovisioning mode for pool.

        If set to "yes", the pool may allow overprovisioning, i.e., the sum
        of the logical sizes of the Stratis filesystems supported by the pool
        may exceed the amount of data space available.

        Args:
            enable: "yes" or "no"
        """
        result = self.run_command(
            subcommand='pool',
            action='overprovision',
            positional_args=[self.name or '', enable],
        )
        if result.succeeded:
            self.refresh_report()

        return result

    @staticmethod
    def explain(code: str) -> str:
        """Explain a pool alert code (e.g. "WS001").

        Returns explanation text from stratis, or empty string on failure.
        """
        base = StratisBase(config=StratisConfig())
        result = base.run_command(
            subcommand='pool',
            action='explain',
            positional_args=[code],
        )
        return result.stdout if not result.failed else ''

    @classmethod
    def from_report(cls, report: PoolReport) -> StratisPool | None:
        """Create pool from report data."""
        if not report.name:
            return None

        # Get paths from report
        paths = report.get_device_paths()

        # Create pool with report already attached
        return cls(
            name=report.name,
            uuid=report.uuid,
            encryption=report.encryption,
            blockdevs=paths,
            report=report,  # Attach the report directly
        )

    @classmethod
    def get_all(cls) -> list[StratisPool]:
        """Get all Stratis pools."""
        pools: list[StratisPool] = []

        # Get all reports
        reports = PoolReport.get_all()

        # Create pools from reports
        pools.extend(pool for report in reports if (pool := cls.from_report(report)))

        return pools

    @classmethod
    def setup_blockdevices(cls) -> list[str]:
        """Prepare physical disks for Stratis pool creation (test helper).

        Lives in production code because it needs ``get_free_disks()``. Groups disks by
        (sector_size, block_size) and selects the largest group — stratisd requires uniform block
        geometry within a pool. Zeroes the first 10 MiB of each selected disk to clear partition
        tables, filesystem signatures, and LUKS headers.
        """
        blockdevices = get_free_disks()
        if not blockdevices:
            logger.warning('No free disks found')
            return []

        # Group disks by block sizes
        filtered_disks_by_block_sizes: dict[tuple[int, int], list[str]] = {}
        for disk in blockdevices:
            block_sizes = (disk.sector_size, disk.block_size)
            if block_sizes in filtered_disks_by_block_sizes:
                filtered_disks_by_block_sizes[block_sizes].append(str(disk.path))
            else:
                filtered_disks_by_block_sizes[block_sizes] = [str(disk.path)]

        # Find devices with the most common block sizes
        most_common_block_sizes: list[str] = []
        for disks in filtered_disks_by_block_sizes.values():
            if len(disks) > len(most_common_block_sizes):
                most_common_block_sizes = disks

        # Clear start of devices
        for disk in most_common_block_sizes:
            run(f'dd if=/dev/zero of={disk} bs=1M count=10')

        return most_common_block_sizes

add_cache(blockdevs)

Add cache devices to pool.

Source code in sts_libs/src/sts/stratis/pool.py 
561
562
563
564
565
566
567
568
569
570
571
def add_cache(self, blockdevs: list[str]) -> CommandResult:
    """Add cache devices to pool."""
    result = self.run_command(
        subcommand='pool',
        action='add-cache',
        positional_args=[self.name or '', *blockdevs],
    )
    if result.succeeded:
        self.refresh_report()

    return result

add_data(blockdevs)

Add data devices to pool.

Source code in sts_libs/src/sts/stratis/pool.py 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
def add_data(self, blockdevs: list[str]) -> CommandResult:
    """Add data devices to pool."""
    result = self.run_command(
        subcommand='pool',
        action='add-data',
        positional_args=[self.name or '', *blockdevs],
    )

    logger.debug(result)

    if result.succeeded:
        self.refresh_report()

    return result

bind_keyring(key_desc)

Bind pool encryption to a kernel keyring key.

Source code in sts_libs/src/sts/stratis/pool.py 
573
574
575
576
577
578
579
580
581
582
583
def bind_keyring(self, key_desc: str) -> CommandResult:
    """Bind pool encryption to a kernel keyring key."""
    result = self.run_command(
        subcommand='pool',
        action=['bind', 'keyring'],
        positional_args=[self.name or '', key_desc],
    )
    if result.succeeded:
        self.refresh_report()

    return result

bind_tang(config)

Bind pool encryption to a Tang server.

Source code in sts_libs/src/sts/stratis/pool.py 
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
def bind_tang(self, config: TangConfig) -> CommandResult:
    """Bind pool encryption to a Tang server."""
    options: StratisOptions = {}
    if config.trust_url:
        options['--trust-url'] = None
    if config.thumbprint:
        options['--thumbprint'] = config.thumbprint

    result = self.run_command(
        subcommand='pool',
        action=['bind', 'tang'],
        options=options,
        positional_args=[self.name or '', config.url or ''],
    )
    if result.succeeded:
        self.refresh_report()

    return result

bind_tpm2()

Bind pool encryption to TPM2.

Source code in sts_libs/src/sts/stratis/pool.py 
604
605
606
607
608
609
610
611
612
613
614
def bind_tpm2(self) -> CommandResult:
    """Bind pool encryption to TPM2."""
    result = self.run_command(
        subcommand='pool',
        action=['bind', 'tpm2'],
        positional_args=[self.name or ''],
    )
    if result.succeeded:
        self.refresh_report()

    return result

create(config=None)

Create pool.

Source code in sts_libs/src/sts/stratis/pool.py 
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
def create(self, config: PoolCreateConfig | None = None) -> CommandResult:
    """Create pool."""
    options: StratisOptions = {}
    if config:
        if config.key_desc:
            options['--key-desc'] = config.key_desc
        if config.clevis:
            options['--clevis'] = config.clevis
        if config.tang_url:
            options['--tang-url'] = config.tang_url
        if config.thumbprint:
            options['--thumbprint'] = config.thumbprint
        if config.trust_url:
            options['--trust-url'] = None  # None = boolean flag, no argument (see StratisOptions in stratis.base)
        if config.no_overprovision:
            options['--no-overprovision'] = None

    result = self.run_command(
        subcommand='pool',
        action='create',
        options=options,
        positional_args=[self.name or '', *self.blockdevs],
    )

    if result.succeeded:
        self.refresh_report()

    return result

destroy()

Destroy pool.

Source code in sts_libs/src/sts/stratis/pool.py 
457
458
459
460
461
462
463
def destroy(self) -> CommandResult:
    """Destroy pool."""
    return self.run_command(
        subcommand='pool',
        action='destroy',
        positional_args=[self.name or ''],
    )

encryption_off(*, in_place=False)

Disable encryption on an encrypted pool.

This is a long-running operation (stratis >= 3.9.0). The CLI prints "Operation initiated" after ~10 seconds and returns, but the actual decryption continues in the background within stratisd. The pool's "Encryption Enabled" status is updated upon completion.

Parameters:

Name Type Description Default
in_place bool

Perform the operation in place without additional devices

 False
Source code in sts_libs/src/sts/stratis/pool.py 
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
def encryption_off(self, *, in_place: bool = False) -> CommandResult:
    """Disable encryption on an encrypted pool.

    This is a long-running operation (stratis >= 3.9.0). The CLI prints
    "Operation initiated" after ~10 seconds and returns, but the actual
    decryption continues in the background within stratisd. The pool's
    "Encryption Enabled" status is updated upon completion.

    Args:
        in_place: Perform the operation in place without additional devices
    """
    options: StratisOptions = {}
    if self.uuid:
        options['--uuid'] = self.uuid
    else:
        options['--name'] = self.name or ''
    if in_place:
        options['--in-place'] = None

    return self.run_command(
        subcommand='pool',
        action=['encryption', 'off'],
        options=options,
    )

encryption_on(*, key_desc=None, clevis=None, tang_url=None, thumbprint=None, trust_url=False, in_place=False)

Enable encryption on an unencrypted pool.

This is a long-running operation (stratis >= 3.9.0). The CLI prints "Operation initiated" after ~10 seconds and returns, but the actual encryption continues in the background within stratisd. The pool's "Encryption Enabled" status is updated upon completion.

At least one of key_desc or clevis must be specified.

Parameters:

Name Type Description Default
key_desc str | None

Key description of key in kernel keyring

 None
clevis str | None

Clevis encryption specification (nbde, tang, tpm2)

 None
tang_url str | None

URL of Clevis tang server (requires clevis=tang or clevis=nbde)

 None
thumbprint str | None

Thumbprint of tang server at specified URL

 None
trust_url bool

Trust tang server URL without verification

 False
in_place bool

Perform the operation in place without additional devices

 False
Source code in sts_libs/src/sts/stratis/pool.py 
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
def encryption_on(
    self,
    *,
    key_desc: str | None = None,
    clevis: str | None = None,
    tang_url: str | None = None,
    thumbprint: str | None = None,
    trust_url: bool = False,
    in_place: bool = False,
) -> CommandResult:
    """Enable encryption on an unencrypted pool.

    This is a long-running operation (stratis >= 3.9.0). The CLI prints
    "Operation initiated" after ~10 seconds and returns, but the actual
    encryption continues in the background within stratisd. The pool's
    "Encryption Enabled" status is updated upon completion.

    At least one of key_desc or clevis must be specified.

    Args:
        key_desc: Key description of key in kernel keyring
        clevis: Clevis encryption specification (nbde, tang, tpm2)
        tang_url: URL of Clevis tang server (requires clevis=tang or clevis=nbde)
        thumbprint: Thumbprint of tang server at specified URL
        trust_url: Trust tang server URL without verification
        in_place: Perform the operation in place without additional devices
    """
    options: StratisOptions = {}
    if self.uuid:
        options['--uuid'] = self.uuid
    else:
        options['--name'] = self.name or ''
    if key_desc:
        options['--key-desc'] = key_desc
    if clevis:
        options['--clevis'] = clevis
    if tang_url:
        options['--tang-url'] = tang_url
    if thumbprint:
        options['--thumbprint'] = thumbprint
    if trust_url:
        options['--trust-url'] = None
    if in_place:
        options['--in-place'] = None

    return self.run_command(
        subcommand='pool',
        action=['encryption', 'on'],
        options=options,
    )

encryption_reencrypt(*, in_place=False)

Reencrypt an encrypted pool with a new master key.

This is a long-running operation (stratis >= 3.9.0). The CLI prints "Operation initiated" after ~10 seconds and returns, but the actual reencryption continues in the background within stratisd. The pool's "Last Time Reencrypted" counter is incremented upon completion. Turning encryption off and back on resets the counter to 0.

Parameters:

Name Type Description Default
in_place bool

Perform the operation in place without additional devices

 False
Source code in sts_libs/src/sts/stratis/pool.py 
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
def encryption_reencrypt(self, *, in_place: bool = False) -> CommandResult:
    """Reencrypt an encrypted pool with a new master key.

    This is a long-running operation (stratis >= 3.9.0). The CLI prints
    "Operation initiated" after ~10 seconds and returns, but the actual
    reencryption continues in the background within stratisd. The pool's
    "Last Time Reencrypted" counter is incremented upon completion.
    Turning encryption off and back on resets the counter to 0.

    Args:
        in_place: Perform the operation in place without additional devices
    """
    options: StratisOptions = {}
    if self.uuid:
        options['--uuid'] = self.uuid
    else:
        options['--name'] = self.name or ''
    if in_place:
        options['--in-place'] = None

    return self.run_command(
        subcommand='pool',
        action=['encryption', 'reencrypt'],
        options=options,
    )

explain(code) staticmethod

Explain a pool alert code (e.g. "WS001").

Returns explanation text from stratis, or empty string on failure.

Source code in sts_libs/src/sts/stratis/pool.py 
818
819
820
821
822
823
824
825
826
827
828
829
830
@staticmethod
def explain(code: str) -> str:
    """Explain a pool alert code (e.g. "WS001").

    Returns explanation text from stratis, or empty string on failure.
    """
    base = StratisBase(config=StratisConfig())
    result = base.run_command(
        subcommand='pool',
        action='explain',
        positional_args=[code],
    )
    return result.stdout if not result.failed else ''

extend_data()

Extend data devices in the pool to use the full device size.

Source code in sts_libs/src/sts/stratis/pool.py 
532
533
534
535
536
537
538
539
540
541
542
543
def extend_data(self) -> CommandResult:
    """Extend data devices in the pool to use the full device size."""
    result = self.run_command(
        subcommand='pool',
        action='extend-data',
        positional_args=[self.name or ''],
    )

    if result.succeeded:
        self.refresh_report()

    return result

from_report(report) classmethod

Create pool from report data.

Source code in sts_libs/src/sts/stratis/pool.py 
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
@classmethod
def from_report(cls, report: PoolReport) -> StratisPool | None:
    """Create pool from report data."""
    if not report.name:
        return None

    # Get paths from report
    paths = report.get_device_paths()

    # Create pool with report already attached
    return cls(
        name=report.name,
        uuid=report.uuid,
        encryption=report.encryption,
        blockdevs=paths,
        report=report,  # Attach the report directly
    )

get_all() classmethod

Get all Stratis pools.

Source code in sts_libs/src/sts/stratis/pool.py 
850
851
852
853
854
855
856
857
858
859
860
861
@classmethod
def get_all(cls) -> list[StratisPool]:
    """Get all Stratis pools."""
    pools: list[StratisPool] = []

    # Get all reports
    reports = PoolReport.get_all()

    # Create pools from reports
    pools.extend(pool for report in reports if (pool := cls.from_report(report)))

    return pools

init_cache(blockdevs)

Initialize cache tier with the given block devices.

Source code in sts_libs/src/sts/stratis/pool.py 
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
def init_cache(self, blockdevs: list[str]) -> CommandResult:
    """Initialize cache tier with the given block devices."""
    # Pass each device as a separate argument
    result = self.run_command(
        subcommand='pool',
        action='init-cache',
        positional_args=[self.name or '', *blockdevs],
    )

    if result.succeeded:
        self.refresh_report()
    else:
        logger.error(f'Failed to initialize cache: {result.stderr}')

    return result

list_pools(*, stopped=False, current_pool=True)

List information about pools.

Parameters:

Name Type Description Default
stopped bool

Display information about stopped pools only

 False
current_pool bool

Filter to this pool (by UUID or name)

 True
Source code in sts_libs/src/sts/stratis/pool.py 
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
def list_pools(self, *, stopped: bool = False, current_pool: bool = True) -> str:
    """List information about pools.

    Args:
        stopped: Display information about stopped pools only
        current_pool: Filter to this pool (by UUID or name)
    """
    options: StratisOptions = {}
    if current_pool:
        if self.uuid:
            options['--uuid'] = self.uuid
        else:
            options['--name'] = self.name
    if stopped:
        options['--stopped'] = None

    result = self.run_command('pool', action='list', options=options)
    return result.stdout

overprovision(enable)

Set overprovisioning mode for pool.

If set to "yes", the pool may allow overprovisioning, i.e., the sum of the logical sizes of the Stratis filesystems supported by the pool may exceed the amount of data space available.

Parameters:

Name Type Description Default
enable str

"yes" or "no"

 required
Source code in sts_libs/src/sts/stratis/pool.py 
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
def overprovision(self, enable: str) -> CommandResult:
    """Set overprovisioning mode for pool.

    If set to "yes", the pool may allow overprovisioning, i.e., the sum
    of the logical sizes of the Stratis filesystems supported by the pool
    may exceed the amount of data space available.

    Args:
        enable: "yes" or "no"
    """
    result = self.run_command(
        subcommand='pool',
        action='overprovision',
        positional_args=[self.name or '', enable],
    )
    if result.succeeded:
        self.refresh_report()

    return result

rebind_clevis()

Rebind pool to Clevis.

Source code in sts_libs/src/sts/stratis/pool.py 
629
630
631
632
633
634
635
636
637
638
639
640
def rebind_clevis(self) -> CommandResult:
    """Rebind pool to Clevis."""
    result = self.run_command(
        subcommand='pool',
        action=['rebind', 'clevis'],
        positional_args=[self.name or ''],
    )

    if result.succeeded:
        self.refresh_report()

    return result

rebind_keyring(key_desc)

Rebind pool encryption to a different keyring key.

Source code in sts_libs/src/sts/stratis/pool.py 
616
617
618
619
620
621
622
623
624
625
626
627
def rebind_keyring(self, key_desc: str) -> CommandResult:
    """Rebind pool encryption to a different keyring key."""
    result = self.run_command(
        subcommand='pool',
        action=['rebind', 'keyring'],
        positional_args=[self.name or '', key_desc],
    )

    if result.succeeded:
        self.refresh_report()

    return result

refresh_report()

Create or refresh the pool report with the latest stratisd data.

Source code in sts_libs/src/sts/stratis/pool.py 
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
def refresh_report(self) -> bool:
    """Create or refresh the pool report with the latest stratisd data."""
    # Create new report if needed
    if not self.report:
        # Set name after init because we are explicitly calling refresh below
        # Setting name at init would result in calling .refresh() twice
        self.report = PoolReport(prevent_update=self.prevent_report_updates)
        self.report.name = self.name

    # Refresh the report data
    success = self.report.refresh()

    # Update pool fields from report if successful
    if success and not self.prevent_report_updates:
        self._update_from_report()

    return success

rename(new_name)

Rename pool.

Source code in sts_libs/src/sts/stratis/pool.py 
768
769
770
771
772
773
774
775
776
777
778
779
780
781
def rename(self, new_name: str) -> CommandResult:
    """Rename pool."""
    result = self.run_command(
        subcommand='pool',
        action='rename',
        positional_args=[self.name or '', new_name],
    )
    if result.succeeded:
        self.name = new_name
        if self.report:
            self.report.name = new_name
        self.refresh_report()

    return result

set_fs_limit(amount)

Set the limit on the number of filesystems allowed per-pool.

This number may only be increased from its current value.

Source code in sts_libs/src/sts/stratis/pool.py 
783
784
785
786
787
788
789
790
791
792
793
794
795
796
def set_fs_limit(self, amount: str) -> CommandResult:
    """Set the limit on the number of filesystems allowed per-pool.

    This number may only be increased from its current value.
    """
    result = self.run_command(
        subcommand='pool',
        action='set-fs-limit',
        positional_args=[self.name or '', amount],
    )
    if result.succeeded:
        self.refresh_report()

    return result

setup_blockdevices() classmethod

Prepare physical disks for Stratis pool creation (test helper).

Lives in production code because it needs get_free_disks(). Groups disks by (sector_size, block_size) and selects the largest group — stratisd requires uniform block geometry within a pool. Zeroes the first 10 MiB of each selected disk to clear partition tables, filesystem signatures, and LUKS headers.

Source code in sts_libs/src/sts/stratis/pool.py 
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
@classmethod
def setup_blockdevices(cls) -> list[str]:
    """Prepare physical disks for Stratis pool creation (test helper).

    Lives in production code because it needs ``get_free_disks()``. Groups disks by
    (sector_size, block_size) and selects the largest group — stratisd requires uniform block
    geometry within a pool. Zeroes the first 10 MiB of each selected disk to clear partition
    tables, filesystem signatures, and LUKS headers.
    """
    blockdevices = get_free_disks()
    if not blockdevices:
        logger.warning('No free disks found')
        return []

    # Group disks by block sizes
    filtered_disks_by_block_sizes: dict[tuple[int, int], list[str]] = {}
    for disk in blockdevices:
        block_sizes = (disk.sector_size, disk.block_size)
        if block_sizes in filtered_disks_by_block_sizes:
            filtered_disks_by_block_sizes[block_sizes].append(str(disk.path))
        else:
            filtered_disks_by_block_sizes[block_sizes] = [str(disk.path)]

    # Find devices with the most common block sizes
    most_common_block_sizes: list[str] = []
    for disks in filtered_disks_by_block_sizes.values():
        if len(disks) > len(most_common_block_sizes):
            most_common_block_sizes = disks

    # Clear start of devices
    for disk in most_common_block_sizes:
        run(f'dd if=/dev/zero of={disk} bs=1M count=10')

    return most_common_block_sizes

start(unlock_method=None, token_slot=None)

Start pool.

Parameters:

Name Type Description Default
unlock_method UnlockMethod | None

Encryption unlock method

 None
token_slot int | None

Token slot number for V2 pools

 None
Source code in sts_libs/src/sts/stratis/pool.py 
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
def start(self, unlock_method: UnlockMethod | None = None, token_slot: int | None = None) -> CommandResult:
    """Start pool.

    Args:
        unlock_method: Encryption unlock method
        token_slot: Token slot number for V2 pools
    """
    options: StratisOptions = {}
    if unlock_method:
        options['--unlock-method'] = unlock_method
    if token_slot is not None:
        options['--token-slot'] = str(token_slot)
    if self.uuid:
        options['--uuid'] = self.uuid
    else:
        options['--name'] = self.name or ''
    result = self.run_command('pool', action='start', options=options)

    if result.succeeded:
        self.refresh_report()

    return result

stop()

Stop pool.

Source code in sts_libs/src/sts/stratis/pool.py 
507
508
509
510
511
512
513
514
515
def stop(self) -> CommandResult:
    """Stop pool."""
    options: StratisOptions = {}
    if self.uuid:
        options['--uuid'] = self.uuid
    else:
        options['--name'] = self.name or ''

    return self.run_command('pool', action='stop', options=options)

unbind_clevis()

Unbind pool from Clevis.

Source code in sts_libs/src/sts/stratis/pool.py 
654
655
656
657
658
659
660
661
662
663
664
def unbind_clevis(self) -> CommandResult:
    """Unbind pool from Clevis."""
    result = self.run_command(
        subcommand='pool',
        action=['unbind', 'clevis'],
        positional_args=[self.name or ''],
    )
    if result.succeeded:
        self.refresh_report()

    return result

unbind_keyring()

Unbind pool from keyring.

Source code in sts_libs/src/sts/stratis/pool.py 
642
643
644
645
646
647
648
649
650
651
652
def unbind_keyring(self) -> CommandResult:
    """Unbind pool from keyring."""
    result = self.run_command(
        subcommand='pool',
        action=['unbind', 'keyring'],
        positional_args=[self.name or ''],
    )
    if result.succeeded:
        self.refresh_report()

    return result

TangConfig pydantic-model

Bases: StsBaseModel

Tang server configuration for Clevis NBDE encryption.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Tang server configuration for Clevis NBDE encryption.",
  "properties": {
    "url": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Url"
    },
    "trust_url": {
      "default": false,
      "title": "Trust Url",
      "type": "boolean"
    },
    "thumbprint": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Thumbprint"
    }
  },
  "title": "TangConfig",
  "type": "object"
}

Fields:

  • url (str | None)
  • trust_url (bool)
  • thumbprint (str | None)
Source code in sts_libs/src/sts/stratis/pool.py 
359
360
361
362
363
364
class TangConfig(StsBaseModel):
    """Tang server configuration for Clevis NBDE encryption."""

    url: str | None = None
    trust_url: bool = False
    thumbprint: str | None = None

Filesystem Management

sts.stratis.filesystem

Stratis filesystem management.

Design notes

Each Stratis filesystem lives inside exactly one pool and uses XFS as the underlying filesystem type. Filesystems are thin-provisioned — they consume pool space only as data is written.

Two data sources are used:

  • Basic metadata (name, UUID, origin) comes from stratis report and is parsed into FilesystemReport by get_all().
  • Byte-accurate size/used/limit values come from stratis report managed_objects_report and are applied by update_from_managed_objects().

See stratis.base module docstring for cross-cutting patterns (D-Bus Maybe-type, two-tier data fetching, last_revision).

FilesystemReport pydantic-model

Bases: ReportModel

Parsed filesystem data from stratis report.

Frozen (immutable) snapshot of one filesystem's report fields. Unknown JSON keys are silently ignored (extra='ignore').

Show JSON schema: 
{
  "description": "Parsed filesystem data from stratis report.\n\nFrozen (immutable) snapshot of one filesystem's report fields.\nUnknown JSON keys are silently ignored (extra='ignore').",
  "properties": {
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "size_limit": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size Limit"
    },
    "origin": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Origin"
    },
    "used": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Used"
    }
  },
  "title": "FilesystemReport",
  "type": "object"
}

Fields:

  • name (str | None)
  • uuid (str | None)
  • size (str | None)
  • size_limit (str | None)
  • origin (str | None)
  • used (str | None)

Validators:

  • _normalize_stratis_sentinel → size_limit, origin, used
Source code in sts_libs/src/sts/stratis/filesystem.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class FilesystemReport(ReportModel):
    """Parsed filesystem data from stratis report.

    Frozen (immutable) snapshot of one filesystem's report fields.
    Unknown JSON keys are silently ignored (extra='ignore').
    """

    name: str | None = None
    uuid: str | None = None
    size: str | None = None  # Human-readable size (e.g. '10 GiB'); contrast with StratisFilesystem.size (int, bytes)
    size_limit: str | None = None  # Maximum size allowed
    origin: str | None = None  # Source filesystem for snapshots
    used: str | None = None  # Space currently in use

    @field_validator('size_limit', 'origin', 'used', mode='before')
    @classmethod
    def _normalize_stratis_sentinel(cls, v: str | None) -> str | None:
        """Convert stratis 'Not set' sentinel to None."""
        if not v or v == 'Not set':
            return None
        return v

StratisFilesystem pydantic-model

Bases: StratisBase

Stratis filesystem representation.

Manages filesystems with thin provisioning, snapshots, and size management. Call update_from_managed_objects() after construction to populate size/used/limit from stratisd.

Example 
fs = StratisFilesystem(name='fs1', pool_name='pool1')
fs.create()
Show JSON schema: 
{
  "$defs": {
    "StratisConfig": {
      "additionalProperties": false,
      "description": "Stratis configuration controlling global CLI options.",
      "properties": {
        "unhyphenated_uuids": {
          "default": false,
          "title": "Unhyphenated Uuids",
          "type": "boolean"
        }
      },
      "title": "StratisConfig",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Stratis filesystem representation.\n\nManages filesystems with thin provisioning, snapshots, and size\nmanagement. Call ``update_from_managed_objects()`` after construction\nto populate size/used/limit from stratisd.\n\nExample:\n    ```python\n    fs = StratisFilesystem(name='fs1', pool_name='pool1')\n    fs.create()\n    ```",
  "properties": {
    "config": {
      "$ref": "#/$defs/StratisConfig"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "pool_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Pool Name"
    },
    "uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "size": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "size_limit": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size Limit"
    },
    "origin": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Origin"
    },
    "used": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Used"
    }
  },
  "title": "StratisFilesystem",
  "type": "object"
}

Fields:

  • config (StratisConfig)
  • name (str | None)
  • pool_name (str | None)
  • uuid (str | None)
  • size (int | None)
  • size_limit (int | None)
  • origin (str | None)
  • used (int | None)
Source code in sts_libs/src/sts/stratis/filesystem.py 
 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
class StratisFilesystem(StratisBase):
    """Stratis filesystem representation.

    Manages filesystems with thin provisioning, snapshots, and size
    management. Call ``update_from_managed_objects()`` after construction
    to populate size/used/limit from stratisd.

    Example:
        ```python
        fs = StratisFilesystem(name='fs1', pool_name='pool1')
        fs.create()
        ```
    """

    name: str | None = None
    pool_name: str | None = None
    uuid: str | None = None
    size: int | None = None  # Size in bytes from managed_objects_report; contrast with FilesystemReport.size (str)
    size_limit: int | None = None  # Size limit in bytes
    origin: str | None = None  # Source filesystem for snapshots
    used: int | None = None  # Used space in bytes

    # Mount point base path — Stratis filesystems mount at /stratis/<pool_name>/<fs_name>
    FS_PATH: ClassVar[str] = '/stratis'

    def update_from_managed_objects(self) -> bool:
        """Update size/used/limit from ``stratis report managed_objects_report``.

        Uses the managed-objects report which provides clean byte values
        directly, avoiding sector/unit parsing issues in the basic report.
        """
        if not self.name:
            return False

        # Get filesystem object path
        result = self.run_command(
            subcommand='filesystem',
            action=['debug', 'get-object-path'],
            positional_args=['--name', self.name],
        )
        if result.failed or not result.stdout:
            logger.debug(f'Failed to get object path for filesystem {self.name}')
            return False

        object_path = result.stdout.strip()

        # Get managed objects report
        result = self.run_command(subcommand='report', action='managed_objects_report')
        if result.failed or not result.stdout:
            logger.error('Failed to get managed objects report')
            return False

        try:
            report_data = json.loads(result.stdout)

            if object_path not in report_data:
                logger.error(f'Object path {object_path} not found in managed objects report')
                return False

            fs_interfaces = report_data[object_path]

            # Highest D-Bus interface revision has the most complete properties (see stratis.base patterns)
            last_revision = list(fs_interfaces.keys())[-1]
            fs_data = fs_interfaces[last_revision]

            # Size is a plain bytes string
            if 'Size' in fs_data:
                self.size = int(fs_data['Size'])

            used_raw = dbus_maybe_value(fs_data, 'Used')
            if used_raw is not None:
                self.used = int(used_raw)

            limit_raw = dbus_maybe_value(fs_data, 'SizeLimit')
            if limit_raw is not None:
                self.size_limit = int(limit_raw)

        except (json.JSONDecodeError, KeyError, ValueError):
            logger.exception('Error parsing managed objects data')
            return False
        else:
            return True

    def get_fs_uuid(self) -> str | None:
        """Get filesystem UUID from the stratis report."""
        result = self.run_command('report')
        if result.failed or not result.stdout:
            return None

        try:
            report = json.loads(result.stdout)
            for pool in report['pools']:
                if self.pool_name != pool['name']:
                    continue
                for fs in pool['filesystems']:
                    if self.name != fs['name']:
                        continue
                    return fs['uuid']
        except (KeyError, ValueError) as e:
            logger.warning(f'Failed to get filesystem UUID: {e}')

        return None

    def create(self, size: str | None = None, size_limit: str | None = None) -> CommandResult:
        """Create filesystem.

        Args:
            size: Initial size (e.g. "10G")
            size_limit: Size limit (e.g. "20G")
        """
        options: StratisOptions = {}
        if size:
            options['--size'] = size
        if size_limit:
            options['--size-limit'] = size_limit

        return self.run_command(
            subcommand='filesystem',
            action='create',
            options=options,
            positional_args=[self.pool_name or '', self.name or ''],
        )

    def destroy(self) -> CommandResult:
        """Destroy filesystem."""
        return self.run_command(
            subcommand='filesystem',
            action='destroy',
            positional_args=[self.pool_name or '', self.name or ''],
        )

    def rename(self, new_name: str) -> CommandResult:
        """Rename filesystem."""
        result = self.run_command(
            subcommand='filesystem',
            action='rename',
            positional_args=[self.pool_name or '', self.name or '', new_name],
        )
        if result.succeeded:
            self.name = new_name
        return result

    def snapshot(self, snapshot_name: str) -> StratisFilesystem | None:
        """Create a copy-on-write snapshot of this filesystem."""
        result = self.run_command(
            subcommand='filesystem',
            action='snapshot',
            positional_args=[self.pool_name or '', self.name or '', snapshot_name],
        )
        if result.failed:
            return None

        return StratisFilesystem(
            name=snapshot_name,
            pool_name=self.pool_name,
            size=self.size,
            origin=self.name,
        )

    def set_size_limit(self, limit: str) -> CommandResult:
        """Set filesystem size limit (e.g. "20G")."""
        result = self.run_command(
            subcommand='filesystem',
            action='set-size-limit',
            positional_args=[self.pool_name or '', self.name or '', limit],
        )
        if result.succeeded:
            self.update_from_managed_objects()
        return result

    def unset_size_limit(self) -> CommandResult:
        """Remove filesystem size limit, allowing growth up to pool capacity."""
        result = self.run_command(
            subcommand='filesystem',
            action='unset-size-limit',
            positional_args=[self.pool_name or '', self.name or ''],
        )
        if result.succeeded:
            self.size_limit = None
        return result

    def schedule_revert(self) -> CommandResult:
        """Schedule a snapshot revert.

        Sets a flag so that when the pool is next started, this snapshot
        will overwrite its origin filesystem. This filesystem must be a
        snapshot. A snapshot scheduled for revert cannot be destroyed;
        the scheduled revert must be cancelled first.
        """
        return self.run_command(
            subcommand='filesystem',
            action='schedule-revert',
            positional_args=[self.pool_name or '', self.name or ''],
        )

    def cancel_revert(self) -> CommandResult:
        """Cancel a previously scheduled snapshot revert."""
        return self.run_command(
            subcommand='filesystem',
            action='cancel-revert',
            positional_args=[self.pool_name or '', self.name or ''],
        )

    @classmethod
    def from_report(cls, report: FilesystemReport, pool_name: str) -> StratisFilesystem | None:
        """Create filesystem from report.

        Size fields (size, used, size_limit) are not populated;
        call ``update_from_managed_objects()`` separately if needed.
        """
        if not report.name:
            return None

        return cls(
            name=report.name,
            pool_name=pool_name,
            uuid=report.uuid,
            origin=_stratis_value(report.origin),
        )

    @classmethod
    def get_all(cls, pool_name: str | None = None) -> list[StratisFilesystem]:
        """Get all Stratis filesystems, optionally filtered by pool name."""
        filesystems: list[StratisFilesystem] = []
        base = cls()

        result = base.run_command('report')
        if result.failed or not result.stdout:
            return filesystems

        try:
            report = json.loads(result.stdout)
            for pool_data in report['pools']:
                current_pool = pool_data.get('name')
                if not current_pool:
                    logger.warning('Pool missing name')
                    continue
                if pool_name and pool_name != current_pool:
                    continue
                # Size/used/limit not populated here — call update_from_managed_objects() if needed
                filesystems.extend(
                    [
                        fs
                        for fs_data in pool_data.get('filesystems', [])
                        if (fs := cls.from_report(FilesystemReport.model_validate(fs_data), current_pool))
                    ]
                )
        except (KeyError, ValueError) as e:
            logger.warning(f'Failed to parse report: {e}')

        return filesystems

cancel_revert()

Cancel a previously scheduled snapshot revert.

Source code in sts_libs/src/sts/stratis/filesystem.py 
273
274
275
276
277
278
279
def cancel_revert(self) -> CommandResult:
    """Cancel a previously scheduled snapshot revert."""
    return self.run_command(
        subcommand='filesystem',
        action='cancel-revert',
        positional_args=[self.pool_name or '', self.name or ''],
    )

create(size=None, size_limit=None)

Create filesystem.

Parameters:

Name Type Description Default
size str | None

Initial size (e.g. "10G")

 None
size_limit str | None

Size limit (e.g. "20G")

 None
Source code in sts_libs/src/sts/stratis/filesystem.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def create(self, size: str | None = None, size_limit: str | None = None) -> CommandResult:
    """Create filesystem.

    Args:
        size: Initial size (e.g. "10G")
        size_limit: Size limit (e.g. "20G")
    """
    options: StratisOptions = {}
    if size:
        options['--size'] = size
    if size_limit:
        options['--size-limit'] = size_limit

    return self.run_command(
        subcommand='filesystem',
        action='create',
        options=options,
        positional_args=[self.pool_name or '', self.name or ''],
    )

destroy()

Destroy filesystem.

Source code in sts_libs/src/sts/stratis/filesystem.py 
201
202
203
204
205
206
207
def destroy(self) -> CommandResult:
    """Destroy filesystem."""
    return self.run_command(
        subcommand='filesystem',
        action='destroy',
        positional_args=[self.pool_name or '', self.name or ''],
    )

from_report(report, pool_name) classmethod

Create filesystem from report.

Size fields (size, used, size_limit) are not populated; call update_from_managed_objects() separately if needed.

Source code in sts_libs/src/sts/stratis/filesystem.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
@classmethod
def from_report(cls, report: FilesystemReport, pool_name: str) -> StratisFilesystem | None:
    """Create filesystem from report.

    Size fields (size, used, size_limit) are not populated;
    call ``update_from_managed_objects()`` separately if needed.
    """
    if not report.name:
        return None

    return cls(
        name=report.name,
        pool_name=pool_name,
        uuid=report.uuid,
        origin=_stratis_value(report.origin),
    )

get_all(pool_name=None) classmethod

Get all Stratis filesystems, optionally filtered by pool name.

Source code in sts_libs/src/sts/stratis/filesystem.py 
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
@classmethod
def get_all(cls, pool_name: str | None = None) -> list[StratisFilesystem]:
    """Get all Stratis filesystems, optionally filtered by pool name."""
    filesystems: list[StratisFilesystem] = []
    base = cls()

    result = base.run_command('report')
    if result.failed or not result.stdout:
        return filesystems

    try:
        report = json.loads(result.stdout)
        for pool_data in report['pools']:
            current_pool = pool_data.get('name')
            if not current_pool:
                logger.warning('Pool missing name')
                continue
            if pool_name and pool_name != current_pool:
                continue
            # Size/used/limit not populated here — call update_from_managed_objects() if needed
            filesystems.extend(
                [
                    fs
                    for fs_data in pool_data.get('filesystems', [])
                    if (fs := cls.from_report(FilesystemReport.model_validate(fs_data), current_pool))
                ]
            )
    except (KeyError, ValueError) as e:
        logger.warning(f'Failed to parse report: {e}')

    return filesystems

get_fs_uuid()

Get filesystem UUID from the stratis report.

Source code in sts_libs/src/sts/stratis/filesystem.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def get_fs_uuid(self) -> str | None:
    """Get filesystem UUID from the stratis report."""
    result = self.run_command('report')
    if result.failed or not result.stdout:
        return None

    try:
        report = json.loads(result.stdout)
        for pool in report['pools']:
            if self.pool_name != pool['name']:
                continue
            for fs in pool['filesystems']:
                if self.name != fs['name']:
                    continue
                return fs['uuid']
    except (KeyError, ValueError) as e:
        logger.warning(f'Failed to get filesystem UUID: {e}')

    return None

rename(new_name)

Rename filesystem.

Source code in sts_libs/src/sts/stratis/filesystem.py 
209
210
211
212
213
214
215
216
217
218
def rename(self, new_name: str) -> CommandResult:
    """Rename filesystem."""
    result = self.run_command(
        subcommand='filesystem',
        action='rename',
        positional_args=[self.pool_name or '', self.name or '', new_name],
    )
    if result.succeeded:
        self.name = new_name
    return result

schedule_revert()

Schedule a snapshot revert.

Sets a flag so that when the pool is next started, this snapshot will overwrite its origin filesystem. This filesystem must be a snapshot. A snapshot scheduled for revert cannot be destroyed; the scheduled revert must be cancelled first.

Source code in sts_libs/src/sts/stratis/filesystem.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
def schedule_revert(self) -> CommandResult:
    """Schedule a snapshot revert.

    Sets a flag so that when the pool is next started, this snapshot
    will overwrite its origin filesystem. This filesystem must be a
    snapshot. A snapshot scheduled for revert cannot be destroyed;
    the scheduled revert must be cancelled first.
    """
    return self.run_command(
        subcommand='filesystem',
        action='schedule-revert',
        positional_args=[self.pool_name or '', self.name or ''],
    )

set_size_limit(limit)

Set filesystem size limit (e.g. "20G").

Source code in sts_libs/src/sts/stratis/filesystem.py 
237
238
239
240
241
242
243
244
245
246
def set_size_limit(self, limit: str) -> CommandResult:
    """Set filesystem size limit (e.g. "20G")."""
    result = self.run_command(
        subcommand='filesystem',
        action='set-size-limit',
        positional_args=[self.pool_name or '', self.name or '', limit],
    )
    if result.succeeded:
        self.update_from_managed_objects()
    return result

snapshot(snapshot_name)

Create a copy-on-write snapshot of this filesystem.

Source code in sts_libs/src/sts/stratis/filesystem.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def snapshot(self, snapshot_name: str) -> StratisFilesystem | None:
    """Create a copy-on-write snapshot of this filesystem."""
    result = self.run_command(
        subcommand='filesystem',
        action='snapshot',
        positional_args=[self.pool_name or '', self.name or '', snapshot_name],
    )
    if result.failed:
        return None

    return StratisFilesystem(
        name=snapshot_name,
        pool_name=self.pool_name,
        size=self.size,
        origin=self.name,
    )

unset_size_limit()

Remove filesystem size limit, allowing growth up to pool capacity.

Source code in sts_libs/src/sts/stratis/filesystem.py 
248
249
250
251
252
253
254
255
256
257
def unset_size_limit(self) -> CommandResult:
    """Remove filesystem size limit, allowing growth up to pool capacity."""
    result = self.run_command(
        subcommand='filesystem',
        action='unset-size-limit',
        positional_args=[self.pool_name or '', self.name or ''],
    )
    if result.succeeded:
        self.size_limit = None
    return result

update_from_managed_objects()

Update size/used/limit from stratis report managed_objects_report.

Uses the managed-objects report which provides clean byte values directly, avoiding sector/unit parsing issues in the basic report.

Source code in sts_libs/src/sts/stratis/filesystem.py 
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
def update_from_managed_objects(self) -> bool:
    """Update size/used/limit from ``stratis report managed_objects_report``.

    Uses the managed-objects report which provides clean byte values
    directly, avoiding sector/unit parsing issues in the basic report.
    """
    if not self.name:
        return False

    # Get filesystem object path
    result = self.run_command(
        subcommand='filesystem',
        action=['debug', 'get-object-path'],
        positional_args=['--name', self.name],
    )
    if result.failed or not result.stdout:
        logger.debug(f'Failed to get object path for filesystem {self.name}')
        return False

    object_path = result.stdout.strip()

    # Get managed objects report
    result = self.run_command(subcommand='report', action='managed_objects_report')
    if result.failed or not result.stdout:
        logger.error('Failed to get managed objects report')
        return False

    try:
        report_data = json.loads(result.stdout)

        if object_path not in report_data:
            logger.error(f'Object path {object_path} not found in managed objects report')
            return False

        fs_interfaces = report_data[object_path]

        # Highest D-Bus interface revision has the most complete properties (see stratis.base patterns)
        last_revision = list(fs_interfaces.keys())[-1]
        fs_data = fs_interfaces[last_revision]

        # Size is a plain bytes string
        if 'Size' in fs_data:
            self.size = int(fs_data['Size'])

        used_raw = dbus_maybe_value(fs_data, 'Used')
        if used_raw is not None:
            self.used = int(used_raw)

        limit_raw = dbus_maybe_value(fs_data, 'SizeLimit')
        if limit_raw is not None:
            self.size_limit = int(limit_raw)

    except (json.JSONDecodeError, KeyError, ValueError):
        logger.exception('Error parsing managed objects data')
        return False
    else:
        return True

Error Handling

sts.stratis.errors

Stratis-related errors.

StratisBlockdevError

Bases: StratisError

Error in a Stratis blockdev operation.

Source code in sts_libs/src/sts/stratis/errors.py 
23
24
class StratisBlockdevError(StratisError):
    """Error in a Stratis blockdev operation."""

StratisError

Bases: STSError

Base class for Stratis-related errors.

Source code in sts_libs/src/sts/stratis/errors.py 
11
12
class StratisError(STSError):
    """Base class for Stratis-related errors."""

StratisFilesystemError

Bases: StratisError

Error in a Stratis filesystem operation.

Source code in sts_libs/src/sts/stratis/errors.py 
19
20
class StratisFilesystemError(StratisError):
    """Error in a Stratis filesystem operation."""

StratisPoolError

Bases: StratisError

Error in a Stratis pool operation.

Source code in sts_libs/src/sts/stratis/errors.py 
15
16
class StratisPoolError(StratisError):
    """Error in a Stratis pool operation."""