Source code for cowbird.api.generic
from typing import Callable, Optional, Tuple, Union, cast
from pyramid.exceptions import PredicateMismatch
from pyramid.httpexceptions import (
HTTPException,
HTTPInternalServerError,
HTTPMethodNotAllowed,
HTTPNotAcceptable,
HTTPNotFound,
HTTPServerError
)
from pyramid.registry import Registry
from pyramid.request import Request
from pyramid.response import Response
from simplejson import JSONDecodeError
from cowbird.api import exception as ax
from cowbird.api import schemas as s
from cowbird.typedefs import JSON, AnyResponseType
from cowbird.utils import (
CONTENT_TYPE_ANY,
CONTENT_TYPE_JSON,
FORMAT_TYPE_MAPPING,
SUPPORTED_ACCEPT_TYPES,
get_header,
get_logger
)
[docs]
class RemoveSlashNotFoundViewFactory(object):
"""
Utility that will try to resolve a path without appended slash if one was provided.
"""
def __init__(self, notfound_view: Optional[Callable[[Request], AnyResponseType]] = None) -> None:
self.notfound_view = notfound_view
[docs]
def __call__(self, request: Request) -> AnyResponseType:
from pyramid.httpexceptions import HTTPMovedPermanently
from pyramid.interfaces import IRoutesMapper
path = request.path
registry = cast(Registry, request.registry) # pyramid improperly reports Type[Registry] instead of the instance
mapper = registry.queryUtility(IRoutesMapper)
if mapper is not None and path.endswith("/"):
no_slash_path = path.rstrip("/")
no_slash_path = no_slash_path.split("/cowbird", 1)[-1]
for route in mapper.get_routes():
if route.match(no_slash_path) is not None:
query = request.query_string
if query:
no_slash_path += "?" + query
return HTTPMovedPermanently(location=no_slash_path)
return self.notfound_view(request)
[docs]
def internal_server_error(request: Request) -> HTTPException:
"""
Overrides default HTTP.
"""
content = get_request_info(request, exception_details=True,
default_message=s.InternalServerErrorResponseSchema.description)
detail: str = content["detail"]
return ax.raise_http(nothrow=True, http_error=HTTPInternalServerError, detail=detail, content=content)
[docs]
def not_found_or_method_not_allowed(request: Request) -> HTTPException:
"""
Overrides the default ``HTTPNotFound`` [404] by appropriate ``HTTPMethodNotAllowed`` [405] when applicable.
Not found response can correspond to underlying process operation not finding a required item, or a completely
unknown route (path did not match any existing API definition).
Method not allowed is more specific to the case where the path matches an existing API route, but the specific
request method (GET, POST, etc.) is not allowed on this path.
Without this fix, both situations return [404] regardless.
"""
if isinstance(request.exception, PredicateMismatch) and request.method not in ["HEAD", "GET"]:
http_err = HTTPMethodNotAllowed
http_msg = "" # auto-generated by HTTPMethodNotAllowed
else:
http_err = HTTPNotFound
http_msg = s.NotFoundResponseSchema.description
content = get_request_info(request, default_message=http_msg)
detail: str = content["detail"]
return ax.raise_http(nothrow=True, http_error=http_err, detail=detail, content=content)
[docs]
def guess_target_format(request: Request) -> Tuple[str, bool]:
"""
Guess the best applicable response ``Content-Type`` header according to request ``Accept`` header and ``format``
query, or defaulting to :py:data:`CONTENT_TYPE_JSON`.
:returns: tuple of matched MIME-type and where it was found (``True``: header, ``False``: query)
"""
content_type = FORMAT_TYPE_MAPPING.get(request.params.get("format"))
is_header = False
if not content_type:
is_header = True
content_type = get_header("accept", request.headers, default=CONTENT_TYPE_JSON, split=";,")
if content_type != CONTENT_TYPE_JSON:
# because most browsers enforce some 'visual' list of accept header, revert to JSON if detected
# explicit request set by other client (e.g.: using 'requests') will have full control over desired content
user_agent = get_header("user-agent", request.headers)
if user_agent and any(browser in user_agent for browser in ["Mozilla", "Chrome", "Safari"]):
content_type = CONTENT_TYPE_JSON
if not content_type or content_type == CONTENT_TYPE_ANY:
is_header = True
content_type = CONTENT_TYPE_JSON
return content_type, is_header
[docs]
def validate_accept_header_tween(handler: Callable[[Request], Response],
registry: Registry, # noqa: F811
) -> Callable[[Request], Response]:
"""
Tween that validates that the specified request ``Accept`` header or ``format`` query (if any) is supported by the
application and for the given context.
:raises HTTPNotAcceptable: if desired ``Content-Type`` is not supported.
"""
def validate_format(request: Request) -> Response:
"""
Validates the specified request according to its ``Accept`` header or ``format`` query, ignoring UI related
routes that require more content-types than the ones supported by the API for displaying purposes of other
elements (styles, images, etc.).
"""
accept, _ = guess_target_format(request)
http_msg = s.NotAcceptableResponseSchema.description
content = get_request_info(request, default_message=http_msg)
ax.verify_param(accept, is_in=True, param_compare=SUPPORTED_ACCEPT_TYPES,
param_name="Accept Header or Format Query",
http_error=HTTPNotAcceptable, msg_on_fail=http_msg,
content=content, content_type=CONTENT_TYPE_JSON) # enforce type to avoid recursion
return handler(request)
return validate_format
[docs]
def apply_response_format_tween(handler: Callable[[Request], HTTPException],
registry: Registry, # noqa: F811
) -> Callable[[Request], Response]:
"""
Tween that obtains the request ``Accept`` header or ``format`` query (if any) to generate the response with the
desired ``Content-Type``.
The target ``Content-Type`` is expected to have been validated by :func:`validate_accept_header_tween` beforehand
to handle not-acceptable errors.
The tween also ensures that additional request metadata extracted from :func:`get_request_info` is applied to
the response body if not already provided by a previous operation.
"""
def apply_format(request: Request) -> HTTPException:
"""
Validates the specified request according to its ``Accept`` header, ignoring UI related routes that request more
content-types than the ones supported by the application for display purposes (styles, images etc.).
Alternatively, if no ``Accept`` header is found, look for equivalent value provided via query parameter.
"""
# all API routes expected to either call 'valid_http' or 'raise_http' of 'cowbird.api.exception' module
# an HTTPException is always returned, and content is a JSON-like string
content_type, is_header = guess_target_format(request)
if not is_header:
# NOTE:
# enforce the accept header in case it was specified with format query, since some renderer implementations
# will afterward erroneously overwrite the 'content-type' value that we enforce when converting the response
# from the HTTPException. See:
# - https://github.com/Pylons/webob/issues/204
# - https://github.com/Pylons/webob/issues/238
# - https://github.com/Pylons/pyramid/issues/1344
request.accept = content_type
resp = handler(request) # no exception when EXCVIEW tween is placed under this tween
# return routes already converted (valid_http/raise_http where not used, pyramid already generated response)
if not isinstance(resp, HTTPException):
return resp
# forward any headers such as session cookies to be applied
metadata = get_request_info(request)
resp_kwargs = {"headers": resp.headers}
return ax.generate_response_http_format(type(resp), resp_kwargs, resp.text, content_type, metadata)
return apply_format
[docs]
def get_exception_info(response: Union[HTTPException, Request, Response],
content: Optional[JSON] = None,
exception_details: bool = False,
) -> JSON:
"""
Obtains additional exception content details about the :paramref:`response` according to available information.
"""
content = content or {}
if hasattr(response, "exception"):
# handle error raised simply by checking for "json" property in python 3 when body is invalid
has_json = False
try:
has_json = hasattr(response.exception, "json")
except JSONDecodeError:
pass
if has_json and isinstance(response.exception.json, dict):
content.update(response.exception.json)
elif isinstance(response.exception, HTTPServerError) and hasattr(response.exception, "message"):
content.update({"exception": str(response.exception.message)})
elif isinstance(response.exception, Exception) and exception_details:
content.update({"exception": type(response.exception).__name__})
# get 'request.exc_info' or 'sys.exc_info', whichever one is available
LOGGER.error("Request exception.", exc_info=getattr(response, "exc_info", True))
if not content.get("detail"):
detail = response.exception
content["detail"] = str(detail) if detail is not None else None
elif hasattr(response, "matchdict"):
if response.matchdict is not None and response.matchdict != "":
content.update(response.matchdict)
return content
[docs]
def get_request_info(request: Union[Request, HTTPException],
default_message: Optional[str] = None,
exception_details: bool = False,
) -> JSON:
"""
Obtains additional content details about the :paramref:`request` according to available information.
"""
content: JSON = {
"path": str(request.upath_info),
"url": str(request.url),
"detail": default_message,
"method": request.method
}
exc_info = get_exception_info(request, content=content, exception_details=exception_details)
content.update(exc_info) # type: ignore[arg-type]
return content