"""
OWSExceptions definitions.
Exceptions are based on :mod:`pyramid.httpexceptions` and :mod:`pywps.exceptions` to handle more cases where they can
be caught whether the running process is via :mod:`weaver` or through :mod:`pywps` service.
Furthermore, interrelation with :mod:`weaver.exceptions` classes (with base :exc:`weaver.exceptions.WeaverException`)
also employ specific :exc:`OWSExceptions` definitions to provide specific error details.
"""
import json as json_pkg
import warnings
from string import Template
from typing import TYPE_CHECKING
from pyramid.httpexceptions import (
HTTPBadRequest,
HTTPException,
HTTPForbidden,
HTTPGone,
HTTPInternalServerError,
HTTPNotAcceptable,
HTTPNotFound,
HTTPNotImplemented,
HTTPOk
)
from pyramid.interfaces import IExceptionResponse
from pyramid.response import Response
from pywps.exceptions import InvalidParameterValue, MissingParameterValue, NoApplicableCode
from webob.acceptparse import create_accept_header
from zope.interface import implementer
from weaver.formats import ContentType
from weaver.warning import MissingParameterWarning, UnsupportedOperationWarning
if TYPE_CHECKING:
from typing import Any, Optional
from weaver.typedefs import JSON, SettingsType
@implementer(IExceptionResponse)
[docs]
class OWSException(Response, Exception):
"""
Base OWS Exception definition.
"""
[docs]
code = "NoApplicableCode"
[docs]
locator = "NoApplicableCode"
[docs]
description = "Unknown Error"
[docs]
page_template = Template("""\
<?xml version="1.0" encoding="utf-8"?>
<ExceptionReport version="1.0.0"
xmlns="http://www.opengis.net/ows/1.1"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.opengis.net/ows/1.1 http://schemas.opengis.net/ows/1.1.0/owsExceptionReport.xsd">
<Exception exceptionCode="${code}" locator="${locator}">
<ExceptionText>${message}</ExceptionText>
</Exception>
</ExceptionReport>""")
def __init__(self, detail=None, value=None, json=None, **kw):
# type: (Optional[str], Optional[Any], Optional[JSON], **Any) -> None
status = kw.pop("status", None)
if isinstance(status, type) and issubclass(status, HTTPException):
status = status().status
elif isinstance(status, str):
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
locator = kw.get("locator")
if isinstance(json, dict):
detail = detail or json.get("detail") or json.get("description")
locator = locator or json.get("locator") or json.get("name")
if value:
json.setdefault("value", value)
self.code = str(kw.pop("code", self.code))
desc = str(detail or kw.pop("description", self.description))
if json is not None:
kw.update({"json": json})
Response.__init__(self, status=status, **kw)
Exception.__init__(self, detail)
[docs]
self.message = detail or self.description or getattr(self, "explanation", None)
[docs]
self.content_type = ContentType.APP_JSON
if locator:
self.locator = locator
try:
json_desc = self.json.get("description")
if json_desc:
desc = json_desc
else:
self.json["description"] = desc
except Exception: # noqa: W0703 # nosec: B110
pass
self.description = desc
def __str__(self, skip_body=False):
return self.message
def __repr__(self):
if self.message:
return f"{type(self)}{self.message}"
return str(type(self))
@staticmethod
[docs]
def prepare(self, environ):
if not self.body: # pylint: disable=E0203,W0201
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 back to defaulting to JSON
# since browsers add HTML automatically, and it is closer to XML, we 'allow' it only to catch this
# explicit case and fallback to JSON manually
matches = accept.acceptable_offers([ContentType.TEXT_HTML, ContentType.APP_JSON,
ContentType.TEXT_XML, ContentType.APP_XML])
match = matches[0] if matches else ContentType.APP_JSON
if match == ContentType.TEXT_HTML:
match = ContentType.APP_JSON
if match == ContentType.APP_JSON:
self.content_type = ContentType.APP_JSON
# json exception response should not have status 200
if self.status_code == HTTPOk.code:
self.status = HTTPInternalServerError.code # pylint: disable=E0203,W0201
class JsonPageTemplate(object):
def __init__(self, excobj):
self.excobj = excobj
def substitute(self, code, locator, message):
status = self.excobj.status
title = getattr(self.excobj, "code", None)
data = self.excobj.json_formatter(status=status, body=message, title=title, environ=environ)
data["exception"] = {
"code": code or "",
"locator": locator or "",
"message": message or "",
}
return json_pkg.dumps(data)
page_template = JsonPageTemplate(self)
args = {"code": self.code, "locator": self.locator, "message": self.message}
else:
self.content_type = ContentType.TEXT_XML
page_template = self.page_template
args = {
"code": self.code,
"locator": self.locator,
"message": self.message or "",
}
page = page_template.substitute(**args)
if isinstance(page, str):
page = page.encode(self.charset if self.charset else "UTF-8")
self.app_iter = [page] # pylint: disable=E0203,W0201
self.body = page # pylint: disable=E0203,W0201
@property
[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]
code = "AccessForbidden"
[docs]
explanation = "Access to this service is forbidden."
def __init__(self, *args, **kwargs):
kwargs["status"] = HTTPForbidden
super(OWSAccessForbidden, self).__init__(*args, **kwargs)
[docs]
class OWSNotFound(OWSException):
[docs]
explanation = "Resource does not exist."
def __init__(self, *args, **kwargs):
kwargs["status"] = HTTPNotFound
super(OWSNotFound, self).__init__(*args, **kwargs)
[docs]
class OWSGone(OWSException):
[docs]
explanation = "Resource is gone."
def __init__(self, *args, **kwargs):
kwargs["status"] = HTTPGone
super(OWSGone, self).__init__(*args, **kwargs)
[docs]
class OWSNotAcceptable(OWSException):
[docs]
explanation = "Cannot produce requested Accept format."
def __init__(self, *args, **kwargs):
kwargs["status"] = HTTPNotAcceptable
super(OWSNotAcceptable, self).__init__(*args, **kwargs)
[docs]
class OWSNoApplicableCode(OWSException, NoApplicableCode):
"""
WPS Bad Request Exception.
"""
[docs]
code = "NoApplicableCode"
[docs]
explanation = "Undefined error"
def __init__(self, *args, **kwargs):
kwargs["status"] = HTTPInternalServerError
super(OWSNoApplicableCode, self).__init__(*args, **kwargs)
warnings.warn(self.message, UnsupportedOperationWarning)
[docs]
class OWSMissingParameterValue(OWSException, MissingParameterValue):
"""
MissingParameterValue WPS Exception.
"""
[docs]
code = "MissingParameterValue"
[docs]
description = "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):
"""
InvalidParameterValue WPS Exception.
"""
[docs]
code = "InvalidParameterValue"
[docs]
description = "Parameter value is not acceptable."
def __init__(self, *args, **kwargs):
kwargs["status"] = HTTPBadRequest
super(OWSInvalidParameterValue, self).__init__(*args, **kwargs)
warnings.warn(self.message, UnsupportedOperationWarning)
[docs]
class OWSNotImplemented(OWSException):
[docs]
code = "NotImplemented"
[docs]
description = "Operation is not implemented."
def __init__(self, *args, **kwargs):
kwargs["status"] = HTTPNotImplemented
super(OWSNotImplemented, self).__init__(*args, **kwargs)
warnings.warn(self.message, UnsupportedOperationWarning)