Utility classes and functions

momotor.bundles.utils.arguments

class momotor.bundles.utils.arguments.BundleFactoryArguments(xml_name=None, use_lxml=None, validate_xml=None, location_base=None, validate_signature=True, legacy=True)
legacy: bool = True

accept XML produced with legacy generator (the momotor-common package)

location_base: Union[str, pathlib.Path] = None

location used in error messages

use_lxml: bool = None

force (True) or prevent (False) use of lxml library. If None, auto-detects lxml availability

validate_signature: bool = True

validate <file> node signatures

validate_xml: bool = None

enable (True) or disable (False) XML validation. If None, uses default setting, which is to enable validation

xml_name: str = None

XML file name if not the default (for directory and zip paths)

class momotor.bundles.utils.arguments.DirectoryConstructionArguments(use_lxml=None, pretty_xml=False, encoding='utf-8', xml_name=None, sign_files=True, hashers=<factory>, legacy=True, optimize=True, dir_mode=448)
dir_mode: int = 448

the mode bits (defaults to 0o700, making the directory only accessible to the current user)

encoding: str = 'utf-8'

encoding of the XML document

hashers: Sequence[str]

hashing algorithms to use when sign_files is set to True. Default ['sha1']

legacy: bool = True

write XML compatible with legacy parser (the momotor-common package)

optimize: bool = True

create optimized bundle

pretty_xml: bool = False

generate a better readable XML document

sign_files: bool = True

sign the <file> nodes. This calculates a hash of the attachments and adds it as an attribute in the XML

use_lxml: Optional[bool] = None

force (True) or prevent (False) use of lxml library. If None, auto-detects lxml availability

xml_name: str = None

name of the xml document. If not provided uses the default name for the bundle (eg. recipe.xml, config.xml, product.xml or result.xml)

class momotor.bundles.utils.arguments.FileConstructionArguments(use_lxml=None, pretty_xml=False, encoding='utf-8', xml_name=None, sign_files=True, hashers=<factory>, legacy=True, optimize=True, zip=False, compression=8, compresslevel=None)
compression: int = 8

compression mode, see zipfile.ZipFile for possible values. (Defaults to ZIP_DEFLATED, only used when generating a zipped bundle)

compresslevel: int = None

compression level, see zipfile.ZipFile for possible values. (Only used when generating a zipped bundle)

encoding: str = 'utf-8'

encoding of the XML document

hashers: Sequence[str]

hashing algorithms to use when sign_files is set to True. Default ['sha1']

legacy: bool = True

write XML compatible with legacy parser (the momotor-common package)

optimize: bool = True

create optimized bundle

pretty_xml: bool = False

generate a better readable XML document

sign_files: bool = True

sign the <file> nodes. This calculates a hash of the attachments and adds it as an attribute in the XML

use_lxml: Optional[bool] = None

force (True) or prevent (False) use of lxml library. If None, auto-detects lxml availability

xml_name: str = None

name of the xml document. If not provided uses the default name for the bundle (eg. recipe.xml, config.xml, product.xml or result.xml)

zip: bool = False

force zip format output

momotor.bundles.utils.assertion

momotor.bundles.utils.assertion.assert_element_mapping_instanceof(mapping, expected_key_type, expected_value_type, bundle)

Combines assert_elements_bundle() and assert_mapping_instanceof()

Parameters
Return type

TypeVar(ElementMappingType, bound= Optional[Mapping[Hashable, Element]])

momotor.bundles.utils.assertion.assert_elements_bundle(elements, bundle)

Assert that all elements are linked the correct bundle. Returns elements

Parameters
Return type

TypeVar(ElementsType, bound= Optional[Iterable[Element]])

momotor.bundles.utils.assertion.assert_elements_instanceof(elements, expected_type, bundle)

Combines assert_elements_bundle() and assert_sequence_instanceof()

Parameters
Return type

TypeVar(ElementsType, bound= Optional[Iterable[Element]])

momotor.bundles.utils.assertion.assert_mapping_instanceof(mapping, expected_key_type, expected_value_type)

Assert that all items in the mapping have the expected key and value types. Returns mapping

Parameters
  • mapping (TypeVar(MappingType, bound= Optional[Mapping])) – A mapping containing items to test

  • expected_key_type (Type[Hashable]) – Expected type for the keys

  • expected_value_type (Type) – Expected type for the values

Return type

TypeVar(MappingType, bound= Optional[Mapping])

momotor.bundles.utils.assertion.assert_sequence_instanceof(items, expected_type)

Assert that all items are of the expected type. Returns items

Parameters
  • items (TypeVar(IterableType, bound= Optional[Iterable])) – The items to test

  • expected_type (Type) – The expected type of each item

Return type

TypeVar(IterableType, bound= Optional[Iterable])

momotor.bundles.utils.boolean

momotor.bundles.utils.boolean.to_bool(val)

Convert a representation of truth to True or False.

True values are ‘y’, ‘yes’, ‘t’, ‘true’, ‘on’, and ‘1’; false values are ‘n’, ‘no’, ‘f’, ‘false’, ‘off’, and ‘0’. Raises ValueError if ‘val’ is anything else.

>>> to_bool(True)
True
>>> to_bool('y')
True
>>> to_bool('yes')
True
>>> to_bool('t')
True
>>> to_bool('true')
True
>>> to_bool('on')
True
>>> to_bool('1')
True
>>> to_bool(1)
True
>>> to_bool(False)
False
>>> to_bool('n')
False
>>> to_bool('no')
False
>>> to_bool('f')
False
>>> to_bool('false')
False
>>> to_bool('off')
False
>>> to_bool('0')
False
>>> to_bool(0)
False
>>> to_bool(None)
False
>>> to_bool('other')
Traceback (most recent call last):
...
ValueError: invalid truth value 'other'
Return type

bool

momotor.bundles.utils.domain

momotor.bundles.utils.domain.merge_domains(*domains)

Merge domains of format <domain>#<subdomain>.

For all domains, the first provided part is returned as part of the merged domain:

abc#def | uvw#xyz = abc#def
abc#def | uvw     = abc#def
abc#def |    #xyz = abc#def
abc     | uvw#xyz = abc#xyz
abc     | uvw     = abc
abc     |    #xyz = abc#xyz
   #def | uvw#xyz = uwv#def
   #def | uvw     = uwv#def
   #def |    #xyz =    #def
>>> merge_domains() is None
True
>>> merge_domains(None) is None
True
>>> merge_domains('abc#def', 'uvw')
'abc#def'
>>> merge_domains('abc#def', '#xyz')
'abc#def'
>>> merge_domains('abc#def', 'uvw#xyz')
'abc#def'
>>> merge_domains('abc', 'uvw#xyz')
'abc#xyz'
>>> merge_domains('abc', 'uvw')
'abc'
>>> merge_domains('abc', '#xyz')
'abc#xyz'
>>> merge_domains('#def', 'uvw#xyz')
'uvw#def'
>>> merge_domains('#def', 'uvw')
'uvw#def'
>>> merge_domains('#def', '#xyz')
'#def'
Return type

Optional[str]

momotor.bundles.utils.domain.split_domain(domain)

Split a domain name <domain>[#<subdomain>] into its parts.

>>> split_domain(None)
(None, None)
>>> split_domain('')
(None, None)
>>> split_domain('#')
(None, None)
>>> split_domain('domain')
('domain', None)
>>> split_domain('domain#')
('domain', None)
>>> split_domain('#subdomain')
(None, 'subdomain')
>>> split_domain('domain#subdomain')
('domain', 'subdomain')
Parameters

domain (Optional[str]) – The domain

Return type

Tuple[Optional[str], Optional[str]]

Returns

a two-tuple with the parts

momotor.bundles.utils.domain.unsplit_domain(domain, subdomain)

Merge <domain> and <subdomain> parts into a <domain>#<subdomain> string, handling None values

The reverse operation of split_domain().

>>> unsplit_domain(None, None) is None
True
>>> unsplit_domain('domain', None)
'domain'
>>> unsplit_domain('domain', '')
'domain'
>>> unsplit_domain(None, 'subdomain')
'#subdomain'
>>> unsplit_domain('domain', 'subdomain')
'domain#subdomain'
Param

domain: The domain part

Param

subdomain: The subdomain part

Return type

Optional[str]

Returns

the joined domain

momotor.bundles.utils.encoding

Data encoding and decoding

momotor.bundles.utils.encoding.ENCODINGS

List of all supported encodings

momotor.bundles.utils.encoding.decode_data(data, encoding)

Decode encoded data.

Parameters
  • data (str) – Data to decode

  • encoding (str) – The encoding used on the data: quopri or base64

Return type

bytes

Returns

The decoded data

Raises

ValueError – For unsupported encodings

momotor.bundles.utils.encoding.encode_data(data)

Encode provided data.

Printable data is encoded using Quoted-printable encoding, any other data is encoded using Base64 encoding.

Parameters

data (str) – Data to encode

Return type

tuple

Returns

tuple(encoded, encoding)

  • encoded: The encoded data

  • encoding (str): The encoding used. quopri or base64

momotor.bundles.utils.encoding.encode_posix_path(path)

Encode a path using idna encoding to ensure it will only contain ascii characters

Return type

PurePosixPath

momotor.bundles.utils.encoding.is_printable(s, low=32, allowed=None)

Detect if string is printable; it should not contain any control characters, except those in allowed, and less than 5% of the characters are allowed to be high-ascii

momotor.bundles.utils.encoding.make_pure_ascii(s)
Return type

str

momotor.bundles.utils.encoding.quopri_decode(data)

Decode quoted printable data

momotor.bundles.utils.encoding.quopri_encode(data)

Encode data using quoted printable method Uses a different algorithm than quopri.encodestring():

  • Encodes first space character on a line, but not the rest

  • Encodes newline, return characters and high-ascii

momotor.bundles.utils.filters

See the filters documentation.

class momotor.bundles.utils.filters.All(*filters, **filter_args)

“All” filter operator

All the lookups in the argument list will need to match for an item to match this filter, see the All object documentation for an example.

(alias of F)

COMBINER

alias of momotor.bundles.utils.filters._All

NONE

alias of momotor.bundles.utils.filters._Always

class momotor.bundles.utils.filters.Any(*filters, **filter_args)

“Any” filter operator

Any of the lookups in the argument list can match for an item to match this filter, see the Any object documentation for an example.

COMBINER

alias of momotor.bundles.utils.filters._Any

NONE

alias of momotor.bundles.utils.filters._Never

class momotor.bundles.utils.filters.F(*filters, **filter_args)

Filter operator

Base for the other filter operators. Can be used to prevent multiple keywords SyntaxErrors in Python, see the F object documentation for an example.

COMBINER

alias of momotor.bundles.utils.filters._All

NONE

alias of momotor.bundles.utils.filters._Always

class momotor.bundles.utils.filters.FilterableList(iterable=(), /)

A list with additional functions to filter the objects in the list

See Sequence filtering section in the documentation

append(object, /)

Append object to the end of the list.

clear()

Remove all items from list.

copy()

Return a shallow copy of the list.

count(value, /)

Return number of occurrences of value.

exclude(*args, **kwargs)

Returns a filtered version of self that only includes elements that do not match the provided query

See Filtering and excluding

Return type

TypeVar(TS, bound= Iterable)

extend(iterable, /)

Extend list by appending elements from the iterable.

filter(*args, **kwargs)

Returns a filtered version of self that only includes elements that match the provided query

See Filtering and excluding

Return type

TypeVar(TS, bound= Iterable)

filter_with(func)

Filter the items using a callable function

Parameters

func (Callable[[TypeVar(TO, bound= object)], bool]) – A callable receiving an item and returning a boolean

Return type

TypeVar(TS, bound= Iterable)

iexclude(*args, **kwargs)

Iterator version of exclude()

Return type

Iterable[TypeVar(TO, bound= object)]

ifilter(*args, **kwargs)

Iterator version of filter()

Return type

Iterable[TypeVar(TO, bound= object)]

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

insert(index, object, /)

Insert object before index.

pop(index=- 1, /)

Remove and return item at index (default last).

Raises IndexError if list is empty or index is out of range.

remove(value, /)

Remove first occurrence of value.

Raises ValueError if the value is not present.

reverse()

Reverse IN PLACE.

sort(*, key=None, reverse=False)

Sort the list in ascending order and return None.

The sort is in-place (i.e. the list itself is modified) and stable (i.e. the order of two equal elements is maintained).

If a key function is given, apply it once to each list item and sort them, ascending or descending, according to their function values.

The reverse flag can be set to sort in descending order.

class momotor.bundles.utils.filters.FilterableTuple(iterable=(), /)

A tuple with additional functions to filter the objects in the tuple

See Sequence filtering section in the documentation

count(value, /)

Return number of occurrences of value.

exclude(*args, **kwargs)

Returns a filtered version of self that only includes elements that do not match the provided query

See Filtering and excluding

Return type

TypeVar(TS, bound= Iterable)

filter(*args, **kwargs)

Returns a filtered version of self that only includes elements that match the provided query

See Filtering and excluding

Return type

TypeVar(TS, bound= Iterable)

filter_with(func)

Filter the items using a callable function

Parameters

func (Callable[[TypeVar(TO, bound= object)], bool]) – A callable receiving an item and returning a boolean

Return type

TypeVar(TS, bound= Iterable)

iexclude(*args, **kwargs)

Iterator version of exclude()

Return type

Iterable[TypeVar(TO, bound= object)]

ifilter(*args, **kwargs)

Iterator version of filter()

Return type

Iterable[TypeVar(TO, bound= object)]

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

class momotor.bundles.utils.filters.Not(*filters, **filter_args)

“Not” filter operator

Any item not matching all the lookup arguments will match this filter, see the Not object documentation for an example.

COMBINER

alias of momotor.bundles.utils.filters._All

NONE

alias of momotor.bundles.utils.filters._Always

momotor.bundles.utils.grouping

momotor.bundles.utils.grouping.group_by_attr(items, *attrs)

Group a list of elements by the value of one or more attributes

Parameters
  • items (Iterable[TypeVar(T, bound= Element)]) – list of elements to group

  • attrs (str) – name(s) of the attribute to group on

Return type

Union[Dict[str, FilterableTuple[TypeVar(T, bound= Element)]], Dict[Tuple[str, str], FilterableTuple[TypeVar(T, bound= Element)]]]

Returns

a dictionary of lists of elements

momotor.bundles.utils.immutable

class momotor.bundles.utils.immutable.ImmutableDict(*args, **kwargs)

An immutable dict

clear()
Return type

NoReturn

move_to_end(key, last=None)
Return type

NoReturn

pop(key)

If key is not found, d is returned if given, otherwise KeyError is raised

Return type

NoReturn

popitem(last=True)

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

Return type

NoReturn

setdefault(key, default=None)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

Return type

NoReturn

update(*args, **kwargs)

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

Return type

NoReturn

class momotor.bundles.utils.immutable.ImmutableOrderedDict(*args, **kwargs)

An immutable OrderedDict

clear()
Return type

NoReturn

pop(key)

value. If key is not found, d is returned if given, otherwise KeyError is raised.

Return type

NoReturn

popitem(last=True)

Remove and return a (key, value) pair from the dictionary.

Pairs are returned in LIFO order if last is true or FIFO order if false.

Return type

NoReturn

setdefault(key, default=None)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

Return type

NoReturn

update(*args, **kwargs)

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

Return type

NoReturn

momotor.bundles.utils.keyedtuple

class momotor.bundles.utils.keyedtuple.KeyedTuple(items=None, *, key_attr=None)

A tuple of objects with an indexable attribute that acts as both an immutable sequence and a mapping. Elements can be accessed by their numeric index or the key attribute.

Parameters

If items is another KeyedTuple, key_attr must either be the same as the key_attr of items, or not provided.

copy()

Create a shallow copy

Return type

TypeVar(TT)

count(item)
Return type

int

get(key_or_index, default=None)

Get an item from the sequence by index, key or item

Parameters
  • key_or_index (Union[str, int, TypeVar(IT)]) – index, key or item

  • default – the value to return if item does not exist

Returns

the item or default

index(value[, start[, stop]]) integer -- return first index of value.

Raises ValueError if the value is not present.

Supporting start and stop arguments is optional, but recommended.

items()

A set-like object providing a view on the items

Return type

ItemsView[str, TypeVar(IT)]

keys()

A set-like object providing a view on the keys

Return type

KeysView[str]

values()

A set-like object providing a view on the values

Return type

ValuesView[TypeVar(IT)]

momotor.bundles.utils.lxml

momotor.bundles.utils.lxml.detect_lxml()

Helper to detect lxml package availability

Return type

bool

Returns

True if lxml is installed

momotor.bundles.utils.lxml.use_lxml(use)

Helper to interpret use_lxml argument

Param

Provided use_lxml argument

Return type

bool

Returns

True if lxml should be used

momotor.bundles.utils.nodes

momotor.bundles.utils.nodes.get_nested_complex_nodes(node, node_name, *sub_node_names)

Get all the nodes in a nested complex node structure.

For example, with the following XML:

<root>
  <level1>
    <level2>
    <level2>
    <level2>
  </level1>
  <level1>
    <level2>
  </level1>
</root>

a call get_nested_complex_list(root_node, 'level1', 'level2') yields a sequence of

[('level1', level1-node),
 ('level2', level2-node),
 ('level2', level2-node),
 ('level2', level2-node),
 ('level1', level1-node),
 ('level2', level2-node)
]
Parameters
  • node (object) – root node containing ‘node_name’ nodes

  • node_name (str) – name of first level nodes to retrieve

  • sub_node_names (str) – subsequent names of nodes to recursively retrieve

Return type

Generator[Tuple[str, object], None, None]

Returns

a generator yielding matching nodes

momotor.bundles.utils.text

Useful functions for string processing, borrowed from Django (with a few small modifications)

momotor.bundles.utils.text.smart_split(text)

Generator that splits a string by spaces, leaving quoted phrases together. Supports both single and double quotes, and supports escaping quotes with backslashes. In the output, strings will keep their initial and trailing quote marks and escaped quotes will remain escaped (the results can then be further processed with unescape_string_literal()).

>>> list(smart_split(r'This is "a person\'s" test.'))
['This', 'is', '"a person\\\'s"', 'test.']
>>> list(smart_split(r"Another 'person\'s' test."))
['Another', "'person\\'s'", 'test.']
>>> list(smart_split(r'A "\"funky\" style" test.'))
['A', '"\\"funky\\" style"', 'test.']
Return type

Generator[str, None, None]

momotor.bundles.utils.text.unescape_string_literal(s)

Convert quoted string literals to unquoted strings with escaped quotes and backslashes unquoted:

>>> unescape_string_literal('"abc"')
'abc'
>>> unescape_string_literal("'abc'")
'abc'
>>> unescape_string_literal('"a \"bc\""')
'a "bc"'
>>> unescape_string_literal("'\'ab\' c'")
"'ab' c"
Return type

str

momotor.bundles.utils.zipwrapper

class momotor.bundles.utils.zipwrapper.ZipWrapper(*, path=None, content=None)

A wrapper around a ZipFile. The zip file can be either located in the filesystem or in memory.

Example usage:

zip_wrapper = ZipWrapper('test.zip')
with zip_wrapper as zip_file, zip_file.open() as f:
    # f is now an open zipfile.ZipFile object
Parameters
close()

Close the wrapped zip file

content: Optional[Union[bytes, memoryview]]
path: Optional[pathlib.Path]
zip_file: Optional[zipfile.ZipFile]

momotor.bundles.elements.refs

momotor.bundles.elements.refs.resolve_ref(tag_name, node, groups)

Resolve reference

Parameters
Return type

Tuple[Optional[object], TypeVar(CT, bound= Union[FileComplexType, OptionComplexType, CheckletComplexType])]

Returns

tuple (parent, node). parent can be none, in that case, ref attribute was None and provided node is returned. Otherwise, returns the resolved reference. Throws an InvalidRefError exception if reference cannot be resolved