weaver.wps.utils ================ .. py:module:: weaver.wps.utils Module Contents --------------- .. py:data:: LOGGER .. py:function:: _get_settings_or_wps_config(container: weaver.typedefs.AnySettingsContainer, weaver_setting_name: str, config_setting_section: str, config_setting_name: str, default_not_found: str, message_not_found: str, load: bool = False) -> str .. py:function:: get_wps_path(container: weaver.typedefs.AnySettingsContainer, load: bool = True) -> str Retrieves the WPS path (without hostname). Searches directly in settings, then `weaver.wps_cfg` file, or finally, uses the default values if not found. .. py:function:: get_wps_url(container: weaver.typedefs.AnySettingsContainer, load: bool = True) -> str Retrieves the full WPS URL (hostname + WPS path). Searches directly in settings, then `weaver.wps_cfg` file, or finally, uses the default values if not found. .. py:function:: get_wps_output_dir(container: weaver.typedefs.AnySettingsContainer, load: bool = True) -> str Retrieves the WPS output directory path where to write XML and result files. Searches directly in settings, then `weaver.wps_cfg` file, or finally, uses the default values if not found. .. py:function:: get_wps_output_path(container: weaver.typedefs.AnySettingsContainer, load: bool = True) -> str Retrieves the WPS output path (without hostname) for staging XML status, logs and process outputs. Searches directly in settings, then `weaver.wps_cfg` file, or finally, uses the default values if not found. .. py:function:: get_wps_output_url(container: weaver.typedefs.AnySettingsContainer, load: bool = True) -> str Retrieves the WPS output URL that maps to WPS output directory path. Searches directly in settings, then `weaver.wps_cfg` file, or finally, uses the default values if not found. .. py:function:: get_wps_output_context(request: weaver.typedefs.AnyRequestType) -> Optional[str] Obtains and validates allowed values for sub-directory context of WPS outputs in header ``X-WPS-Output-Context``. :raises HTTPUnprocessableEntity: if the header was provided and contains invalid or illegal value. :returns: validated context or None if not specified. .. py:function:: get_wps_local_status_location(url_status_location: str, container: weaver.typedefs.AnySettingsContainer, must_exist: bool = True) -> Optional[str] Attempts to retrieve the local :term:`XML` file path corresponding to the :term:`WPS` status location as URL. :param url_status_location: URL reference pointing to some WPS status location :term:`XML`. :param container: any settings container to map configured local paths. :param must_exist: return only existing path if enabled, otherwise return the parsed value without validation. :returns: found local file path if it exists, ``None`` otherwise. .. py:function:: map_wps_output_location(reference: str, container: weaver.typedefs.AnySettingsContainer, url: bool = False, exists: bool = True, file_scheme: bool = False) -> Optional[str] Obtains the mapped WPS output location of a file where applicable. :param reference: Local file path or file URL to be mapped. :param container: Retrieve application settings. :param url: Perform URL mapping (``True``: local path -> URL endpoint, ``False``: URL endpoint -> local path). :param exists: Ensure that the mapped file exists, otherwise don't map it (otherwise ``None``). :param file_scheme: Ensure that the 'file://' scheme is applied to resulting local file location when mapped from WPS output URL. When in 'reverse' mode, 'file://' is always removed if present to form a potential local file path. :returns: Mapped reference that corresponds to the local/URL WPS output location. .. py:function:: _describe_process_cached(self: owslib.wps.WebProcessingService, identifier: str, xml: Optional[weaver.xml_util.XML] = None) -> weaver.typedefs.ProcessOWS .. py:function:: _get_wps_client_cached(url: str, headers: weaver.typedefs.HeadersType, verify: bool, language: Optional[str]) -> owslib.wps.WebProcessingService .. py:function:: get_wps_client(url: str, container: Optional[weaver.typedefs.AnySettingsContainer] = None, verify: bool = None, headers: Optional[weaver.typedefs.HeadersType] = None, language: Optional[str] = None) -> owslib.wps.WebProcessingService Obtains a :class:`WebProcessingService` with pre-configured request options for the given URL. :param url: WPS URL location. :param container: request or settings container to retrieve headers and other request options. :param verify: flag to enable SSL verification (overrides request options from container). :param headers: specific headers to apply (overrides retrieved ones from container). :param language: preferred response language if supported by the service. :returns: created WPS client object with configured request options. .. py:function:: get_wps_client_filtered_headers(headers: Optional[weaver.typedefs.HeadersType], container: weaver.typedefs.AnySettingsContainer) -> weaver.typedefs.HeadersType Filters out any headers configured for the :term:`WPS` client by the ``weaver.wps_client_headers_filter`` setting. :param headers: Headers to filter as applicable. :param container: Any settings container to retrieve application settings. :return: Filtered :term:`WPS` headers. .. py:function:: check_wps_status(location: Optional[str] = None, response: Optional[weaver.xml_util.XML] = None, sleep_secs: int = 2, verify: bool = True, settings: Optional[weaver.typedefs.AnySettingsContainer] = None) -> owslib.wps.WPSExecution Run :func:`owslib.wps.WPSExecution.checkStatus` with additional exception handling. :param location: job URL or file path where to look for job status. :param response: WPS response document of job status. :param sleep_secs: number of seconds to sleep before returning control to the caller. :param verify: flag to enable SSL verification. :param settings: application settings to retrieve any additional request parameters as applicable. :returns: OWSLib.wps.WPSExecution object. .. py:function:: get_exception_from_xml_status(xml: Union[weaver.xml_util.XML, str]) -> Optional[weaver.owsexceptions.OWSException] Retrieves the :term:`OWS` exception that corresponds to the :term:`XML` status. Expects the following :term:`XML` status response structure (``ows:Exception`` block can be at any level): .. code-block:: xml Error message about the cause of the exception. :param xml: XML tree object with exception details. :return: Matched :class:`owsexceptions.OWSException`. If no matching exception code is found within available exception classes, defaults to generic ``OWSException``. If any parsing error occurs, returns nothing. .. py:function:: load_pywps_config(container: weaver.typedefs.AnySettingsContainer, config: Optional[Union[str, Dict[str, str]]] = None) -> configparser.ConfigParser Loads and updates the PyWPS configuration using Weaver settings. .. py:function:: set_wps_language(wps: owslib.wps.WebProcessingService, accept_language: Optional[str] = None, request: Optional[weaver.typedefs.AnyRequestType] = None) -> Optional[str] Applies the best match between requested accept languages and supported ones by the WPS server. Given the `Accept-Language` header value, match the best language to the supported languages retrieved from WPS. By default, and if no match is found, sets :attr:`WebProcessingService.language` property to ``None``. .. seealso:: Details about the format of the ``Accept-Language`` header: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Accept-Language .. note:: This function considers quality-factor weighting and parsing resolution of ``Accept-Language`` header according to :rfc:`7231#section-5.3.2`. :param wps: service for which to apply a supported language if matched. :param accept_language: value of the Accept-Language header. :param request: request from which to extract Accept-Language header if not provided directly. :returns: language that has been set, or ``None`` if no match could be found.