onyo.lib.inventory module

class onyo.lib.inventory.Inventory(repo)[source]

Bases: object

__init__(repo)[source]
add_asset(asset)[source]
Return type:

list[InventoryOperation]

add_directory(path)[source]
Return type:

list[InventoryOperation]

asset_paths_available(assets)[source]

Test whether path used by assets are available in the inventory.

Availability not only requires the path to not yet exist, but also the filename to be unique.

Raises:

ValueError – if any of the paths can not be used for a new asset

Return type:

None

commit(message)[source]

Execute and git-commit pending operations.

Return type:

None

diff()[source]
Return type:

Generator[str, None, None]

generate_asset_name(asset)[source]
Return type:

str

get_asset(path)[source]
Return type:

UserDict

get_asset_from_template(template)[source]
Return type:

DotNotationWrapper

get_assets(include=None, exclude=None, depth=0)[source]

Yield all assets under paths up to depth directory levels.

Generator, because it needs to read file content. This allows to act upon results while they are coming in.

Parameters:
  • include (Optional[Iterable[Path]] (default: None)) – Paths to look for assets under. Defaults to the root of the inventory.

  • exclude (Union[Iterable[Path], Path, None] (default: None)) – Paths to exclude, meaning that assets underneath any of these are not being returned. Defaults to None.

  • depth (int (default: 0)) – Number of levels to descend into. Must be greater equal 0. If 0, descend recursively without limit. Defaults to 0.

Returns:

All matching assets in the inventory.

Return type:

Generator of dict

get_assets_by_query(include=None, exclude=None, depth=0, match=None)[source]

Get assets matching paths and filters.

Convenience to run the builtin filter on all assets retrieved by self.get(paths, depth) for each callable in filters, thus combining the filters by a logical AND.

Parameters:
  • include (list[Path] | None (default: None)) – Paths to look for assets under. Defaults to the root of the inventory. Passed to self.get_assets.

  • exclude (list[Path] | Path | None (default: None)) – Paths to exclude, meaning that assets underneath any of these are not being returned. Defaults to None. Passed to self.get_assets.

  • depth (int | None (default: 0)) – Number of levels to descend into. Must be greater or equal 0. If 0, descend recursively without limit. Defaults to 0. Passed to self.get_assets.

  • match (list[Callable[[dict | UserDict], bool]] | None (default: None)) – Callable suitable for the builtin filter, when called on a list of assets (dictionaries).

Returns:

All assets found underneath paths up to depth levels, for which all filters returned True.

Return type:

Generator of dict

get_faux_serials(length=6, num=1)[source]

Generate a unique faux serial.

Generate a faux serial and verify that it is not used by any other asset in the repository. The length of the faux serial must be 4 or greater.

Returns a set of unique faux serials.

Return type:

set[str]

modify_asset(asset, new_asset)[source]
Return type:

list[InventoryOperation]

move_asset(src, dst)[source]
Return type:

list[InventoryOperation]

move_directory(src, dst)[source]
Return type:

list[InventoryOperation]

operations_pending()[source]

Returns whether there’s something to commit.

Return type:

bool

operations_summary()[source]
Return type:

str

raise_empty_keys(asset)[source]

Whether asset has empty keys.

Validation helper

Return type:

None

raise_required_key_empty_value(asset)[source]

Whether asset has an empty value for a required key.

Validation helper.

Return type:

None

Notes

This is currently considering asset name keys only. However, proper asset validation with ways to declare other keys required is anticipated. This would need to account for those as well.

remove_asset(asset)[source]
Return type:

list[InventoryOperation]

remove_directory(directory, recursive=True)[source]
Return type:

list[InventoryOperation]

rename_asset(asset, name=None)[source]
Return type:

list[InventoryOperation]

rename_directory(src, dst)[source]
Return type:

list[InventoryOperation]

reset()[source]

Discard pending operations.

Return type:

None

property root

Path to root inventory directory.

class onyo.lib.inventory.InventoryOperation(operator, operands, repo)[source]

Bases: object

__init__(operator, operands, repo)
diff()[source]
Return type:

Generator[str, None, None]

execute()[source]
Return type:

tuple[list[Path], list[Path]]

operands: tuple
operator: InventoryOperator
repo: OnyoRepo
class onyo.lib.inventory.InventoryOperator(executor, differ, recorder)[source]

Bases: object

__init__(executor, differ, recorder)
differ: Callable
executor: Callable
recorder: Callable