weaver.utils

Module Contents

weaver.utils.LOGGER[source]
class weaver.utils._Singleton[source]

Bases: type

class weaver.utils._NullType[source]

Bases: six.with_metaclass()

Represents a null value to differentiate from None.

weaver.utils.null[source]
weaver.utils.get_weaver_url(container) → AnyStr[source]

Retrieves the home URL of the weaver application.

weaver.utils.get_any_id(info) → Union[AnyStr, None][source]

Retrieves a dictionary id-like key using multiple common variations [id, identifier, _id]. :param info: dictionary that potentially contains an id-like key. :returns: value of the matched id-like key or None if not found.

weaver.utils.get_any_value(info) → AnyValue[source]

Retrieves a dictionary value-like key using multiple common variations [href, value, reference]. :param info: dictionary that potentially contains a value-like key. :returns: value of the matched value-like key or None if not found.

weaver.utils.get_any_message(info) → AnyStr[source]

Retrieves a dictionary ‘value’-like key using multiple common variations [message]. :param info: dictionary that potentially contains a ‘message’-like key. :returns: value of the matched ‘message’-like key or an empty string if not found.

weaver.utils.get_registry(container) → Registry[source]

Retrieves the application registry from various containers referencing to it.

weaver.utils.get_settings(container) → SettingsType[source]

Retrieves the application settings from various containers referencing to it.

weaver.utils.get_header(header_name, header_container) → Union[AnyStr, None][source]

Searches for the specified header by case/dash/underscore-insensitive header_name inside header_container.

Looks for cookie_header_name header within header_container. :returns: new header container in the form {'Cookie': <found_cookie>} if it was matched, or empty otherwise.

weaver.utils.get_url_without_query(url) → AnyStr[source]

Removes the query string part of an URL.

weaver.utils.is_valid_url(url) → bool[source]
weaver.utils.parse_extra_options(option_str)[source]

Parses the extra options parameter.

The option_str is a string with coma separated opt=value pairs. Example:

tempdir=/path/to/tempdir,archive_root=/path/to/archive
Parameters:option_str – A string parameter with the extra options.
Returns:A dict with the parsed extra options.
weaver.utils.fully_qualified_name(obj) → str[source]

Obtains the '<module>.<name>' full path definition of the object to allow finding and importing it.

weaver.utils.now() → datetime[source]
weaver.utils.now_secs() → int[source]

Return the current time in seconds since the Epoch.

weaver.utils.wait_secs(run_step=-1)[source]
weaver.utils.expires_at(hours=1) → int[source]
weaver.utils.localize_datetime(dt, tz_name='UTC') → datetime[source]

Provide a timezone-aware object for a given datetime and timezone name

weaver.utils.get_base_url(url) → AnyStr[source]

Obtains the base URL from the given url.

weaver.utils.path_elements(path) → List[AnyStr][source]
weaver.utils.lxml_strip_ns(tree) → None[source]
weaver.utils.ows_context_href(href, partial=False) → JSON[source]

Returns the complete or partial dictionary defining an OWSContext from a reference.

weaver.utils.pass_http_error(exception, expected_http_error) → None[source]

Given an HTTPError of any type (pyramid, requests), ignores (pass) the exception if the actual error matches the status code. Other exceptions are re-raised.

Parameters:
  • exception – any Exception instance (“object” from a try..except exception as “object” block).
  • expected_http_error – single or list of specific pyramid HTTPError to handle and ignore.
Raises:

exception – if it doesn’t match the status code or is not an HTTPError of any module.

weaver.utils.raise_on_xml_exception(xml_node)[source]

Raises an exception with the description if the XML response document defines an ExceptionReport. :param xml_node: instance of etree.Element :raise Exception: on found ExceptionReport document.

weaver.utils.str2bytes(string) → bytes[source]

Obtains the bytes representation of the string.

weaver.utils.bytes2str(string) → str[source]

Obtains the unicode representation of the string.

weaver.utils.islambda(func) → bool[source]
weaver.utils.first_cap_re[source]
weaver.utils.all_cap_re[source]
weaver.utils.convert_snake_case(name) → AnyStr[source]
weaver.utils.parse_request_query(request) → Dict[AnyStr, Dict[AnyKey, AnyStr]][source]
Parameters:request
Returns:dict of dict where k=v are accessible by d[k][0] == v and q=k=v are accessible by d[q][k] == v, lowercase
weaver.utils.get_log_fmt() → AnyStr[source]
weaver.utils.get_log_date_fmt() → AnyStr[source]
weaver.utils.get_log_monitor_msg(job_id, status, percent, message, location) → AnyStr[source]
weaver.utils.get_job_log_msg(status, message, progress=0, duration=None) → AnyStr[source]
weaver.utils.make_dirs(path, mode=493, exist_ok=True)[source]

Alternative to os.makedirs with exists_ok parameter only available for python>3.5.

weaver.utils.fetch_file(file_reference, file_outdir) → AnyStr[source]

Fetches a file from a local path or remote URL and dumps it’s content to the specified output directory.

Parameters:
  • file_reference – Local filesystem path or remote URL file reference.
  • file_outdir – Output directory path of the fetched file.
Returns:

Path of the local copy of the fetched file.

weaver.utils.REGEX_SEARCH_INVALID_CHARACTERS[source]
weaver.utils.REGEX_ASSERT_INVALID_CHARACTERS[source]
weaver.utils.get_sane_name(name, min_len=3, max_len=None, assert_invalid=True, replace_character='_') → Union[AnyStr, None][source]

Returns a cleaned-up version of the input name, replacing invalid characters matched with REGEX_SEARCH_INVALID_CHARACTERS by replace_character.

Parameters:
  • name – value to clean
  • min_len – Minimal length of name to be respected, raises or returns None on fail according to assert_invalid.
  • max_len – Maximum length of name to be respected, raises or returns trimmed name on fail according to assert_invalid. If None, condition is ignored for assertion or full name is returned respectively.
  • assert_invalid – If True, fail conditions or invalid characters will raise an error instead of replacing.
  • replace_character – Single character to use for replacement of invalid ones if assert_invalid=False.
weaver.utils.assert_sane_name(name, min_len=3, max_len=None)[source]

Asserts that the sane name respects conditions.

See also

weaver.utils.clean_json_text_body(body) → AnyStr[source]

Cleans a textual body field of superfluous characters to provide a better human-readable text in a JSON response.