Source code for weaver.owsexceptions

OWSExceptions are based on pyramid.httpexceptions.

See also:
import json
import warnings
from string import Template
from typing import TYPE_CHECKING, AnyStr

import six
from pyramid.compat import text_type
from pyramid.httpexceptions import (
from pyramid.interfaces import IExceptionResponse
from pyramid.response import Response
from webob import html_escape as _html_escape
from webob.acceptparse import create_accept_header
from zope.interface import implementer

from weaver.utils import clean_json_text_body
from weaver.warning import MissingParameterWarning, UnsupportedOperationWarning

    from weaver.typedefs import JSON, SettingsType  # noqa: F401

[docs]@implementer(IExceptionResponse) class OWSException(Response, Exception):
[docs] code = "NoApplicableCode"
[docs] value = None
[docs] locator = "NoApplicableCode"
[docs] explanation = "Unknown Error"
[docs] page_template = Template("""\ <?xml version="1.0" encoding="utf-8"?> <ExceptionReport version="1.0.0" xmlns="" xmlns:xsi="" xsi:schemaLocation=""> <Exception exceptionCode="${code}" locator="${locator}"> <ExceptionText>${message}</ExceptionText> </Exception> </ExceptionReport>""")
def __init__(self, detail=None, value=None, **kw): status = kw.pop("status", None) if isinstance(status, type) and issubclass(status, HTTPException): status = status().status elif isinstance(status, six.class_types): try: int(status.split()[0]) except Exception: raise ValueError("status specified as string must be of format '<code> <title>'") elif isinstance(status, HTTPException): status = status.status elif not status: status = HTTPOk().status Response.__init__(self, status=status, **kw) Exception.__init__(self, detail) self.message = detail or self.explanation self.content_type = CONTENT_TYPE_TEXT_XML if value: self.locator = value def __str__(self, skip_body=False): return self.message def __repr__(self): if self.message: return "{}{}".format(type(self), self.message) return str(type(self)) @staticmethod
[docs] def json_formatter(status, body, title, environ): # noqa: F811 # type: (AnyStr, AnyStr, AnyStr, SettingsType) -> JSON body = clean_json_text_body(body) return {"description": body, "code": int(status.split()[0]), "status": status, "title": title}
[docs] def prepare(self, environ): if not self.body: accept_value = environ.get("HTTP_ACCEPT", "") accept = create_accept_header(accept_value) # Attempt to match xml or json, if those don't match, we will fall through to defaulting to xml match = accept.best_match([CONTENT_TYPE_TEXT_XML, CONTENT_TYPE_APP_JSON]) if match == CONTENT_TYPE_APP_JSON: self.content_type = CONTENT_TYPE_APP_JSON # json exception response should not have status 200 if self.status_code == HTTPOk.code: self.status = HTTPInternalServerError.code class JsonPageTemplate(object): def __init__(self, excobj): self.excobj = excobj def substitute(self, code, locator, message): # noqa: W0613 return json.dumps(self.excobj.json_formatter( status=self.excobj.status, body=message, title=None, environ=environ)) page_template = JsonPageTemplate(self) else: self.content_type = CONTENT_TYPE_TEXT_XML page_template = self.page_template args = { "code": _html_escape(self.code), "locator": _html_escape(self.locator), "message": _html_escape(self.message or ""), } page = page_template.substitute(**args) if isinstance(page, text_type): page = page.encode(self.charset if self.charset else "UTF-8") self.app_iter = [page] self.body = page
[docs] def wsgi_response(self): # bw compat only return self
[docs] exception = wsgi_response # bw compat only
def __call__(self, environ, start_response): # differences from webob.exc.WSGIHTTPException # # - does not try to deal with HEAD requests # # - does not manufacture a new response object when generating # the default response # self.prepare(environ) return Response.__call__(self, environ, start_response)
[docs]class OWSAccessForbidden(OWSException):
[docs] locator = "AccessUnauthorized"
[docs] explanation = "Access to this service is unauthorized."
def __init__(self, *args, **kwargs): kwargs["status"] = HTTPUnauthorized super(OWSAccessForbidden, self).__init__(*args, **kwargs)
[docs]class OWSNotFound(OWSException):
[docs] locator = "NotFound"
[docs] explanation = "This resource does not exist."
def __init__(self, *args, **kwargs): kwargs["status"] = HTTPNotFound super(OWSNotFound, self).__init__(*args, **kwargs)
[docs]class OWSNotAcceptable(OWSException):
[docs] locator = "NotAcceptable"
[docs] explanation = "Access to this service failed."
def __init__(self, *args, **kwargs): kwargs["status"] = HTTPNotAcceptable super(OWSNotAcceptable, self).__init__(*args, **kwargs)
[docs]class OWSNoApplicableCode(OWSException): """WPS Bad Request Exception"""
[docs] code = "NoApplicableCode"
[docs] locator = ""
[docs] explanation = "Parameter value is missing"
def __init__(self, *args, **kwargs): kwargs["status"] = HTTPBadRequest super(OWSNoApplicableCode, self).__init__(*args, **kwargs) warnings.warn(self.message, UnsupportedOperationWarning)
[docs]class OWSMissingParameterValue(OWSException): """MissingParameterValue WPS Exception"""
[docs] code = "MissingParameterValue"
[docs] locator = ""
[docs] explanation = "Parameter value is missing"
def __init__(self, *args, **kwargs): kwargs["status"] = HTTPBadRequest super(OWSMissingParameterValue, self).__init__(*args, **kwargs) warnings.warn(self.message, MissingParameterWarning)
[docs]class OWSInvalidParameterValue(OWSException): """InvalidParameterValue WPS Exception"""
[docs] code = "InvalidParameterValue"
[docs] locator = ""
[docs] explanation = "Parameter value is not acceptable."
def __init__(self, *args, **kwargs): kwargs["status"] = HTTPNotAcceptable super(OWSInvalidParameterValue, self).__init__(*args, **kwargs) warnings.warn(self.message, UnsupportedOperationWarning)
[docs]class OWSNotImplemented(OWSException):
[docs] code = "NotImplemented"
[docs] locator = ""
[docs] explanation = "Operation is not implemented."
def __init__(self, *args, **kwargs): kwargs["status"] = HTTPNotImplemented super(OWSNotImplemented, self).__init__(*args, **kwargs) warnings.warn(self.message, UnsupportedOperationWarning)