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