Errors & Exception Views

from pyramid.config import Configurator
from pyramid.response import Response
from webtest import TestApp


class BookshelfError(Exception):
    status = 500

    def __init__(self, message):
        super().__init__(message)
        self.message = message


class ShelfFull(BookshelfError):
    status = 409


def add_book(request):
    raise ShelfFull("shelf 1 holds at most 2 books")


def crash(request):
    raise BookshelfError("unexpected")


def bookshelf_error(exc, request):
    # the exception arrives as `context`; with a (request) signature it is request.exception
    request.response.status = exc.status
    return {"error": type(exc).__name__, "message": exc.message}


def shelf_full_get(request):
    return Response("full (GET)", status=409)


with Configurator() as config:
    config.add_route("add", "/add")
    config.add_route("crash", "/crash")
    config.add_view(add_book, route_name="add")
    config.add_view(crash, route_name="crash")
    config.add_exception_view(bookshelf_error, context=BookshelfError, renderer="json")
    # same family, but only for GET: more predicates make a view more specific
    config.add_exception_view(shelf_full_get, context=ShelfFull, request_method="GET")
    client = TestApp(config.make_wsgi_app())

print(client.get("/add", status=409).text)
print(client.post("/add", status=409).json)
print(client.get("/crash", status=500).json)

# norun  (decorators are collected by config.scan() from an importable module)
from pyramid.view import exception_view_config, forbidden_view_config, notfound_view_config


@exception_view_config(BookshelfError, renderer="json")
def bookshelf_error(exc, request):
    request.response.status = exc.status
    return {"error": type(exc).__name__, "message": exc.message}


@notfound_view_config(append_slash=True)
def not_found(request): ...


@forbidden_view_config()
def forbidden(request): ...

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPMovedPermanently, HTTPNotFound
from pyramid.response import Response
from webtest import TestApp


def list_shelves(request):
    return Response("shelves")


def post_only(request):
    return Response("posted")


def not_found(request):
    return Response(f"No such page: {request.path}", status=404)


def not_found_post(request):
    return Response("No such POST target", status=404)


with Configurator() as config:
    config.add_route("shelves", "/shelves/")
    config.add_route("post", "/post")
    config.add_view(list_shelves, route_name="shelves")
    config.add_view(post_only, route_name="post", request_method="POST")
    # default redirect class (307)
    config.add_notfound_view(not_found, append_slash=True, request_method="GET")
    # predicates narrow a Not Found View exactly as they narrow a view
    config.add_notfound_view(not_found_post, request_method="POST")
    client = TestApp(config.make_wsgi_app())

print(client.get("/shelves", status=307).location)
print(client.get("/nothing", status=404).text)
print(client.get("/post", status=404).text)  # GET hits a POST-only view: predicate mismatch
print(client.post("/elsewhere", status=404).text)


with Configurator() as config:
    config.add_route("shelves", "/shelves/")
    config.add_view(list_shelves, route_name="shelves")
    config.add_notfound_view(append_slash=HTTPMovedPermanently)  # the view argument is optional
    client = TestApp(config.make_wsgi_app())
print(client.get("/shelves", status=301).location)

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPFound
from pyramid.response import Response
from pyramid.security import Allowed, Denied
from webtest import TestApp


class HeaderPolicy:
    """Demo policy: the user is whatever the X-User header says. Never do this in production."""

    def identity(self, request):
        return request.headers.get("X-User")

    def authenticated_userid(self, request):
        return self.identity(request)

    def permits(self, request, context, permission):
        if self.identity(request) == "ada":
            return Allowed("ada may do anything")
        return Denied(f"{permission!r} needs the owner")

    def remember(self, request, userid, **kw):
        return []

    def forget(self, request, **kw):
        return []


def edit_shelf(request):
    return Response("editing")


def login(request):
    return Response("login page")


def forbidden(request):
    if request.authenticated_userid is None:
        return HTTPFound(request.route_url("login", _query={"next": request.path}))
    return Response(f"Denied: {request.exception.result.msg}", status=403)


with Configurator() as config:
    config.set_security_policy(HeaderPolicy())
    config.add_route("edit", "/shelves/1/edit")
    config.add_route("login", "/login")
    config.add_view(edit_shelf, route_name="edit", permission="edit")
    config.add_view(login, route_name="login")
    config.add_forbidden_view(forbidden)
    client = TestApp(config.make_wsgi_app())

print(client.get("/shelves/1/edit", status=302).location)
print(client.get("/shelves/1/edit", headers={"X-User": "grace"}, status=403).text)
print(client.get("/shelves/1/edit", headers={"X-User": "ada"}).text)

import logging

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPBadRequest, HTTPError, HTTPFound
from pyramid.response import Response
from webtest import TestApp

log = logging.getLogger("bookshelf")


def problem(request, status, title, detail=None):
    request.response.status = status
    request.response.content_type = "application/problem+json"
    return {
        "type": "about:blank",
        "title": title,
        "status": status,
        "detail": detail,
        "instance": request.path,
    }


def api_shelf(request):
    if request.matchdict["id"] == "0":
        raise HTTPBadRequest("id must be positive")
    raise RuntimeError("database exploded at /var/db")  # an internal detail, must not leak


def api_move(request):
    raise HTTPFound("/api/shelves/1")


def html_home(request):
    raise RuntimeError("html side")


# HTTPError covers 4xx and 5xx; HTTPException would also catch redirects such as HTTPFound
def api_http_error(exc, request):
    return problem(request, exc.code, exc.title, exc.detail)


def api_not_found(request):
    return problem(request, 404, "Not Found", "No such API resource")


def api_crash(exc, request):
    log.exception("unhandled error on %s", request.path)  # details go to the log only
    return problem(request, 500, "Internal Server Error")


def html_crash(exc, request):
    return Response("Something went wrong", status=500)


with Configurator() as config:
    config.add_route("api_shelf", "/api/shelves/{id}")
    config.add_route("api_move", "/api/move")
    config.add_route("home", "/")
    config.add_view(api_shelf, route_name="api_shelf", renderer="json")
    config.add_view(api_move, route_name="api_move")
    config.add_view(html_home, route_name="home")
    config.add_exception_view(api_http_error, context=HTTPError, path_info=r"^/api/", renderer="json")
    config.add_notfound_view(api_not_found, path_info=r"^/api/", renderer="json")
    config.add_exception_view(api_crash, path_info=r"^/api/", renderer="json")
    config.add_exception_view(html_crash)
    client = TestApp(config.make_wsgi_app())

logging.disable(logging.CRITICAL)  # keep the demo output short; the traceback goes to the logger
for path, status in (("/api/shelves/0", 400), ("/api/shelves/13", 500), ("/api/nothing", 404), ("/", 500)):
    response = client.get(path, status=status)
    print(response.status_int, response.content_type, response.text.strip()[:90])
print(client.get("/api/move", status=302).location)

import transaction
from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPBadRequest
from pyramid.response import Response
from transaction.interfaces import IDataManager, TransientError
from webtest import TestApp
from zope.interface import implementer

EVENTS = []


@implementer(IDataManager)
class RecordingDataManager:
    """A stand-in for a database connection that records how the transaction ended."""

    transaction_manager = transaction.manager

    def __init__(self, label):
        self.label = label

    def abort(self, txn):
        EVENTS.append(f"{self.label}: abort")

    def tpc_begin(self, txn): ...
    def commit(self, txn): ...
    def tpc_vote(self, txn): ...

    def tpc_finish(self, txn):
        EVENTS.append(f"{self.label}: commit")

    def tpc_abort(self, txn):
        EVENTS.append(f"{self.label}: abort")

    def sortKey(self):
        return self.label


def saved(request):
    request.tm.get().join(RecordingDataManager("saved"))
    return Response("saved")


def raised(request):
    request.tm.get().join(RecordingDataManager("raised"))
    raise HTTPBadRequest("rejected")


def returned(request):
    request.tm.get().join(RecordingDataManager("returned"))
    return HTTPBadRequest("rejected")


def while_open(request):
    EVENTS.append("exception view ran inside the transaction")
    return Response("bad", status=400)


def build(settings=None):
    with Configurator(settings=settings) as config:
        config.include("pyramid_tm")
        for name, view in (("saved", saved), ("raised", raised), ("returned", returned)):
            config.add_route(name, f"/{name}")
            config.add_view(view, route_name=name)
        config.add_exception_view(while_open, context=HTTPBadRequest, tm_active=True)
        return TestApp(config.make_wsgi_app())


client = build()
client.get("/saved")
client.get("/raised", status=400)
client.get("/returned", status=400)
print(*EVENTS, sep="\n")

EVENTS.clear()
client = build({"tm.commit_veto": "pyramid_tm.default_commit_veto"})
client.get("/returned", status=400)
print(*EVENTS, sep="\n")

from pyramid.config import Configurator
from pyramid.response import Response
from transaction.interfaces import TransientError
from webtest import TestApp

ATTEMPTS = []


def flaky(request):
    ATTEMPTS.append(request.environ["retry.attempt"])
    raise TransientError("serialization failure")


def give_up(request):
    return Response("busy, try again later", status=503)


with Configurator(settings={"retry.attempts": "3"}) as config:
    config.include("pyramid_retry")
    config.include("pyramid_tm")
    config.add_route("flaky", "/flaky")
    config.add_view(flaky, route_name="flaky")
    config.add_exception_view(give_up, context=TransientError, last_retry_attempt=True)
    client = TestApp(config.make_wsgi_app())

response = client.get("/flaky", status=503)
print(ATTEMPTS, response.text)

import sys

from pyramid.config import Configurator
from pyramid.response import Response
from webtest import TestApp


class Boom(Exception):
    pass


def start(request):
    try:
        raise Boom()
    except Boom:
        # render through the registered exception views, from application code
        return request.invoke_exception_view(sys.exc_info(), reraise=True)


def nice(request):
    return Response(f"handled {type(request.exception).__name__}", status=418)


with Configurator() as config:
    config.add_route("start", "/start")
    config.add_view(start, route_name="start")
    config.add_exception_view(nice, context=Boom)
    client = TestApp(config.make_wsgi_app())

print(client.get("/start", status=418).text)

import pytest
from pyramid import testing
from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPNotFound
from webtest import TestApp
from webtest.app import AppError


def show_shelf(request):
    raise HTTPNotFound("no shelf")


# unit test: the view raises, no exception view is involved
with pytest.raises(HTTPNotFound):
    show_shelf(testing.DummyRequest())

# integration test: the exception view's response and status are checked
with Configurator() as config:
    config.add_route("shelf", "/shelves/1")
    config.add_view(show_shelf, route_name="shelf")
    client = TestApp(config.make_wsgi_app())

client.get("/shelves/1", status=404)  # passes: the 404 is the expectation
try:
    client.get("/shelves/1")  # an unexpected 404 fails the test
except AppError as error:
    print("unexpected status:", str(error).splitlines()[0][:50])