1  from __future__ import annotations
   2  
   3  import contextvars
   4  import sys
   5  import typing as t
   6  from functools import update_wrapper
   7  from types import TracebackType
   8  
   9  from werkzeug.exceptions import HTTPException
  10  
  11  from . import typing as ft
  12  from .globals import _cv_app
  13  from .globals import _cv_request
  14  from .signals import appcontext_popped
  15  from .signals import appcontext_pushed
  16  
  17  if t.TYPE_CHECKING:  # pragma: no cover
  18      from _typeshed.wsgi import WSGIEnvironment
  19  
  20      from .app import Flask
  21      from .sessions import SessionMixin
  22      from .wrappers import Request
  23  
  24  
  25  # a singleton sentinel value for parameter defaults
  26  _sentinel = object()
  27  
  28  
  29  class _AppCtxGlobals:
  30      """A plain object. Used as a namespace for storing data during an
  31      application context.
  32  
  33      Creating an app context automatically creates this object, which is
  34      made available as the :data:`g` proxy.
  35  
  36      .. describe:: 'key' in g
  37  
  38          Check whether an attribute is present.
  39  
  40          .. versionadded:: 0.10
  41  
  42      .. describe:: iter(g)
  43  
  44          Return an iterator over the attribute names.
  45  
  46          .. versionadded:: 0.10
  47      """
  48  
  49      # Define attr methods to let mypy know this is a namespace object
  50      # that has arbitrary attributes.
  51  
  52      def __getattr__(self, name: str) -> t.Any:
  53          try:
  54              return self.__dict__[name]
  55          except KeyError:
  56              raise AttributeError(name) from None
  57  
  58      def __setattr__(self, name: str, value: t.Any) -> None:
  59          self.__dict__[name] = value
  60  
  61      def __delattr__(self, name: str) -> None:
  62          try:
  63              del self.__dict__[name]
  64          except KeyError:
  65              raise AttributeError(name) from None
  66  
  67      def get(self, name: str, default: t.Any | None = None) -> t.Any:
  68          """Get an attribute by name, or a default value. Like
  69          :meth:`dict.get`.
  70  
  71          :param name: Name of attribute to get.
  72          :param default: Value to return if the attribute is not present.
  73  
  74          .. versionadded:: 0.10
  75          """
  76          return self.__dict__.get(name, default)
  77  
  78      def pop(self, name: str, default: t.Any = _sentinel) -> t.Any:
  79          """Get and remove an attribute by name. Like :meth:`dict.pop`.
  80  
  81          :param name: Name of attribute to pop.
  82          :param default: Value to return if the attribute is not present,
  83              instead of raising a ``KeyError``.
  84  
  85          .. versionadded:: 0.11
  86          """
  87          if default is _sentinel:
  88              return self.__dict__.pop(name)
  89          else:
  90              return self.__dict__.pop(name, default)
  91  
  92      def setdefault(self, name: str, default: t.Any = None) -> t.Any:
  93          """Get the value of an attribute if it is present, otherwise
  94          set and return a default value. Like :meth:`dict.setdefault`.
  95  
  96          :param name: Name of attribute to get.
  97          :param default: Value to set and return if the attribute is not
  98              present.
  99  
 100          .. versionadded:: 0.11
 101          """
 102          return self.__dict__.setdefault(name, default)
 103  
 104      def __contains__(self, item: str) -> bool:
 105          return item in self.__dict__
 106  
 107      def __iter__(self) -> t.Iterator[str]:
 108          return iter(self.__dict__)
 109  
 110      def __repr__(self) -> str:
 111          ctx = _cv_app.get(None)
 112          if ctx is not None:
 113              return f"<flask.g of '{ctx.app.name}'>"
 114          return object.__repr__(self)
 115  
 116  
 117  def after_this_request(
 118      f: ft.AfterRequestCallable[t.Any],
 119  ) -> ft.AfterRequestCallable[t.Any]:
 120      """Executes a function after this request.  This is useful to modify
 121      response objects.  The function is passed the response object and has
 122      to return the same or a new one.
 123  
 124      Example::
 125  
 126          @app.route('/')
 127          def index():
 128              @after_this_request
 129              def add_header(response):
 130                  response.headers['X-Foo'] = 'Parachute'
 131                  return response
 132              return 'Hello World!'
 133  
 134      This is more useful if a function other than the view function wants to
 135      modify a response.  For instance think of a decorator that wants to add
 136      some headers without converting the return value into a response object.
 137  
 138      .. versionadded:: 0.9
 139      """
 140      ctx = _cv_request.get(None)
 141  
 142      if ctx is None:
 143          raise RuntimeError(
 144              "'after_this_request' can only be used when a request"
 145              " context is active, such as in a view function."
 146          )
 147  
 148      ctx._after_request_functions.append(f)
 149      return f
 150  
 151  
 152  F = t.TypeVar("F", bound=t.Callable[..., t.Any])
 153  
 154  
 155  def copy_current_request_context(f: F) -> F:
 156      """A helper function that decorates a function to retain the current
 157      request context.  This is useful when working with greenlets.  The moment
 158      the function is decorated a copy of the request context is created and
 159      then pushed when the function is called.  The current session is also
 160      included in the copied request context.
 161  
 162      Example::
 163  
 164          import gevent
 165          from flask import copy_current_request_context
 166  
 167          @app.route('/')
 168          def index():
 169              @copy_current_request_context
 170              def do_some_work():
 171                  # do some work here, it can access flask.request or
 172                  # flask.session like you would otherwise in the view function.
 173                  ...
 174              gevent.spawn(do_some_work)
 175              return 'Regular response'
 176  
 177      .. versionadded:: 0.10
 178      """
 179      ctx = _cv_request.get(None)
 180  
 181      if ctx is None:
 182          raise RuntimeError(
 183              "'copy_current_request_context' can only be used when a"
 184              " request context is active, such as in a view function."
 185          )
 186  
 187      ctx = ctx.copy()
 188  
 189      def wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any:
 190          with ctx:  # type: ignore[union-attr]
 191              return ctx.app.ensure_sync(f)(*args, **kwargs)  # type: ignore[union-attr]
 192  
 193      return update_wrapper(wrapper, f)  # type: ignore[return-value]
 194  
 195  
 196  def has_request_context() -> bool:
 197      """If you have code that wants to test if a request context is there or
 198      not this function can be used.  For instance, you may want to take advantage
 199      of request information if the request object is available, but fail
 200      silently if it is unavailable.
 201  
 202      ::
 203  
 204          class User(db.Model):
 205  
 206              def __init__(self, username, remote_addr=None):
 207                  self.username = username
 208                  if remote_addr is None and has_request_context():
 209                      remote_addr = request.remote_addr
 210                  self.remote_addr = remote_addr
 211  
 212      Alternatively you can also just test any of the context bound objects
 213      (such as :class:`request` or :class:`g`) for truthness::
 214  
 215          class User(db.Model):
 216  
 217              def __init__(self, username, remote_addr=None):
 218                  self.username = username
 219                  if remote_addr is None and request:
 220                      remote_addr = request.remote_addr
 221                  self.remote_addr = remote_addr
 222  
 223      .. versionadded:: 0.7
 224      """
 225      return _cv_request.get(None) is not None
 226  
 227  
 228  def has_app_context() -> bool:
 229      """Works like :func:`has_request_context` but for the application
 230      context.  You can also just do a boolean check on the
 231      :data:`current_app` object instead.
 232  
 233      .. versionadded:: 0.9
 234      """
 235      return _cv_app.get(None) is not None
 236  
 237  
 238  class AppContext:
 239      """The app context contains application-specific information. An app
 240      context is created and pushed at the beginning of each request if
 241      one is not already active. An app context is also pushed when
 242      running CLI commands.
 243      """
 244  
 245      def __init__(self, app: Flask) -> None:
 246          self.app = app
 247          self.url_adapter = app.create_url_adapter(None)
 248          self.g: _AppCtxGlobals = app.app_ctx_globals_class()
 249          self._cv_tokens: list[contextvars.Token[AppContext]] = []
 250  
 251      def push(self) -> None:
 252          """Binds the app context to the current context."""
 253          self._cv_tokens.append(_cv_app.set(self))
 254          appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync)
 255  
 256      def pop(self, exc: BaseException | None = _sentinel) -> None:  # type: ignore
 257          """Pops the app context."""
 258          try:
 259              if len(self._cv_tokens) == 1:
 260                  if exc is _sentinel:
 261                      exc = sys.exc_info()[1]
 262                  self.app.do_teardown_appcontext(exc)
 263          finally:
 264              ctx = _cv_app.get()
 265              _cv_app.reset(self._cv_tokens.pop())
 266  
 267          if ctx is not self:
 268              raise AssertionError(
 269                  f"Popped wrong app context. ({ctx!r} instead of {self!r})"
 270              )
 271  
 272          appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync)
 273  
 274      def __enter__(self) -> AppContext:
 275          self.push()
 276          return self
 277  
 278      def __exit__(
 279          self,
 280          exc_type: type | None,
 281          exc_value: BaseException | None,
 282          tb: TracebackType | None,
 283      ) -> None:
 284          self.pop(exc_value)
 285  
 286  
 287  class RequestContext:
 288      """The request context contains per-request information. The Flask
 289      app creates and pushes it at the beginning of the request, then pops
 290      it at the end of the request. It will create the URL adapter and
 291      request object for the WSGI environment provided.
 292  
 293      Do not attempt to use this class directly, instead use
 294      :meth:`~flask.Flask.test_request_context` and
 295      :meth:`~flask.Flask.request_context` to create this object.
 296  
 297      When the request context is popped, it will evaluate all the
 298      functions registered on the application for teardown execution
 299      (:meth:`~flask.Flask.teardown_request`).
 300  
 301      The request context is automatically popped at the end of the
 302      request. When using the interactive debugger, the context will be
 303      restored so ``request`` is still accessible. Similarly, the test
 304      client can preserve the context after the request ends. However,
 305      teardown functions may already have closed some resources such as
 306      database connections.
 307      """
 308  
 309      def __init__(
 310          self,
 311          app: Flask,
 312          environ: WSGIEnvironment,
 313          request: Request | None = None,
 314          session: SessionMixin | None = None,
 315      ) -> None:
 316          self.app = app
 317          if request is None:
 318              request = app.request_class(environ)
 319              request.json_module = app.json
 320          self.request: Request = request
 321          self.url_adapter = None
 322          try:
 323              self.url_adapter = app.create_url_adapter(self.request)
 324          except HTTPException as e:
 325              self.request.routing_exception = e
 326          self.flashes: list[tuple[str, str]] | None = None
 327          self.session: SessionMixin | None = session
 328          # Functions that should be executed after the request on the response
 329          # object.  These will be called before the regular "after_request"
 330          # functions.
 331          self._after_request_functions: list[ft.AfterRequestCallable[t.Any]] = []
 332  
 333          self._cv_tokens: list[
 334              tuple[contextvars.Token[RequestContext], AppContext | None]
 335          ] = []
 336  
 337      def copy(self) -> RequestContext:
 338          """Creates a copy of this request context with the same request object.
 339          This can be used to move a request context to a different greenlet.
 340          Because the actual request object is the same this cannot be used to
 341          move a request context to a different thread unless access to the
 342          request object is locked.
 343  
 344          .. versionadded:: 0.10
 345  
 346          .. versionchanged:: 1.1
 347             The current session object is used instead of reloading the original
 348             data. This prevents `flask.session` pointing to an out-of-date object.
 349          """
 350          return self.__class__(
 351              self.app,
 352              environ=self.request.environ,
 353              request=self.request,
 354              session=self.session,
 355          )
 356  
 357      def match_request(self) -> None:
 358          """Can be overridden by a subclass to hook into the matching
 359          of the request.
 360          """
 361          try:
 362              result = self.url_adapter.match(return_rule=True)  # type: ignore
 363              self.request.url_rule, self.request.view_args = result  # type: ignore
 364          except HTTPException as e:
 365              self.request.routing_exception = e
 366  
 367      def push(self) -> None:
 368          # Before we push the request context we have to ensure that there
 369          # is an application context.
 370          app_ctx = _cv_app.get(None)
 371  
 372          if app_ctx is None or app_ctx.app is not self.app:
 373              app_ctx = self.app.app_context()
 374              app_ctx.push()
 375          else:
 376              app_ctx = None
 377  
 378          self._cv_tokens.append((_cv_request.set(self), app_ctx))
 379  
 380          # Open the session at the moment that the request context is available.
 381          # This allows a custom open_session method to use the request context.
 382          # Only open a new session if this is the first time the request was
 383          # pushed, otherwise stream_with_context loses the session.
 384          if self.session is None:
 385              session_interface = self.app.session_interface
 386              self.session = session_interface.open_session(self.app, self.request)
 387  
 388              if self.session is None:
 389                  self.session = session_interface.make_null_session(self.app)
 390  
 391          # Match the request URL after loading the session, so that the
 392          # session is available in custom URL converters.
 393          if self.url_adapter is not None:
 394              self.match_request()
 395  
 396      def pop(self, exc: BaseException | None = _sentinel) -> None:  # type: ignore
 397          """Pops the request context and unbinds it by doing that.  This will
 398          also trigger the execution of functions registered by the
 399          :meth:`~flask.Flask.teardown_request` decorator.
 400  
 401          .. versionchanged:: 0.9
 402             Added the `exc` argument.
 403          """
 404          clear_request = len(self._cv_tokens) == 1
 405  
 406          try:
 407              if clear_request:
 408                  if exc is _sentinel:
 409                      exc = sys.exc_info()[1]
 410                  self.app.do_teardown_request(exc)
 411  
 412                  request_close = getattr(self.request, "close", None)
 413                  if request_close is not None:
 414                      request_close()
 415          finally:
 416              ctx = _cv_request.get()
 417              token, app_ctx = self._cv_tokens.pop()
 418              _cv_request.reset(token)
 419  
 420              # get rid of circular dependencies at the end of the request
 421              # so that we don't require the GC to be active.
 422              if clear_request:
 423                  ctx.request.environ["werkzeug.request"] = None
 424  
 425              if app_ctx is not None:
 426                  app_ctx.pop(exc)
 427  
 428              if ctx is not self:
 429                  raise AssertionError(
 430                      f"Popped wrong request context. ({ctx!r} instead of {self!r})"
 431                  )
 432  
 433      def __enter__(self) -> RequestContext:
 434          self.push()
 435          return self
 436  
 437      def __exit__(
 438          self,
 439          exc_type: type | None,
 440          exc_value: BaseException | None,
 441          tb: TracebackType | None,
 442      ) -> None:
 443          self.pop(exc_value)
 444  
 445      def __repr__(self) -> str:
 446          return (
 447              f"<{type(self).__name__} {self.request.url!r}"
 448              f" [{self.request.method}] of {self.app.name}>"
 449          )
 450