Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/flask/ctx.py: 29%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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 _typeshed.wsgi import WSGIEnvironment
20 from .app import Flask
21 from .sessions import SessionMixin
22 from .wrappers import Request
25# a singleton sentinel value for parameter defaults
26_sentinel = object()
29class _AppCtxGlobals:
30 """A plain object. Used as a namespace for storing data during an
31 application context.
33 Creating an app context automatically creates this object, which is
34 made available as the :data:`g` proxy.
36 .. describe:: 'key' in g
38 Check whether an attribute is present.
40 .. versionadded:: 0.10
42 .. describe:: iter(g)
44 Return an iterator over the attribute names.
46 .. versionadded:: 0.10
47 """
49 # Define attr methods to let mypy know this is a namespace object
50 # that has arbitrary attributes.
52 def __getattr__(self, name: str) -> t.Any:
53 try:
54 return self.__dict__[name]
55 except KeyError:
56 raise AttributeError(name) from None
58 def __setattr__(self, name: str, value: t.Any) -> None:
59 self.__dict__[name] = value
61 def __delattr__(self, name: str) -> None:
62 try:
63 del self.__dict__[name]
64 except KeyError:
65 raise AttributeError(name) from None
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`.
71 :param name: Name of attribute to get.
72 :param default: Value to return if the attribute is not present.
74 .. versionadded:: 0.10
75 """
76 return self.__dict__.get(name, default)
78 def pop(self, name: str, default: t.Any = _sentinel) -> t.Any:
79 """Get and remove an attribute by name. Like :meth:`dict.pop`.
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``.
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)
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`.
96 :param name: Name of attribute to get.
97 :param default: Value to set and return if the attribute is not
98 present.
100 .. versionadded:: 0.11
101 """
102 return self.__dict__.setdefault(name, default)
104 def __contains__(self, item: str) -> bool:
105 return item in self.__dict__
107 def __iter__(self) -> t.Iterator[str]:
108 return iter(self.__dict__)
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)
117def 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.
124 Example::
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!'
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.
138 .. versionadded:: 0.9
139 """
140 ctx = _cv_request.get(None)
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 )
148 ctx._after_request_functions.append(f)
149 return f
152F = t.TypeVar("F", bound=t.Callable[..., t.Any])
155def 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.
162 Example::
164 import gevent
165 from flask import copy_current_request_context
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'
177 .. versionadded:: 0.10
178 """
179 ctx = _cv_request.get(None)
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 )
187 ctx = ctx.copy()
189 def wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any:
190 with ctx:
191 return ctx.app.ensure_sync(f)(*args, **kwargs)
193 return update_wrapper(wrapper, f) # type: ignore[return-value]
196def 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.
202 ::
204 class User(db.Model):
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
212 Alternatively you can also just test any of the context bound objects
213 (such as :class:`request` or :class:`g`) for truthness::
215 class User(db.Model):
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
223 .. versionadded:: 0.7
224 """
225 return _cv_request.get(None) is not None
228def 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.
233 .. versionadded:: 0.9
234 """
235 return _cv_app.get(None) is not None
238class 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 """
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]] = []
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)
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())
267 if ctx is not self:
268 raise AssertionError(
269 f"Popped wrong app context. ({ctx!r} instead of {self!r})"
270 )
272 appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync)
274 def __enter__(self) -> AppContext:
275 self.push()
276 return self
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)
287class 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.
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.
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`).
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 """
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]] = []
333 self._cv_tokens: list[
334 tuple[contextvars.Token[RequestContext], AppContext | None]
335 ] = []
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.
344 .. versionadded:: 0.10
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 )
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
367 @property
368 def session(self) -> SessionMixin:
369 """The session data associated with this request. Not available until
370 this context has been pushed. Accessing this property, also accessed by
371 the :data:`~flask.session` proxy, sets :attr:`.SessionMixin.accessed`.
372 """
373 assert self._session is not None, "The session has not yet been opened."
374 self._session.accessed = True
375 return self._session
377 def push(self) -> None:
378 # Before we push the request context we have to ensure that there
379 # is an application context.
380 app_ctx = _cv_app.get(None)
382 if app_ctx is None or app_ctx.app is not self.app:
383 app_ctx = self.app.app_context()
384 app_ctx.push()
385 else:
386 app_ctx = None
388 self._cv_tokens.append((_cv_request.set(self), app_ctx))
390 # Open the session at the moment that the request context is available.
391 # This allows a custom open_session method to use the request context.
392 # Only open a new session if this is the first time the request was
393 # pushed, otherwise stream_with_context loses the session.
394 if self._session is None:
395 session_interface = self.app.session_interface
396 self._session = session_interface.open_session(self.app, self.request)
398 if self._session is None:
399 self._session = session_interface.make_null_session(self.app)
401 # Match the request URL after loading the session, so that the
402 # session is available in custom URL converters.
403 if self.url_adapter is not None:
404 self.match_request()
406 def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore
407 """Pops the request context and unbinds it by doing that. This will
408 also trigger the execution of functions registered by the
409 :meth:`~flask.Flask.teardown_request` decorator.
411 .. versionchanged:: 0.9
412 Added the `exc` argument.
413 """
414 clear_request = len(self._cv_tokens) == 1
416 try:
417 if clear_request:
418 if exc is _sentinel:
419 exc = sys.exc_info()[1]
420 self.app.do_teardown_request(exc)
422 request_close = getattr(self.request, "close", None)
423 if request_close is not None:
424 request_close()
425 finally:
426 ctx = _cv_request.get()
427 token, app_ctx = self._cv_tokens.pop()
428 _cv_request.reset(token)
430 # get rid of circular dependencies at the end of the request
431 # so that we don't require the GC to be active.
432 if clear_request:
433 ctx.request.environ["werkzeug.request"] = None
435 if app_ctx is not None:
436 app_ctx.pop(exc)
438 if ctx is not self:
439 raise AssertionError(
440 f"Popped wrong request context. ({ctx!r} instead of {self!r})"
441 )
443 def __enter__(self) -> RequestContext:
444 self.push()
445 return self
447 def __exit__(
448 self,
449 exc_type: type | None,
450 exc_value: BaseException | None,
451 tb: TracebackType | None,
452 ) -> None:
453 self.pop(exc_value)
455 def __repr__(self) -> str:
456 return (
457 f"<{type(self).__name__} {self.request.url!r}"
458 f" [{self.request.method}] of {self.app.name}>"
459 )