Files
connexion/connexion/context.py
Robbe Sneyders b9ba13cde5 Centralize error handling in ExceptionMiddleware (#1754)
I was writing the documentation on exception handling, and I noticed
that it was very hard to explain our current behavior.

Error handlers can be registered either on the internal Flask app (not
the Starlette one) or on the Connexion app, which leads to some
undefined (actually just really hard to explain) behavior. Eg.
- Registering error handlers on a status code would capture
`starlette.HTTPException` errors on the Connexion app, and
`werkzeug.HTTPException` errors on the Flask App, which means that
registering an error handler on a status code doesn't catch all the
errors with that status code.
- Flask does some default error handling which leads to some exceptions
never reaching the error handlers registered on the Connexion app.

So I made the following changes:
- Replaced the default error handlers we registered on the Flask app
with a default handler on the `ExceptionMiddleware` that takes into
account other handlers registered on status codes.
- Configured Flask to propagate exceptions instead of catching them.
- Abstracted away the Starlette `Request` and `Response` types, so users
can and must now use `ConnexionRequest`
  and `ConnexionResponse` types in error handlers.
- Renamed the `ASGIRequest` class to `ConnexionRequest` since it is the
only Request class part of the high level
  Connexion interface.

We could also rename `ConnexionRequest` and `ConnexionResponse` to just
`Request` and `Response`. Wdyt?
2023-10-29 09:37:54 +01:00

30 lines
991 B
Python

from contextvars import ContextVar
from starlette.types import Receive, Scope
from werkzeug.local import LocalProxy
from connexion.lifecycle import ConnexionRequest
from connexion.operations import AbstractOperation
UNBOUND_MESSAGE = (
"Working outside of operation context. Make sure your app is wrapped in a "
"ContextMiddleware and you're processing a request while accessing the context."
)
_context: ContextVar[dict] = ContextVar("CONTEXT")
context = LocalProxy(_context, unbound_message=UNBOUND_MESSAGE)
_operation: ContextVar[AbstractOperation] = ContextVar("OPERATION")
operation = LocalProxy(_operation, unbound_message=UNBOUND_MESSAGE)
_receive: ContextVar[Receive] = ContextVar("RECEIVE")
receive = LocalProxy(_receive, unbound_message=UNBOUND_MESSAGE)
_scope: ContextVar[Scope] = ContextVar("SCOPE")
scope = LocalProxy(_scope, unbound_message=UNBOUND_MESSAGE)
request = LocalProxy(
lambda: ConnexionRequest(scope, receive), unbound_message=UNBOUND_MESSAGE
)