weaver.utils

Module Contents

weaver.utils.LOGGER

weaver.utils.SUPPORTED_FILE_SCHEMES

class weaver.utils._Singleton

class weaver.utils.NullType

Represents a null value to differentiate from None.

weaver.utils.null

weaver.utils.get_weaver_url(container: weaver.typedefs.AnySettingsContainer) → str

Retrieves the home URL of the weaver application.

weaver.utils.get_any_id(info: weaver.typedefs.JSON) → Union[str, None]

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: weaver.typedefs.JSON) → weaver.typedefs.AnyValue

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: weaver.typedefs.JSON) → str

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: weaver.typedefs.AnyRegistryContainer, nothrow: bool = False) → Optional[Registry]

Retrieves the application registry from various containers referencing to it.

weaver.utils.get_settings(container: Optional[AnySettingsContainer] = None) → weaver.typedefs.SettingsType

Retrieves the application settings from various containers referencing to it.

weaver.utils.get_header(header_name: str, header_container: weaver.typedefs.AnyHeadersContainer) → Union[str, None]

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

weaver.utils.get_cookie_headers(header_container: weaver.typedefs.AnyHeadersContainer, cookie_header_name: Optional[str] = 'Cookie') → weaver.typedefs.HeadersType

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: Union[str, ParseResult]) → str

Removes the query string part of an URL.

weaver.utils.is_valid_url(url: Optional[str]) → bool

weaver.utils.UUID_PATTERN

weaver.utils.is_uuid(maybe_uuid: Any) → bool

Evaluates if the provided input is a UUID-like string.

weaver.utils.parse_extra_options(option_str: str) → Dict[str, str]

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: Union[Any, Type[Any]]) → str

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

weaver.utils.now() → datetime.datetime

weaver.utils.now_secs() → int

Return the current time in seconds since the Epoch.

weaver.utils.wait_secs(run_step=- 1)

weaver.utils.expires_at(hours: Optional[int] = 1) → int

weaver.utils.localize_datetime(dt: datetime.datetime, tz_name: Optional[str] = 'UTC') → datetime.datetime

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

weaver.utils.get_base_url(url: str) → str

Obtains the base URL from the given url.

weaver.utils.xml_path_elements(path: str) → List[str]

weaver.utils.xml_strip_ns(tree: weaver.typedefs.XML) → None

weaver.utils.ows_context_href(href: str, partial: Optional[bool] = False) → weaver.typedefs.JSON

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

weaver.utils.pass_http_error(exception: Exception, expected_http_error: Union[Type[PyramidHTTPError], Iterable[Type[PyramidHTTPError]]]) → None

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: weaver.typedefs.XML) → Optional[NoReturn]

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

weaver.utils.str2bytes(string: Union[str, bytes]) → bytes

Obtains the bytes representation of the string.

weaver.utils.bytes2str(string: Union[str, bytes]) → str

Obtains the unicode representation of the string.

weaver.utils.islambda(func: Any) → bool

weaver.utils.first_cap_re

weaver.utils.all_cap_re

weaver.utils.convert_snake_case(name: str) → str

weaver.utils.parse_request_query(request: pyramid.request.Request) → Dict[str, Dict[AnyKey, str]]

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_path_kvp(path: str, sep: str = ',', **params: weaver.typedefs.KVP_Item) → str

Generates the WPS URL with Key-Value-Pairs (KVP) query parameters.

Parameters

• path – WPS URL or Path

• sep – separator to employ when multiple values are provided.

• params – keyword parameters and their corresponding single or multi values to generate KVP.

Returns

combined path and query parameters as KVP.

weaver.utils.get_log_fmt() → str

Logging format employed for job output reporting.

weaver.utils.get_log_date_fmt() → str

Logging date format employed for job output reporting.

weaver.utils.get_log_monitor_msg(job_id: str, status: str, percent: weaver.typedefs.Number, message: str, location: str) → str

weaver.utils.get_job_log_msg(status: str, message: str, progress: Optional[Number] = 0, duration: Optional[str] = None) → str

weaver.utils.setup_loggers(settings: weaver.typedefs.AnySettingsContainer, level: Optional[Union[int, str]] = None) → None

Update logging configuration known loggers based on application settings.

When weaver.log_level exists in settings, it overrides any other INI configuration logging levels. Otherwise, undefined logger levels will be set according to whichever is found first between weaver.log_level, the level parameter or default logging.INFO.

weaver.utils.make_dirs(path, mode=493, exist_ok=False)

Alternative to os.makedirs with exists_ok parameter only available for python>3.5. Also using a reduced set of permissions 755 instead of original default 777.

Note

The method employed in this function is safer then if os.pat.exists or if os.pat.isdir pre-check to calling os.makedirs as this can result in race condition (between evaluation and actual creation).

weaver.utils.get_caller_name(skip=2, base_class=False)

Returns the name of a caller in the format module.class.method.

Parameters

• skip – specifies how many levels of stack to skip while getting the caller.

• base_class – Specified if the base class should be returned or the top-most class in case of inheritance If the caller is not a class, this doesn’t do anything.

Returns

An empty string if skipped levels exceed stack height; otherwise, the requested caller name.

weaver.utils.setup_cache(settings: weaver.typedefs.SettingsType) → None

Prepares the settings with default caching options.

weaver.utils.invalidate_region(caching_args: Tuple[Callable, str, Tuple[Any]]) → None

Caching region invalidation with handling to ignore errors generated by of unknown regions.

Parameters

caching_args – tuple of (function, region, *function-args)

weaver.utils.get_ssl_verify_option(method: str, url: str, settings: weaver.typedefs.AnySettingsContainer, request_options: Optional[SettingsType] = None) → bool

Obtains the SSL verification option from combined settings from weaver.ssl_verify and parsed weaver.request_options file for the corresponding request.

Parameters

• method – request method (GET, POST, etc.).

• url – request URL.

• settings – application setting container with pre-loaded request options specifications.

• request_options – pre-processed request options for method/URL to avoid re-parsing the settings.

Returns

SSL verify option to be passed down to some request function.

weaver.utils.get_no_cache_option(request_headers: weaver.typedefs.HeadersType, request_options: weaver.typedefs.SettingsType) → bool

Obtains the No-Cache result from request headers and configured request options.

See also

• Request.headers()

• get_request_options()

Parameters

• request_headers – specific request headers that could indicate Cache-Control: no-cache

• request_options – specific request options that could define cache: True|False

Returns

whether to disable cache or not

weaver.utils.get_request_options(method: str, url: str, settings: weaver.typedefs.AnySettingsContainer) → weaver.typedefs.SettingsType

Obtains the request options corresponding to the request according to configuration file specified by pre-loaded setting weaver.request_options.

If no file was pre-loaded or no match is found for the request, an empty options dictionary is returned.

See also

• get_ssl_verify_option()

• config/request_options.yml.example

Parameters

• method – request method (GET, POST, etc.).

• url – request URL.

• settings – application setting container with pre-loaded request options specifications.

Returns

dictionary with keyword options to be applied to the corresponding request if matched.

weaver.utils.retry_on_cache_error(func: Callable[[…], Any]) → Callable

Decorator to handle invalid cache setup.

Any function wrapped with this decorator will retry execution once if missing cache setup was the cause of error.

weaver.utils._request_call(method: str, url: str, kwargs: Dict[str, AnyValue]) → requests.Response

Request operation employed by request_extra() without caching.

weaver.utils._request_cached(method: str, url: str, kwargs: Dict[str, AnyValue]) → requests.Response

Cached-enabled request operation employed by request_extra().

weaver.utils.request_extra(method: str, url: str, retries: Optional[int] = None, backoff: Optional[Number] = None, intervals: Optional[List[Number]] = None, retry_after: bool = True, allowed_codes: Optional[List[int]] = None, only_server_errors: bool = True, ssl_verify: Optional[bool] = None, settings: Optional[AnySettingsContainer] = None, **request_kwargs: str) → weaver.typedefs.AnyResponseType

Standard library requests with additional functional utilities.

Retry operation

Implements request retry if the previous request failed, up to the specified number of retries. Using backoff factor, you can control the interval between request attempts such as:

delay = backoff * (2 ^ retry)

Alternatively, you can explicitly define intervals=[...] with the list values being the number of seconds to wait between each request attempt. In this case, backoff is ignored and retries is overridden accordingly with the number of items specified in the list.

Furthermore, retry_after (default: True) indicates if HTTP status code 429 (Too Many Requests) should be automatically handled during retries. If enabled and provided in the previously failed request response through the Retry-After header, the next request attempt will be executed only after the server-specified delay instead of following the calculated delay from retries and backoff, or from corresponding index of interval, accordingly to specified parameters. This will avoid uselessly calling the server and automatically receive a denied response. You can disable this feature by passing False, which will result into requests being retried blindly without consideration of the called server instruction.

Because different request implementations use different parameter naming conventions, all following keywords are looked for:

• Both variants of backoff and backoff_factor are accepted.

• All variants of retires, retry and max_retries are accepted.

Note

Total amount of executed request attempts will be +1 the number of retries or intervals items as first request is done immediately, and following attempts are done with the appropriate delay.

File Transport Scheme

Any request with file:// scheme or empty scheme (no scheme specified) will be automatically handled as potential local file path. The path should be absolute to ensure it to be correctly resolved.

All access errors due to file permissions return 403 status code, and missing file returns 404. Any other IOError types are converted to a 400 responses.

See also

• FileAdapter

SSL Verification

Allows SSL verify option to be enabled or disabled according to configuration settings or explicit parameters. Any variation of verify or ssl_verify keyword arguments are considered. If they all resolve to True, then application settings are retrieved from weaver.ini to parse additional SSL options that could disable it.

Following weaver settings are considered :

• weaver.ssl_verify = True|False

• weaver.request_options = request_options.yml

Note

Argument settings must also be provided through any supported container by get_settings() to retrieve and apply any weaver-specific configurations.

See also

• get_request_options()

• get_ssl_verify_option()

Parameters

• method – HTTP method to set request.

• url – URL of the request to execute.

• retries – Number of request retries to attempt if first attempt failed (according to allowed codes or error).

• backoff – Factor by which to multiply delays between retries.

• intervals – Explicit intervals in seconds between retries.

• retry_after – If enabled, honor Retry-After response header of provided by a failing request attempt.

• allowed_codes – HTTP status codes that are considered valid to stop retrying (default: any non-4xx/5xx code).

• ssl_verify – Explicit parameter to disable SSL verification (overrides any settings, default: True).

• settings – Additional settings from which to retrieve configuration details for requests.

• only_server_errors – Only HTTP status codes in the 5xx values will be considered for retrying the request (default: True). This catches sporadic server timeout, connection error, etc., but 4xx errors are still considered valid results. This parameter is ignored if allowed codes are explicitly specified.

• request_kwargs – All other keyword arguments are passed down to the request call.

weaver.utils.fetch_file(file_reference: str, file_outdir: str, settings: Optional[AnySettingsContainer] = None, **request_kwargs: Any) → str

Fetches a file from a local path, an AWS-S3 bucket or remote URL, and dumps it’s content to the specified output directory.

The output directory is expected to exist prior to this function call. The file reference scheme (protocol) determines from where to fetch the content. Output file name and extension will be the same as the original. Requests will consider weaver.request_options when using http(s):// scheme.

Parameters

• file_reference – Local filesystem path (optionally prefixed with file://), s3:// bucket location or http(s):// remote URL file reference. Reference https://s3.[...] are also considered as s3://.

• file_outdir – Output local directory path under which to place the fetched file.

• settings – Additional request-related settings from the application configuration (notably request-options).

• request_kwargs – Additional keywords to forward to request call (if needed).

Returns

Path of the local copy of the fetched file.

Raises

• HTTPException – applicable HTTP-based exception if any occurred during the operation.

• ValueError – when the reference scheme cannot be identified.

weaver.utils.REGEX_SEARCH_INVALID_CHARACTERS

weaver.utils.REGEX_ASSERT_INVALID_CHARACTERS

weaver.utils.get_sane_name(name: str, min_len: Optional[int] = 3, max_len: Optional[Union[int, None]] = None, assert_invalid: Optional[bool] = True, replace_character: str = '_') → Union[str, None]

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

See also

weaver.wps_restapi.swagger_definitions.SLUG

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 is False.

weaver.utils.assert_sane_name(name, min_len=3, max_len=None)

Asserts that the sane name respects conditions.

See also

• argument details in get_sane_name()

weaver.utils.clean_json_text_body(body: str, remove_newlines: bool = True, remove_indents: bool = True) → str

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

weaver.utils.transform_json(json_data: weaver.typedefs.JSON, rename: Optional[Dict[AnyKey, Any]] = None, remove: Optional[List[AnyKey]] = None, add: Optional[Dict[AnyKey, Any]] = None, replace_values: Optional[Dict[AnyKey, Any]] = None, replace_func: Optional[Dict[AnyKey, Callable[[Any], Any]]] = None) → weaver.typedefs.JSON

Transforms the input json_data with different methods. The transformations are applied in the same order as the arguments.