Module Contents

class weaver.utils.CaseInsensitive(_str: str)[source]

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

Initialize self. See help(type(self)) for accurate signature.

weaver.utils.UNIT_SI_POWER_DOWN = ['m', 'µ', 'n', 'p', 'f', 'a', 'z', 'y'][source]
weaver.utils.UNIT_BIN_POWER = ['Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei', 'Zi', 'Yi'][source]
class weaver.utils._Singleton[source]
class weaver.utils.NullType[source]

Represents a null value to differentiate from None.

class weaver.utils.SchemaRefResolver(base_uri, referrer, store=m(), cache_remote=True, handlers=(), urljoin_cache=None, remote_cache=None)[source]

Reference resolver that supports both JSON and YAML files from a remote location.

resolve_remote(uri: str) weaver.typedefs.OpenAPISchema[source]

Resolve a remote uri.

If called directly, does not check the store first, but after retrieving the document at the specified URI it will be saved in the store if cache_remote is True.


If the requests library is present, jsonschema will use it to request the remote uri, so that the correct encoding is detected and used.

If it isn’t, or if the scheme of the uri is not http or https, UTF-8 is assumed.


uri (str):

The URI to resolve


The retrieved document

weaver.utils.get_weaver_url(container: weaver.typedefs.AnySettingsContainer) str[source]

Retrieves the home URL of the Weaver application.

weaver.utils.get_any_id(info: MutableMapping, default: Optional[str] = None, pop: bool = False, key: bool = False) Optional[str][source]

Retrieves a dictionary id-like key using multiple common variations [id, identifier, _id].

  • info – dictionary that potentially contains an id-like key.

  • default – Default identifier to be returned if none of the known keys were matched.

  • pop – If enabled, remove the matched key from the input mapping.

  • key – If enabled, return the matched key instead of the value.


value of the matched id-like key or None if not found.

weaver.utils.get_any_value(info: MutableMapping, default: Any = None, file: bool = True, data: bool = True, pop: bool = False, key: bool = False) weaver.typedefs.AnyValueType[source]

Retrieves a dictionary value-like key using multiple common variations [href, value, reference, data].

  • info – Dictionary that potentially contains a value-like key.

  • default – Default value to be returned if none of the known keys were matched.

  • file – If enabled, file-related key names will be considered.

  • data – If enabled, data-related key names will be considered.

  • pop – If enabled, remove the matched key from the input mapping.

  • key – If enabled, return the matched key instead of the value.


Value (or key if requested) of the matched value-like key or None if not found.

weaver.utils.get_any_message(info: weaver.typedefs.JSON) str[source]

Retrieves a dictionary ‘value’-like key using multiple common variations [message].


info – dictionary that potentially contains a ‘message’-like key.


value of the matched ‘message’-like key or an empty string if not found.

weaver.utils.get_registry(container: Optional[weaver.typedefs.AnyRegistryContainer] = None, nothrow: bool = False) Optional[pyramid.registry.Registry][source]

Retrieves the application registry from various containers referencing to it.

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

Retrieves the application settings from various containers referencing to it.

weaver.utils.get_header(header_name: Ellipsis, header_container: weaver.typedefs.AnyHeadersContainer, default: Optional[str], Optional[Union[str, List[str]]], bool = None, pop: bool = False) Optional[Union[str, List[str]]][source]

Find the specified header within a header container.

Retrieves header_name by fuzzy match (independently of upper/lower-case and underscore/dash) from various framework implementations of Headers.

  • header_name – header to find.

  • header_container – where to look for header_name.

  • default – returned value if header_container is invalid or header_name is not found.

  • pop – remove the matched header(s) by name from the input container.

Looks for cookie_header_name header within header_container.


new header container in the form {'Cookie': <found_cookie>} if it was matched, or empty otherwise.

weaver.utils.parse_kvp(query: Ellipsis, key_value_sep: str = '=', pair_sep: str = ';', nested_pair_sep: Optional[str] = '', multi_value_sep: Optional[str] = ',', accumulate_keys: bool = True, unescape_quotes: bool = True, strip_spaces: bool = True, case_insensitive: bool = True) weaver.typedefs.KVP[source]

Parse key-value pairs using specified separators.

All values are normalized under a list, whether their have an unique or multi-value definition. When a key is by itself (without separator and value), the resulting value will be an empty list.

When accumulate_keys is enabled, entries such as {key}={val};{key}={val} will be joined together under the same list as if they were specified using directly {key}={val},{val} (default separators employed only for demonstration purpose). Both nomenclatures can also be employed simultaneously.

When nested_pair_sep is provided, definitions that contain nested key_value_sep character within an already established KVP will be parsed once again. This will parse {key}={subkey1}={val1},{subkey2}={val2} into a nested KVP dictionary as value under the top level KVP entry {key}. Separators are passed down for nested parsing, except pair_sep that is replaced by nested_pair_sep.

>> parse_kvp("format=json&inputs=key1=value1;key2=val2,val3", pair_sep="&", nested_pair_sep=";")
    'format': ['json'],
    'inputs': {
        'key1': ['value1'],
        'key2': ['val2', 'val3']
  • query – Definition to be parsed as KVP.

  • key_value_sep – Separator that delimits the keys from their values.

  • pair_sep – Separator that distinguish between different (key, value) entries.

  • nested_pair_sep – Separator to parse values of pairs containing nested KVP definition.

  • multi_value_sep – Separator that delimits multiple values associated to the same key. If empty or None, values will be left as a single entry in the list under the key.

  • accumulate_keys – Whether replicated keys should be considered equivalent to multi-value entries.

  • unescape_quotes – Whether to remove single and double quotes around values.

  • strip_spaces – Whether to remove spaces around values after splitting them.

  • case_insensitive – Whether to consider keys as case-insensitive. If True, resulting keys will be normalized to lowercase. Otherwise, original keys are employed.


Parsed KVP.


HTTPBadRequest – If parsing cannot be accomplished based on parsing conditions.

weaver.utils.parse_prefer_header_execute_mode(header_container: Ellipsis, supported_modes: Optional[List[weaver.execute.AnyExecuteControlOption]] = None, wait_max: int = 10) Tuple[weaver.execute.AnyExecuteMode, Optional[int], weaver.typedefs.HeadersType][source]

Obtain execution preference if provided in request headers.

See also

See also

If Prefer format is valid, but server decides it cannot be respected, it can be transparently ignored (RFC 7240#section-2). The server must respond with Preference-Applied indicating preserved preferences it decided to respect.

  • header_container – Request headers to retrieve preference, if any available.

  • supported_modes – Execute modes that are permitted for the operation that received the Prefer header. Resolved mode will respect this constrain following specification requirements of OGC API - Processes.

  • wait_max – Maximum wait time enforced by the server. If requested wait time is greater, wait preference will not be applied and will fallback to asynchronous response.


Tuple of resolved execution mode, wait time if specified, and header of applied preferences if possible. Maximum wait time indicates duration until synchronous response should fallback to asynchronous response.


HTTPBadRequest – If contents of Prefer are not valid.

weaver.utils.get_url_without_query(url: Union[str, urllib.parse.ParseResult]) str[source]

Removes the query string part of an URL.

weaver.utils.is_valid_url(url: Optional[str]) typing_extensions.TypeGuard[str][source]
class weaver.utils.VersionLevel[source]

Constants container that provides similar functionalities to ExtendedEnum without explicit Enum membership.

MAJOR = major[source]
MINOR = minor[source]
PATCH = patch[source]
class weaver.utils.VersionFormat[source]

Constants container that provides similar functionalities to ExtendedEnum without explicit Enum membership.

OBJECT = object[source]
STRING = string[source]
PARTS = parts[source]
weaver.utils.as_version_major_minor_patch(version: weaver.typedefs.AnyVersion, version_format: weaver.typedefs.Literal[VersionFormat]) distutils.version.LooseVersion[source]
weaver.utils.as_version_major_minor_patch(version: weaver.typedefs.AnyVersion, version_format: weaver.typedefs.Literal[VersionFormat]) str
weaver.utils.as_version_major_minor_patch(version: weaver.typedefs.AnyVersion, version_format: weaver.typedefs.Literal[VersionFormat]) Tuple[int, int, int]
weaver.utils.as_version_major_minor_patch(version: weaver.typedefs.AnyVersion) Tuple[int, int, int]

Generates a MAJOR.MINOR.PATCH version with padded with zeros for any missing parts.

weaver.utils.is_update_version(version: weaver.typedefs.AnyVersion, taken_versions: Iterable[weaver.typedefs.AnyVersion], version_level: VersionLevel = VersionLevel.PATCH) typing_extensions.TypeGuard[weaver.typedefs.AnyVersion][source]

Determines if the version corresponds to an available update version of specified level compared to existing ones.

If the specified version corresponds to an older version compared to available ones (i.e.: a taken more recent version also exists), the specified version will have to fit within the version level range to be considered valid. For example, requesting PATCH level will require that the specified version is greater than the last available version against other existing versions with equivalent MAJOR.MINOR parts. If 1.2.0 and 2.0.0 were taken versions, and 1.2.3 has to be verified as the update version, it will be considered valid since its PATCH number 3 is greater than all other 1.2.x versions (it falls within the [1.2.x, 1.3.x[ range). Requesting instead MINOR level will require that the specified version is greater than the last available version against existing versions of same MAJOR part only. Using again the same example values, 1.3.0 would be valid since its MINOR part 3 is greater than any other 1.x taken versions. On the other hand, version 1.2.4 would not be valid as x = 2 is already taken by other versions considering same 1.x portion (PATCH part is ignored in this case since MINOR is requested, and 2.0.0 is ignored as not the same MAJOR portion of 1 as the tested version). Finally, requesting a MAJOR level will require necessarily that the specified version is greater than all other existing versions for update, since MAJOR is the highest possible semantic part, and higher parts are not available to define an upper version bound.


As long as the version level is respected, the actual number of this level and all following ones can be anything as long as they are not taken. For example, PATCH with existing 1.2.3 does not require that the update version be 1.2.4. It can be 1.2.5, 1.2.24, etc. as long as 1.2.x is respected. Similarly, MINOR update can provide any PATCH number, since 1.x only must be respected. From existing 1.2.3, MINOR update could specify 1.4.99 as valid version. The PATCH part does not need to start back at 0.

  • version – Version to validate as potential update revision.

  • taken_versions – Existing versions that cannot be reused.

  • version_level – Minimum level to consider availability of versions as valid revision number for update.


Status of availability of the version.

weaver.utils.is_uuid(maybe_uuid: Any) typing_extensions.TypeGuard[weaver.typedefs.AnyUUID][source]

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

weaver.utils.as_int(value: Any, default: int) int[source]

Ensures a value is converted to int.

weaver.utils.parse_extra_options(option_str: str, sep: str = ',') Dict[str, Optional[str]][source]

Parses the extra options parameter.

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

  • option_str – A string parameter with the extra options.

  • sep – separator to employ in order to split the multiple values within the option string.


A dict with the parsed extra options.

weaver.utils.extend_instance(obj: OriginalClass, cls: Type[ExtenderMixin]) ExtendedClass[source]

Extend an existing instance of a given class by applying new definitions from the specified mixin class type.

weaver.utils.fully_qualified_name(obj: Union[Any, Type[Any]]) str[source]

Obtains the full path definition of the object to allow finding and importing it.

For classes, functions and exceptions, the following format is returned:


The module is omitted if it is a builtin object or type.

For methods, the class is also represented, resulting in the following format:

weaver.utils.import_target(target: str, default_root: Optional[str] = None) Optional[Any][source]

Imports a target resource class or function from a Python script as module or directly from a module reference.

The Python script does not need to be defined within a module directory (i.e.: with __init__.py). Files can be imported from virtually anywhere. To avoid name conflicts in generated module references, each imported target employs its full escaped file path as module name.

Formats expected as follows:

  • target – Resource to be imported.

  • default_root – Root directory to employ if target is relative (default magpie.constants.MAGPIE_ROOT).


Found and imported resource or None.

weaver.utils.now(tz_name: Optional[str] = None) datetime.datetime[source]

Obtain the current time with timezone-awareness.


tz_name – If specified, returned current time will be localized to specified timezone.

weaver.utils.wait_secs(run_step: int = - 1) int[source]

Obtain a wait time in seconds within increasing delta intervals based on iteration index.

weaver.utils.localize_datetime(dt: datetime.datetime, tz_name: Optional[str] = None) datetime.datetime[source]

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


Any datetime provided as input that is not already timezone-aware will be assumed to be relative to the current locale timezone. This is the default returned by naive datetime.datetime instances.

If no timezone name is provided, the timezone-aware datatime will be localized with locale timezone offset. Otherwise, the desired localization will be applied with the specified timezone offset.

weaver.utils.get_file_header_datetime(dt: datetime.datetime) str[source]

Obtains the standard header datetime representation.

See also

Format of the date defined in RFC 5322#section-3.3.

weaver.utils.get_file_headers(path: str, download_headers: bool = False, content_headers: bool = False, content_type: Optional[str] = None) weaver.typedefs.HeadersType[source]

Obtain headers applicable for the provided file.

  • path – File to describe.

  • download_headers – If enabled, add the attachment filename for downloading the file.

  • content_headers – If enabled, add Content- prefixed headers.

  • content_type – Explicit Content-Type to provide. Otherwise, use default guessed by file system.


Headers for the file.

weaver.utils.get_base_url(url: str) str[source]

Obtains the base URL from the given url.

weaver.utils.xml_path_elements(path: str) List[str][source]
weaver.utils.xml_strip_ns(tree: weaver.xml_util.XML) None[source]
weaver.utils.ows_context_href(href: str, partial: Optional[bool] = False) weaver.typedefs.JSON[source]

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

weaver.utils.pass_http_error(exception: Exception, expected_http_error: Union[Type[pyramid.httpexceptions.HTTPError], Iterable[Type[pyramid.httpexceptions.HTTPError]]]) None[source]

Silently ignore a raised HTTP error that matches the specified error code of the reference exception class.

Given an HTTPError of any type (pyramid, requests), ignores the exception if the actual error matches the status code. Other exceptions are re-raised. This is equivalent to capturing a specific Exception within an except block and calling pass to drop it.

  • 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.


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.xml_util.XML) Optional[NoReturn][source]

Raises an exception with the description if the XML response document defines an ExceptionReport.


xml_node – instance of XML


Exception – on found ExceptionReport document.

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

Obtains the bytes representation of the string.

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

Obtains the unicode representation of the string.

weaver.utils.islambda(func: Any) bool[source]
weaver.utils.get_path_kvp(path: str, sep: str = ',', **params: weaver.typedefs.KVP_Item) str[source]

Generates the URL with Key-Value-Pairs (KVP) query 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.


combined path and query parameters as KVP.

weaver.utils.get_log_fmt() str[source]

Logging format employed for job output reporting.

weaver.utils.get_log_date_fmt() str[source]

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[source]
weaver.utils.get_job_log_msg(status: Union[weaver.status.Status, str], message: str, progress: Optional[weaver.typedefs.Number] = 0, duration: Optional[str] = None) str[source]
weaver.utils.setup_loggers(settings: Ellipsis = None, level: Optional[Union[int, str]] = None, force_stdout: bool = False, message_format: Optional[str] = None, datetime_format: Optional[str] = None, log_file: Optional[str] = None) logging.Logger[source]

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: str, mode: int = 493, exist_ok: bool = False) None[source]

Backward compatible make_dirs with reduced set of default mode flags.

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.


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: int = 2, base_class: bool = False) str[source]

Find the name of a parent caller function or method.

The name is returned with respective formats module.class.method or module.function.

  • 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.


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

weaver.utils.setup_cache(settings: weaver.typedefs.SettingsType) None[source]

Prepares the settings with default caching options.

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

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


caching_args – tuple of (function, region, *function-args) representing caching key to invalidate.

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

Obtains the SSL verification option considering multiple setting definitions and the provided request context.

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

  • 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.


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[source]

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

See also

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

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


whether to disable cache or not

weaver.utils.get_request_options(method: str, url: str, settings: weaver.typedefs.AnySettingsContainer) weaver.typedefs.SettingsType[source]

Obtains the request options corresponding to the request from the configuration file.

The configuration file specified is expected to be pre-loaded within setting weaver.request_options. If no file was pre-loaded or no match is found for the request, an empty options dictionary is returned.

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

  • url – request URL.

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


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

weaver.utils.retry_on_condition(operation: Ellipsis, *args: RetryCondition, condition: int = Exception, retries=1, **kwargs) weaver.typedefs.ReturnValue[source]

Retries the operation call up to the amount of specified retries if the condition is encountered.

  • operation – Any callable lambda, function, method, class that sporadically raises an exception to catch.

  • condition – Exception(s) to catch or callable that takes the raised exception to handle it with more conditions. In case of a callable, success/failure result should be returned to indicate if retry is needed. If retry is not requested by the handler for the specified exception, it is raised directly.

  • retries – Amount of retries to perform. If retries are exhausted, the final exception is re-raised.


Expected normal operation return value if it was handled within the specified amount of retries.

weaver.utils.retry_on_cache_error(func: weaver.typedefs.AnyCallable) weaver.typedefs.AnyCallable[source]

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, weaver.typedefs.AnyValueType]) requests.Response[source]

Request operation employed by request_extra() without caching.

weaver.utils._request_cached(method: str, url: str, kwargs: Dict[str, weaver.typedefs.AnyValueType]) requests.Response[source]

Cached-enabled request operation employed by request_extra().

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

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.


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


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

  • 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.download_file_http(file_reference: str, file_outdir: str, settings: Optional[weaver.typedefs.AnySettingsContainer] = None, **request_kwargs: Any) str[source]

Downloads the file referenced by an HTTP URL location.

Respects RFC 2183, RFC 5987 and RFC 6266 regarding Content-Disposition header handling to resolve any preferred file name. This value is employed if it fulfill validation criteria. Otherwise, the name is extracted from the last part of the URL path.

  • file_reference – HTTP URL where the file is hosted.

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

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

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


Path of the local copy of the fetched file.

  • HTTPException – applicable HTTP-based exception if any unrecoverable problem occurred during fetch request.

  • ValueError – when resulting file name value is considered invalid.

weaver.utils.fetch_file(file_reference: str, file_outdir: str, settings: Optional[weaver.typedefs.AnySettingsContainer] = None, link: Optional[bool] = None, move: bool = False, **request_kwargs: Any) str[source]

Fetches a file from local path, AWS-S3 bucket or remote URL, and dumps it’s content to the 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 (after link resolution if applicable). Requests will consider weaver.request_options when using http(s):// scheme.

  • 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).

  • link – If True, force generation of a symbolic link instead of hard copy, regardless if source is a file or link. If False, force hard copy of the file to destination, regardless if source is a file or link. If None (default), resolve automatically as follows. When the source is a symbolic link itself, the destination will also be a link. When the source is a direct file reference, the destination will be a hard copy of the file. Only applicable when the file reference is local.

  • move – Move local file to the output directory instead of copying or linking it. No effect if the output directory already contains the local file. No effect if download must occurs for remote file.

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


Path of the local copy of the fetched file.

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

  • ValueError – when the reference scheme cannot be identified.

weaver.utils.load_file(file_path: str, text: bool = False) Union[weaver.typedefs.JSON, str][source]

Load JSON or YAML file contents from local path or remote URL.

If URL, get the content and validate it by loading, otherwise load file directly.

  • file_path – Local path or URL endpoint where file to load is located.

  • text – load contents as plain text rather than parsing it from JSON/YAML.


loaded contents either parsed and converted to Python objects or as plain text.


ValueError – if YAML or JSON cannot be parsed or loaded from location.

weaver.utils.is_remote_file(file_location: str) typing_extensions.TypeGuard[str][source]

Parses to file location to figure out if it is remotely available or a local path.

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][source]

Cleans up the name to allow only specified characters and conditions.

Returns a cleaned-up version of the name, replacing invalid characters not matched with REGEX_SEARCH_INVALID_CHARACTERS by replace_character. Also, ensure that the resulting name respects specified length conditions.

  • 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: str, min_len: int = 3, max_len: Optional[int] = None) None[source]

Asserts that the sane name respects conditions.

See also

weaver.utils.clean_json_text_body(body: str, remove_newlines: bool = True, remove_indents: bool = True) str[source]

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: Ellipsis, rename: Optional[Dict[weaver.typedefs.AnyKey, Any]] = None, remove: Optional[List[weaver.typedefs.AnyKey]] = None, add: Optional[Dict[weaver.typedefs.AnyKey, Any]] = None, replace_values: Optional[Dict[weaver.typedefs.AnyKey, Any]] = None, replace_func: Optional[Dict[weaver.typedefs.AnyKey, Callable[[Any], Any]]] = None) Dict[str, weaver.typedefs.JSON][source]

Transforms the input JSON with different methods.

The transformations are applied in-place and in the same order as the arguments (rename, remove, add, etc.). All operations are applied onto the top-level fields of the mapping. No nested operations are applied, unless handled by replace functions.


Because fields and values are iterated over the provided mappings, replacements of previous iterations could be re-replaced by following ones if the renamed item corresponds to a following item to match. For example, renaming field1 -> field2 and field2 -> field3` within the same operation type would result in successive replacements with ``field3 as result. The parameter order is important in this case as swapping the definitions would not find field2 on the first iteration (not in mapping yet), and then find field1, making the result to be field2.

  • json_data – JSON mapping structure to transform.

  • rename – rename matched fields key name to the associated value name.

  • remove – remove matched fields by name.

  • add – add or override the fields names with associated values.

  • replace_values – replace matched values by the associated new values regardless of field names.

  • replace_func – Replace values under matched fields by name with the returned value from the associated function. Mapping functions will receive the original value as input. If the result is to be serialized to JSON, they should return a valid JSON-serializable value.


transformed JSON (same as modified in-place input JSON).

weaver.utils.generate_diff(val: Any, ref: Any, val_name: str = 'Test', ref_name: str = 'Reference') str[source]

Generates a line-by-line diff result of the test value against the reference value.

Attempts to parse the contents as JSON to provide better diff of matched/sorted lines, and falls back to plain line-based string representations otherwise.

  • val – Test input value.

  • ref – Reference input value.

  • val_name – Name to apply in diff for test input value.

  • ref_name – Name to apply in diff for reference input value.


Formatted multiline diff,

weaver.utils.apply_number_with_unit(number: weaver.typedefs.Number, unit: str = '', binary: bool = False, decimals: int = 3) str[source]

Apply the relevant unit and prefix factor to the specified number to create a human-readable value.

  • number – Numeric value with no unit.

  • unit – Unit to be applied. Auto-resolved to ‘B’ if binary requested. Factor applied accordingly to number.

  • binary – Use binary multiplier (powers of 2) instead of SI decimal multipliers (powers of 10).

  • decimals – Number of decimals to preserve after unit is applied.


String of the numeric value with appropriate unit.

weaver.utils.parse_number_with_unit(number: str, binary: Optional[bool] = None) weaver.typedefs.Number[source]

Parses a numeric value accompanied with a unit to generate the unit-less value without prefix factor.

  • number – Numerical value and unit. Unit is dissociated from value with first non-numerical match. Unit is assumed to be present (not only the multiplier by itself). This is important to avoid confusion (e.g.: m used for meters vs m prefix for “milli”).

  • binary – Force use (True) or non-use (False) of binary multiplier (powers of 2) instead of SI decimal multipliers (powers of 10) for converting value (with applicable unit multiplier if available). If unspecified (None), auto-detect from unit (e.g.: powers of 2 for MiB, powers of 10 for MB). When unspecified, the B character is used to auto-detect if binary should apply, SI multipliers are otherwise assumed.


Literal value.

weaver.utils.copy_doc(copy_func: weaver.typedefs.AnyCallableAnyArgs) weaver.typedefs.AnyCallableAnyArgs[source]

Decorator to copy the docstring from one callable to another.


def copy_func(self): pass