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, 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:

  • mapping (TypeVar(ElementMappingType, bound= Optional[Mapping[Hashable, momotor.bundles.elements.base.Element]])) – A mapping containing items to test

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

  • expected_value_type (Type[Element]) – Expected type for the values

  • bundle (Bundle) – expected bundle instance for each element

Return type:

TypeVar(ElementMappingType, bound= Optional[Mapping[Hashable, momotor.bundles.elements.base.Element]])

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

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

Parameters:

  • elements (TypeVar(ElementsType, bound= Optional[Iterable[momotor.bundles.elements.base.Element]])) – sequence of Element objects

  • bundle (Bundle) – expected bundle instance for each element

Return type:

TypeVar(ElementsType, bound= Optional[Iterable[momotor.bundles.elements.base.Element]])

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

Combines assert_elements_bundle() and assert_sequence_instanceof()

Parameters:

  • elements (TypeVar(ElementsType, bound= Optional[Iterable[momotor.bundles.elements.base.Element]])) – sequence of Element objects

  • expected_type (Type[Element]) – The expected type of each element

  • bundle (Bundle) – expected bundle instance for each element

Return type:

TypeVar(ElementsType, bound= Optional[Iterable[momotor.bundles.elements.base.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 _All

NONE

alias of _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 _Any

NONE

alias of _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 _All

NONE

alias of _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 _All

NONE

alias of _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[Path]
zip_file: Optional[ZipFile]

momotor.bundles.elements.content

exception momotor.bundles.elements.content.AttachmentContent
exception momotor.bundles.elements.content.NoContent
class momotor.bundles.elements.content.ContentAttachmentElement(bundle)

A ContentElement variant exposing the type and src properties

Parameters:

bundle (Bundle) – The Bundle containing this element

file_size(path=None)

Get file size for the content.

Return type:

Optional[int]

Returns:

The file size

has_writable_content()
is_dir(path=None)
Return type:

bool

open(path=None)

Open the attachment file for reading. Handles opening files directly from filesystem and from zipped bundles

Parameters:

path (Union[str, PurePosixPath, None]) – for directory attachments, path selects a file in that directory

Return type:

BinaryIO

Returns:

the opened file

Raises:

  • FileNotFoundError – if the element has no attachment (when src is None), or when it’s a directory and path does not exist in that directory

  • IsADirectoryError – when it’s a directory

read(path=None)

Read the contents of the attachment

Parameters:

path (Union[str, PurePosixPath, None]) – for directory attachments, path selects a file in that directory

Return type:

bytes

Returns:

A bytes object with the full file contents

property type: Optional[str]

The type attribute.

Return type:

Optional[str]

class momotor.bundles.elements.content.ContentBasicElement(bundle)
HAS_ENCODING: ClassVar = False
VALID_PROCESSED_TYPES: ClassVar = (<class 'str'>, <class 'bool'>)
class momotor.bundles.elements.content.ContentFullElement(bundle)

A ContentElement variant exposing the type property

Parameters:

bundle (Bundle) – The Bundle containing this element

HAS_ENCODING: ClassVar = True
VALID_PROCESSED_TYPES: ClassVar = (<class 'str'>, <class 'bool'>, <class 'bytes'>, <class 'int'>, <class 'float'>, <class 'decimal.Decimal'>)
property encoding: Optional[str]

encoding attribute: read-only, the encoding is automatically determined from the value

Return type:

Optional[str]

property type: Optional[str]

The type attribute. Indicates the type of the value attribute: string, integer or float

Return type:

Optional[str]

momotor.bundles.elements.content.NO_CONTENT = <object object>

A sentinel value indicating there is no content.

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