Source code for weaver.wps_restapi.utils

import functools
import inspect
import logging
from copy import deepcopy
from typing import TYPE_CHECKING

import colander
from pyramid.httpexceptions import HTTPBadRequest, HTTPSuccessful, status_map

from weaver.utils import get_header, get_settings, get_weaver_url
from weaver.wps_restapi import swagger_definitions as sd

    from typing import Any, Callable, Dict, Optional

    from weaver.typedefs import AnySettingsContainer, HeadersType

[docs]LOGGER = logging.getLogger(__name__)
[docs]class HTTPHeadFileResponse(HTTPSuccessful): """ Provides additional header handing when returning a response to HTTP HEAD request. When returning from HTTP HEAD, the body contents are omitted from the response. The response **MUST NOT** contain the body contents, but the HTTP headers **SHOULD** be identical to the corresponding HTTP GET request. .. seealso:: :rfc:`2616#section-9.4` from pyramid.httpexceptions import HTTPException .. note:: Even though no content is provided for HEAD response, ``204`` **SHOULD NOT** be used since it must emulate the GET response that would contain the content. When setting :attr:`HTTPException.empty_body` on :class:`pyramid.httpexceptions.HTTPException` derived classes, :mod:`pyramid` incorrectly drops important headers such as ``Content-Type`` and ``Content-Length`` that should be reported as if the file was returned when the represented entity is a file, although no content is actually present. When instead the body is omitted (``text=""`` or ``body=b''``), the :meth:`HTTPException.prepare` method also incorrectly overrides the ``Content-Type`` and ``Content-Length`` values. Finally, ``Content-Length`` also gets recalculated when the content iterator is created from the initialization parameters. This class takes care of all these edge cases to properly report content headers of HEAD requests although none is provided. """ def __init__(self, code=200, headers=None, **kwargs): # type: (int, Optional[HeadersType], **Any) -> None # drop any 'app_iter' content generator that would recalculate and reset the content_length kwargs.pop("body", None) kwargs.pop("json", None) kwargs.pop("text", None) self.code = code http_class = status_map[code] self.title = http_class.title self.explanation = http_class.explanation content_type = None if headers: # in order to automatically add charset when needed, 'content_type' creates a duplicate # remove the original preemptively to avoid errors in parsers receiving the response headers = deepcopy(headers) content_type = get_header("Content-Type", headers, pop=True) super(HTTPHeadFileResponse, self).__init__( content_type=content_type, # don't override content-type headerlist=None, # extend content-type with charset as applicable headers=headers, app_iter=[b""], # don't recalculate content-length **kwargs )
[docs] def prepare(self, environ): """ No contents for HEAD request. """
[docs]def wps_restapi_base_path(container): # type: (AnySettingsContainer) -> str settings = get_settings(container) restapi_path = settings.get("weaver.wps_restapi_path", "").rstrip("/").strip() return restapi_path
[docs]def get_wps_restapi_base_url(container): # type: (AnySettingsContainer) -> str settings = get_settings(container) weaver_rest_url = settings.get("weaver.wps_restapi_url") if not weaver_rest_url: weaver_url = get_weaver_url(settings) restapi_path = wps_restapi_base_path(settings) weaver_rest_url = weaver_url + restapi_path return weaver_rest_url.rstrip("/").strip()
[docs]def get_schema_ref(schema, container=None, ref_type="$schema", ref_name=True): # type: (colander.SchemaNode, Optional[AnySettingsContainer], str, True) -> Dict[str, str] """ Generates the JSON OpenAPI schema reference relative to the current `Weaver` instance. The provided schema should be one of the items listed in ``#/definitions`` of the ``/json`` endpoint. No validation is accomplished to avoid long processing of all references. If setting ``weaver.schema_url`` is set, this value will be used direct as fully-defined base URL. This could be used to refer to a static endpoint where schemas are hosted. Otherwise, the current Web Application resolved location is employed with JSON OpenAPI path. :param schema: schema-node instance or type for which to generate the OpenAPI reference. :param container: application settings to retrieve the base URL of the schema location. :param ref_type: key employed to form the reference (e.g.: "$schema", "$ref", "@schema", etc.) :param ref_name: indicate if the plain name should also be included under field ``"schema"``. :return: OpenAPI schema reference """ is_instance = isinstance(schema, colander.SchemaNode) assert is_instance or (inspect.isclass(schema) and issubclass(schema, colander.SchemaNode)) if is_instance: schema = type(schema) schema_name = schema.__name__ settings = get_settings(container) weaver_schema_url = settings.get("weaver.schema_url") if not weaver_schema_url: restapi_path = get_wps_restapi_base_url(container) weaver_schema_url = f"{restapi_path}{sd.openapi_json_service.path}#/definitions" weaver_schema_url = weaver_schema_url.rstrip("/").strip() schema_ref = {ref_type: f"{weaver_schema_url}/{schema_name}"} if ref_name: schema_ref.update({"schema": schema_name}) return schema_ref
[docs]def handle_schema_validation(schema=None): # type: (Optional[colander.SchemaNode]) -> Callable """ Convert a schema validation error into an HTTP error with error details about the failure. :param schema: If provided, document this schema as the reference of the failed schema validation. :raises HTTPBadRequest: If any schema validation error occurs when handling the decorated function. """ def decorator(func): @functools.wraps(func) def wrapped(*args, **kwargs): try: return func(*args, **kwargs) except colander.Invalid as ex: data = { "type": "InvalidSchema", "detail": "Invalid value failed schema validation.", "error": colander.Invalid.__name__, "cause": ex.asdict(), "value": ex.value, } if schema: data.update({ "schema": get_schema_ref(schema) }) raise HTTPBadRequest(json=data) return wrapped return decorator