Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/flask/ctx.py: 37%
174 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:03 +0000
1import sys
2import typing as t
3from functools import update_wrapper
4from types import TracebackType
6from werkzeug.exceptions import HTTPException
8from . import typing as ft
9from .globals import _app_ctx_stack
10from .globals import _request_ctx_stack
11from .signals import appcontext_popped
12from .signals import appcontext_pushed
14if t.TYPE_CHECKING:
15 from .app import Flask
16 from .sessions import SessionMixin
17 from .wrappers import Request
20# a singleton sentinel value for parameter defaults
21_sentinel = object()
24class _AppCtxGlobals:
25 """A plain object. Used as a namespace for storing data during an
26 application context.
28 Creating an app context automatically creates this object, which is
29 made available as the :data:`g` proxy.
31 .. describe:: 'key' in g
33 Check whether an attribute is present.
35 .. versionadded:: 0.10
37 .. describe:: iter(g)
39 Return an iterator over the attribute names.
41 .. versionadded:: 0.10
42 """
44 # Define attr methods to let mypy know this is a namespace object
45 # that has arbitrary attributes.
47 def __getattr__(self, name: str) -> t.Any:
48 try:
49 return self.__dict__[name]
50 except KeyError:
51 raise AttributeError(name) from None
53 def __setattr__(self, name: str, value: t.Any) -> None:
54 self.__dict__[name] = value
56 def __delattr__(self, name: str) -> None:
57 try:
58 del self.__dict__[name]
59 except KeyError:
60 raise AttributeError(name) from None
62 def get(self, name: str, default: t.Optional[t.Any] = None) -> t.Any:
63 """Get an attribute by name, or a default value. Like
64 :meth:`dict.get`.
66 :param name: Name of attribute to get.
67 :param default: Value to return if the attribute is not present.
69 .. versionadded:: 0.10
70 """
71 return self.__dict__.get(name, default)
73 def pop(self, name: str, default: t.Any = _sentinel) -> t.Any:
74 """Get and remove an attribute by name. Like :meth:`dict.pop`.
76 :param name: Name of attribute to pop.
77 :param default: Value to return if the attribute is not present,
78 instead of raising a ``KeyError``.
80 .. versionadded:: 0.11
81 """
82 if default is _sentinel:
83 return self.__dict__.pop(name)
84 else:
85 return self.__dict__.pop(name, default)
87 def setdefault(self, name: str, default: t.Any = None) -> t.Any:
88 """Get the value of an attribute if it is present, otherwise
89 set and return a default value. Like :meth:`dict.setdefault`.
91 :param name: Name of attribute to get.
92 :param default: Value to set and return if the attribute is not
93 present.
95 .. versionadded:: 0.11
96 """
97 return self.__dict__.setdefault(name, default)
99 def __contains__(self, item: str) -> bool:
100 return item in self.__dict__
102 def __iter__(self) -> t.Iterator[str]:
103 return iter(self.__dict__)
105 def __repr__(self) -> str:
106 top = _app_ctx_stack.top
107 if top is not None:
108 return f"<flask.g of {top.app.name!r}>"
109 return object.__repr__(self)
112def after_this_request(f: ft.AfterRequestCallable) -> ft.AfterRequestCallable:
113 """Executes a function after this request. This is useful to modify
114 response objects. The function is passed the response object and has
115 to return the same or a new one.
117 Example::
119 @app.route('/')
120 def index():
121 @after_this_request
122 def add_header(response):
123 response.headers['X-Foo'] = 'Parachute'
124 return response
125 return 'Hello World!'
127 This is more useful if a function other than the view function wants to
128 modify a response. For instance think of a decorator that wants to add
129 some headers without converting the return value into a response object.
131 .. versionadded:: 0.9
132 """
133 top = _request_ctx_stack.top
135 if top is None:
136 raise RuntimeError(
137 "This decorator can only be used when a request context is"
138 " active, such as within a view function."
139 )
141 top._after_request_functions.append(f)
142 return f
145def copy_current_request_context(f: t.Callable) -> t.Callable:
146 """A helper function that decorates a function to retain the current
147 request context. This is useful when working with greenlets. The moment
148 the function is decorated a copy of the request context is created and
149 then pushed when the function is called. The current session is also
150 included in the copied request context.
152 Example::
154 import gevent
155 from flask import copy_current_request_context
157 @app.route('/')
158 def index():
159 @copy_current_request_context
160 def do_some_work():
161 # do some work here, it can access flask.request or
162 # flask.session like you would otherwise in the view function.
163 ...
164 gevent.spawn(do_some_work)
165 return 'Regular response'
167 .. versionadded:: 0.10
168 """
169 top = _request_ctx_stack.top
171 if top is None:
172 raise RuntimeError(
173 "This decorator can only be used when a request context is"
174 " active, such as within a view function."
175 )
177 reqctx = top.copy()
179 def wrapper(*args, **kwargs):
180 with reqctx:
181 return reqctx.app.ensure_sync(f)(*args, **kwargs)
183 return update_wrapper(wrapper, f)
186def has_request_context() -> bool:
187 """If you have code that wants to test if a request context is there or
188 not this function can be used. For instance, you may want to take advantage
189 of request information if the request object is available, but fail
190 silently if it is unavailable.
192 ::
194 class User(db.Model):
196 def __init__(self, username, remote_addr=None):
197 self.username = username
198 if remote_addr is None and has_request_context():
199 remote_addr = request.remote_addr
200 self.remote_addr = remote_addr
202 Alternatively you can also just test any of the context bound objects
203 (such as :class:`request` or :class:`g`) for truthness::
205 class User(db.Model):
207 def __init__(self, username, remote_addr=None):
208 self.username = username
209 if remote_addr is None and request:
210 remote_addr = request.remote_addr
211 self.remote_addr = remote_addr
213 .. versionadded:: 0.7
214 """
215 return _request_ctx_stack.top is not None
218def has_app_context() -> bool:
219 """Works like :func:`has_request_context` but for the application
220 context. You can also just do a boolean check on the
221 :data:`current_app` object instead.
223 .. versionadded:: 0.9
224 """
225 return _app_ctx_stack.top is not None
228class AppContext:
229 """The application context binds an application object implicitly
230 to the current thread or greenlet, similar to how the
231 :class:`RequestContext` binds request information. The application
232 context is also implicitly created if a request context is created
233 but the application is not on top of the individual application
234 context.
235 """
237 def __init__(self, app: "Flask") -> None:
238 self.app = app
239 self.url_adapter = app.create_url_adapter(None)
240 self.g = app.app_ctx_globals_class()
242 # Like request context, app contexts can be pushed multiple times
243 # but there a basic "refcount" is enough to track them.
244 self._refcnt = 0
246 def push(self) -> None:
247 """Binds the app context to the current context."""
248 self._refcnt += 1
249 _app_ctx_stack.push(self)
250 appcontext_pushed.send(self.app)
252 def pop(self, exc: t.Optional[BaseException] = _sentinel) -> None: # type: ignore
253 """Pops the app context."""
254 try:
255 self._refcnt -= 1
256 if self._refcnt <= 0:
257 if exc is _sentinel:
258 exc = sys.exc_info()[1]
259 self.app.do_teardown_appcontext(exc)
260 finally:
261 rv = _app_ctx_stack.pop()
262 assert rv is self, f"Popped wrong app context. ({rv!r} instead of {self!r})"
263 appcontext_popped.send(self.app)
265 def __enter__(self) -> "AppContext":
266 self.push()
267 return self
269 def __exit__(
270 self,
271 exc_type: t.Optional[type],
272 exc_value: t.Optional[BaseException],
273 tb: t.Optional[TracebackType],
274 ) -> None:
275 self.pop(exc_value)
278class RequestContext:
279 """The request context contains all request relevant information. It is
280 created at the beginning of the request and pushed to the
281 `_request_ctx_stack` and removed at the end of it. It will create the
282 URL adapter and request object for the WSGI environment provided.
284 Do not attempt to use this class directly, instead use
285 :meth:`~flask.Flask.test_request_context` and
286 :meth:`~flask.Flask.request_context` to create this object.
288 When the request context is popped, it will evaluate all the
289 functions registered on the application for teardown execution
290 (:meth:`~flask.Flask.teardown_request`).
292 The request context is automatically popped at the end of the request
293 for you. In debug mode the request context is kept around if
294 exceptions happen so that interactive debuggers have a chance to
295 introspect the data. With 0.4 this can also be forced for requests
296 that did not fail and outside of ``DEBUG`` mode. By setting
297 ``'flask._preserve_context'`` to ``True`` on the WSGI environment the
298 context will not pop itself at the end of the request. This is used by
299 the :meth:`~flask.Flask.test_client` for example to implement the
300 deferred cleanup functionality.
302 You might find this helpful for unittests where you need the
303 information from the context local around for a little longer. Make
304 sure to properly :meth:`~werkzeug.LocalStack.pop` the stack yourself in
305 that situation, otherwise your unittests will leak memory.
306 """
308 def __init__(
309 self,
310 app: "Flask",
311 environ: dict,
312 request: t.Optional["Request"] = None,
313 session: t.Optional["SessionMixin"] = None,
314 ) -> None:
315 self.app = app
316 if request is None:
317 request = app.request_class(environ)
318 self.request = request
319 self.url_adapter = None
320 try:
321 self.url_adapter = app.create_url_adapter(self.request)
322 except HTTPException as e:
323 self.request.routing_exception = e
324 self.flashes = None
325 self.session = session
327 # Request contexts can be pushed multiple times and interleaved with
328 # other request contexts. Now only if the last level is popped we
329 # get rid of them. Additionally if an application context is missing
330 # one is created implicitly so for each level we add this information
331 self._implicit_app_ctx_stack: t.List[t.Optional["AppContext"]] = []
333 # indicator if the context was preserved. Next time another context
334 # is pushed the preserved context is popped.
335 self.preserved = False
337 # remembers the exception for pop if there is one in case the context
338 # preservation kicks in.
339 self._preserved_exc = None
341 # Functions that should be executed after the request on the response
342 # object. These will be called before the regular "after_request"
343 # functions.
344 self._after_request_functions: t.List[ft.AfterRequestCallable] = []
346 @property
347 def g(self) -> _AppCtxGlobals:
348 import warnings
350 warnings.warn(
351 "Accessing 'g' on the request context is deprecated and"
352 " will be removed in Flask 2.2. Access `g` directly or from"
353 "the application context instead.",
354 DeprecationWarning,
355 stacklevel=2,
356 )
357 return _app_ctx_stack.top.g
359 @g.setter
360 def g(self, value: _AppCtxGlobals) -> None:
361 import warnings
363 warnings.warn(
364 "Setting 'g' on the request context is deprecated and"
365 " will be removed in Flask 2.2. Set it on the application"
366 " context instead.",
367 DeprecationWarning,
368 stacklevel=2,
369 )
370 _app_ctx_stack.top.g = value
372 def copy(self) -> "RequestContext":
373 """Creates a copy of this request context with the same request object.
374 This can be used to move a request context to a different greenlet.
375 Because the actual request object is the same this cannot be used to
376 move a request context to a different thread unless access to the
377 request object is locked.
379 .. versionadded:: 0.10
381 .. versionchanged:: 1.1
382 The current session object is used instead of reloading the original
383 data. This prevents `flask.session` pointing to an out-of-date object.
384 """
385 return self.__class__(
386 self.app,
387 environ=self.request.environ,
388 request=self.request,
389 session=self.session,
390 )
392 def match_request(self) -> None:
393 """Can be overridden by a subclass to hook into the matching
394 of the request.
395 """
396 try:
397 result = self.url_adapter.match(return_rule=True) # type: ignore
398 self.request.url_rule, self.request.view_args = result # type: ignore
399 except HTTPException as e:
400 self.request.routing_exception = e
402 def push(self) -> None:
403 """Binds the request context to the current context."""
404 # If an exception occurs in debug mode or if context preservation is
405 # activated under exception situations exactly one context stays
406 # on the stack. The rationale is that you want to access that
407 # information under debug situations. However if someone forgets to
408 # pop that context again we want to make sure that on the next push
409 # it's invalidated, otherwise we run at risk that something leaks
410 # memory. This is usually only a problem in test suite since this
411 # functionality is not active in production environments.
412 top = _request_ctx_stack.top
413 if top is not None and top.preserved:
414 top.pop(top._preserved_exc)
416 # Before we push the request context we have to ensure that there
417 # is an application context.
418 app_ctx = _app_ctx_stack.top
419 if app_ctx is None or app_ctx.app != self.app:
420 app_ctx = self.app.app_context()
421 app_ctx.push()
422 self._implicit_app_ctx_stack.append(app_ctx)
423 else:
424 self._implicit_app_ctx_stack.append(None)
426 _request_ctx_stack.push(self)
428 # Open the session at the moment that the request context is available.
429 # This allows a custom open_session method to use the request context.
430 # Only open a new session if this is the first time the request was
431 # pushed, otherwise stream_with_context loses the session.
432 if self.session is None:
433 session_interface = self.app.session_interface
434 self.session = session_interface.open_session(self.app, self.request)
436 if self.session is None:
437 self.session = session_interface.make_null_session(self.app)
439 # Match the request URL after loading the session, so that the
440 # session is available in custom URL converters.
441 if self.url_adapter is not None:
442 self.match_request()
444 def pop(self, exc: t.Optional[BaseException] = _sentinel) -> None: # type: ignore
445 """Pops the request context and unbinds it by doing that. This will
446 also trigger the execution of functions registered by the
447 :meth:`~flask.Flask.teardown_request` decorator.
449 .. versionchanged:: 0.9
450 Added the `exc` argument.
451 """
452 app_ctx = self._implicit_app_ctx_stack.pop()
453 clear_request = False
455 try:
456 if not self._implicit_app_ctx_stack:
457 self.preserved = False
458 self._preserved_exc = None
459 if exc is _sentinel:
460 exc = sys.exc_info()[1]
461 self.app.do_teardown_request(exc)
463 request_close = getattr(self.request, "close", None)
464 if request_close is not None:
465 request_close()
466 clear_request = True
467 finally:
468 rv = _request_ctx_stack.pop()
470 # get rid of circular dependencies at the end of the request
471 # so that we don't require the GC to be active.
472 if clear_request:
473 rv.request.environ["werkzeug.request"] = None
475 # Get rid of the app as well if necessary.
476 if app_ctx is not None:
477 app_ctx.pop(exc)
479 assert (
480 rv is self
481 ), f"Popped wrong request context. ({rv!r} instead of {self!r})"
483 def auto_pop(self, exc: t.Optional[BaseException]) -> None:
484 if self.request.environ.get("flask._preserve_context") or (
485 exc is not None and self.app.preserve_context_on_exception
486 ):
487 self.preserved = True
488 self._preserved_exc = exc # type: ignore
489 else:
490 self.pop(exc)
492 def __enter__(self) -> "RequestContext":
493 self.push()
494 return self
496 def __exit__(
497 self,
498 exc_type: t.Optional[type],
499 exc_value: t.Optional[BaseException],
500 tb: t.Optional[TracebackType],
501 ) -> None:
502 # do not pop the request stack if we are in debug mode and an
503 # exception happened. This will allow the debugger to still
504 # access the request object in the interactive shell. Furthermore
505 # the context can be force kept alive for the test client.
506 # See flask.testing for how this works.
507 self.auto_pop(exc_value)
509 def __repr__(self) -> str:
510 return (
511 f"<{type(self).__name__} {self.request.url!r}"
512 f" [{self.request.method}] of {self.app.name}>"
513 )