123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513 |
- import sys
- import typing as t
- from functools import update_wrapper
- from types import TracebackType
- from werkzeug.exceptions import HTTPException
- from .globals import _app_ctx_stack
- from .globals import _request_ctx_stack
- from .signals import appcontext_popped
- from .signals import appcontext_pushed
- from .typing import AfterRequestCallable
- if t.TYPE_CHECKING:
- from .app import Flask
- from .sessions import SessionMixin
- from .wrappers import Request
- # a singleton sentinel value for parameter defaults
- _sentinel = object()
- class _AppCtxGlobals:
- """A plain object. Used as a namespace for storing data during an
- application context.
- Creating an app context automatically creates this object, which is
- made available as the :data:`g` proxy.
- .. describe:: 'key' in g
- Check whether an attribute is present.
- .. versionadded:: 0.10
- .. describe:: iter(g)
- Return an iterator over the attribute names.
- .. versionadded:: 0.10
- """
- # Define attr methods to let mypy know this is a namespace object
- # that has arbitrary attributes.
- def __getattr__(self, name: str) -> t.Any:
- try:
- return self.__dict__[name]
- except KeyError:
- raise AttributeError(name) from None
- def __setattr__(self, name: str, value: t.Any) -> None:
- self.__dict__[name] = value
- def __delattr__(self, name: str) -> None:
- try:
- del self.__dict__[name]
- except KeyError:
- raise AttributeError(name) from None
- def get(self, name: str, default: t.Optional[t.Any] = None) -> t.Any:
- """Get an attribute by name, or a default value. Like
- :meth:`dict.get`.
- :param name: Name of attribute to get.
- :param default: Value to return if the attribute is not present.
- .. versionadded:: 0.10
- """
- return self.__dict__.get(name, default)
- def pop(self, name: str, default: t.Any = _sentinel) -> t.Any:
- """Get and remove an attribute by name. Like :meth:`dict.pop`.
- :param name: Name of attribute to pop.
- :param default: Value to return if the attribute is not present,
- instead of raising a ``KeyError``.
- .. versionadded:: 0.11
- """
- if default is _sentinel:
- return self.__dict__.pop(name)
- else:
- return self.__dict__.pop(name, default)
- def setdefault(self, name: str, default: t.Any = None) -> t.Any:
- """Get the value of an attribute if it is present, otherwise
- set and return a default value. Like :meth:`dict.setdefault`.
- :param name: Name of attribute to get.
- :param default: Value to set and return if the attribute is not
- present.
- .. versionadded:: 0.11
- """
- return self.__dict__.setdefault(name, default)
- def __contains__(self, item: str) -> bool:
- return item in self.__dict__
- def __iter__(self) -> t.Iterator[str]:
- return iter(self.__dict__)
- def __repr__(self) -> str:
- top = _app_ctx_stack.top
- if top is not None:
- return f"<flask.g of {top.app.name!r}>"
- return object.__repr__(self)
- def after_this_request(f: AfterRequestCallable) -> AfterRequestCallable:
- """Executes a function after this request. This is useful to modify
- response objects. The function is passed the response object and has
- to return the same or a new one.
- Example::
- @app.route('/')
- def index():
- @after_this_request
- def add_header(response):
- response.headers['X-Foo'] = 'Parachute'
- return response
- return 'Hello World!'
- This is more useful if a function other than the view function wants to
- modify a response. For instance think of a decorator that wants to add
- some headers without converting the return value into a response object.
- .. versionadded:: 0.9
- """
- top = _request_ctx_stack.top
- if top is None:
- raise RuntimeError(
- "This decorator can only be used when a request context is"
- " active, such as within a view function."
- )
- top._after_request_functions.append(f)
- return f
- def copy_current_request_context(f: t.Callable) -> t.Callable:
- """A helper function that decorates a function to retain the current
- request context. This is useful when working with greenlets. The moment
- the function is decorated a copy of the request context is created and
- then pushed when the function is called. The current session is also
- included in the copied request context.
- Example::
- import gevent
- from flask import copy_current_request_context
- @app.route('/')
- def index():
- @copy_current_request_context
- def do_some_work():
- # do some work here, it can access flask.request or
- # flask.session like you would otherwise in the view function.
- ...
- gevent.spawn(do_some_work)
- return 'Regular response'
- .. versionadded:: 0.10
- """
- top = _request_ctx_stack.top
- if top is None:
- raise RuntimeError(
- "This decorator can only be used when a request context is"
- " active, such as within a view function."
- )
- reqctx = top.copy()
- def wrapper(*args, **kwargs):
- with reqctx:
- return reqctx.app.ensure_sync(f)(*args, **kwargs)
- return update_wrapper(wrapper, f)
- def has_request_context() -> bool:
- """If you have code that wants to test if a request context is there or
- not this function can be used. For instance, you may want to take advantage
- of request information if the request object is available, but fail
- silently if it is unavailable.
- ::
- class User(db.Model):
- def __init__(self, username, remote_addr=None):
- self.username = username
- if remote_addr is None and has_request_context():
- remote_addr = request.remote_addr
- self.remote_addr = remote_addr
- Alternatively you can also just test any of the context bound objects
- (such as :class:`request` or :class:`g`) for truthness::
- class User(db.Model):
- def __init__(self, username, remote_addr=None):
- self.username = username
- if remote_addr is None and request:
- remote_addr = request.remote_addr
- self.remote_addr = remote_addr
- .. versionadded:: 0.7
- """
- return _request_ctx_stack.top is not None
- def has_app_context() -> bool:
- """Works like :func:`has_request_context` but for the application
- context. You can also just do a boolean check on the
- :data:`current_app` object instead.
- .. versionadded:: 0.9
- """
- return _app_ctx_stack.top is not None
- class AppContext:
- """The application context binds an application object implicitly
- to the current thread or greenlet, similar to how the
- :class:`RequestContext` binds request information. The application
- context is also implicitly created if a request context is created
- but the application is not on top of the individual application
- context.
- """
- def __init__(self, app: "Flask") -> None:
- self.app = app
- self.url_adapter = app.create_url_adapter(None)
- self.g = app.app_ctx_globals_class()
- # Like request context, app contexts can be pushed multiple times
- # but there a basic "refcount" is enough to track them.
- self._refcnt = 0
- def push(self) -> None:
- """Binds the app context to the current context."""
- self._refcnt += 1
- _app_ctx_stack.push(self)
- appcontext_pushed.send(self.app)
- def pop(self, exc: t.Optional[BaseException] = _sentinel) -> None: # type: ignore
- """Pops the app context."""
- try:
- self._refcnt -= 1
- if self._refcnt <= 0:
- if exc is _sentinel:
- exc = sys.exc_info()[1]
- self.app.do_teardown_appcontext(exc)
- finally:
- rv = _app_ctx_stack.pop()
- assert rv is self, f"Popped wrong app context. ({rv!r} instead of {self!r})"
- appcontext_popped.send(self.app)
- def __enter__(self) -> "AppContext":
- self.push()
- return self
- def __exit__(
- self,
- exc_type: t.Optional[type],
- exc_value: t.Optional[BaseException],
- tb: t.Optional[TracebackType],
- ) -> None:
- self.pop(exc_value)
- class RequestContext:
- """The request context contains all request relevant information. It is
- created at the beginning of the request and pushed to the
- `_request_ctx_stack` and removed at the end of it. It will create the
- URL adapter and request object for the WSGI environment provided.
- Do not attempt to use this class directly, instead use
- :meth:`~flask.Flask.test_request_context` and
- :meth:`~flask.Flask.request_context` to create this object.
- When the request context is popped, it will evaluate all the
- functions registered on the application for teardown execution
- (:meth:`~flask.Flask.teardown_request`).
- The request context is automatically popped at the end of the request
- for you. In debug mode the request context is kept around if
- exceptions happen so that interactive debuggers have a chance to
- introspect the data. With 0.4 this can also be forced for requests
- that did not fail and outside of ``DEBUG`` mode. By setting
- ``'flask._preserve_context'`` to ``True`` on the WSGI environment the
- context will not pop itself at the end of the request. This is used by
- the :meth:`~flask.Flask.test_client` for example to implement the
- deferred cleanup functionality.
- You might find this helpful for unittests where you need the
- information from the context local around for a little longer. Make
- sure to properly :meth:`~werkzeug.LocalStack.pop` the stack yourself in
- that situation, otherwise your unittests will leak memory.
- """
- def __init__(
- self,
- app: "Flask",
- environ: dict,
- request: t.Optional["Request"] = None,
- session: t.Optional["SessionMixin"] = None,
- ) -> None:
- self.app = app
- if request is None:
- request = app.request_class(environ)
- self.request = request
- self.url_adapter = None
- try:
- self.url_adapter = app.create_url_adapter(self.request)
- except HTTPException as e:
- self.request.routing_exception = e
- self.flashes = None
- self.session = session
- # Request contexts can be pushed multiple times and interleaved with
- # other request contexts. Now only if the last level is popped we
- # get rid of them. Additionally if an application context is missing
- # one is created implicitly so for each level we add this information
- self._implicit_app_ctx_stack: t.List[t.Optional["AppContext"]] = []
- # indicator if the context was preserved. Next time another context
- # is pushed the preserved context is popped.
- self.preserved = False
- # remembers the exception for pop if there is one in case the context
- # preservation kicks in.
- self._preserved_exc = None
- # Functions that should be executed after the request on the response
- # object. These will be called before the regular "after_request"
- # functions.
- self._after_request_functions: t.List[AfterRequestCallable] = []
- @property
- def g(self) -> _AppCtxGlobals:
- import warnings
- warnings.warn(
- "Accessing 'g' on the request context is deprecated and"
- " will be removed in Flask 2.2. Access `g` directly or from"
- "the application context instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- return _app_ctx_stack.top.g
- @g.setter
- def g(self, value: _AppCtxGlobals) -> None:
- import warnings
- warnings.warn(
- "Setting 'g' on the request context is deprecated and"
- " will be removed in Flask 2.2. Set it on the application"
- " context instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- _app_ctx_stack.top.g = value
- def copy(self) -> "RequestContext":
- """Creates a copy of this request context with the same request object.
- This can be used to move a request context to a different greenlet.
- Because the actual request object is the same this cannot be used to
- move a request context to a different thread unless access to the
- request object is locked.
- .. versionadded:: 0.10
- .. versionchanged:: 1.1
- The current session object is used instead of reloading the original
- data. This prevents `flask.session` pointing to an out-of-date object.
- """
- return self.__class__(
- self.app,
- environ=self.request.environ,
- request=self.request,
- session=self.session,
- )
- def match_request(self) -> None:
- """Can be overridden by a subclass to hook into the matching
- of the request.
- """
- try:
- result = self.url_adapter.match(return_rule=True) # type: ignore
- self.request.url_rule, self.request.view_args = result # type: ignore
- except HTTPException as e:
- self.request.routing_exception = e
- def push(self) -> None:
- """Binds the request context to the current context."""
- # If an exception occurs in debug mode or if context preservation is
- # activated under exception situations exactly one context stays
- # on the stack. The rationale is that you want to access that
- # information under debug situations. However if someone forgets to
- # pop that context again we want to make sure that on the next push
- # it's invalidated, otherwise we run at risk that something leaks
- # memory. This is usually only a problem in test suite since this
- # functionality is not active in production environments.
- top = _request_ctx_stack.top
- if top is not None and top.preserved:
- top.pop(top._preserved_exc)
- # Before we push the request context we have to ensure that there
- # is an application context.
- app_ctx = _app_ctx_stack.top
- if app_ctx is None or app_ctx.app != self.app:
- app_ctx = self.app.app_context()
- app_ctx.push()
- self._implicit_app_ctx_stack.append(app_ctx)
- else:
- self._implicit_app_ctx_stack.append(None)
- _request_ctx_stack.push(self)
- # Open the session at the moment that the request context is available.
- # This allows a custom open_session method to use the request context.
- # Only open a new session if this is the first time the request was
- # pushed, otherwise stream_with_context loses the session.
- if self.session is None:
- session_interface = self.app.session_interface
- self.session = session_interface.open_session(self.app, self.request)
- if self.session is None:
- self.session = session_interface.make_null_session(self.app)
- # Match the request URL after loading the session, so that the
- # session is available in custom URL converters.
- if self.url_adapter is not None:
- self.match_request()
- def pop(self, exc: t.Optional[BaseException] = _sentinel) -> None: # type: ignore
- """Pops the request context and unbinds it by doing that. This will
- also trigger the execution of functions registered by the
- :meth:`~flask.Flask.teardown_request` decorator.
- .. versionchanged:: 0.9
- Added the `exc` argument.
- """
- app_ctx = self._implicit_app_ctx_stack.pop()
- clear_request = False
- try:
- if not self._implicit_app_ctx_stack:
- self.preserved = False
- self._preserved_exc = None
- if exc is _sentinel:
- exc = sys.exc_info()[1]
- self.app.do_teardown_request(exc)
- request_close = getattr(self.request, "close", None)
- if request_close is not None:
- request_close()
- clear_request = True
- finally:
- rv = _request_ctx_stack.pop()
- # get rid of circular dependencies at the end of the request
- # so that we don't require the GC to be active.
- if clear_request:
- rv.request.environ["werkzeug.request"] = None
- # Get rid of the app as well if necessary.
- if app_ctx is not None:
- app_ctx.pop(exc)
- assert (
- rv is self
- ), f"Popped wrong request context. ({rv!r} instead of {self!r})"
- def auto_pop(self, exc: t.Optional[BaseException]) -> None:
- if self.request.environ.get("flask._preserve_context") or (
- exc is not None and self.app.preserve_context_on_exception
- ):
- self.preserved = True
- self._preserved_exc = exc # type: ignore
- else:
- self.pop(exc)
- def __enter__(self) -> "RequestContext":
- self.push()
- return self
- def __exit__(
- self,
- exc_type: t.Optional[type],
- exc_value: t.Optional[BaseException],
- tb: t.Optional[TracebackType],
- ) -> None:
- # do not pop the request stack if we are in debug mode and an
- # exception happened. This will allow the debugger to still
- # access the request object in the interactive shell. Furthermore
- # the context can be force kept alive for the test client.
- # See flask.testing for how this works.
- self.auto_pop(exc_value)
- def __repr__(self) -> str:
- return (
- f"<{type(self).__name__} {self.request.url!r}"
- f" [{self.request.method}] of {self.app.name}>"
- )
|