orbit-oss/flask
4
4
Fork 2
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

merge app and request context

Browse source
This commit is contained in:
David Lord 2025-09-12 14:52:03 -07:00
parent 330123258e
commit c2705ffd9c
No known key found for this signature in database
GPG key ID: 43368A7AA8CC5926
36 changed files with 779 additions and 1007 deletions
Download patch file Download diff file Expand all files Collapse all files

176
src/flask/app.py
View file

@ -29,20 +29,15 @@ from werkzeug.wsgi import get_host
from . import cli
from . import typing as ft
from .ctx import AppContext
from .ctx import RequestContext
from .globals import _cv_app
from .globals import _cv_request
from .globals import current_app
from .globals import g
from .globals import request
from .globals import request_ctx
from .globals import session
from .helpers import get_debug_flag
from .helpers import get_flashed_messages
from .helpers import get_load_dotenv
from .helpers import send_from_directory
from .sansio.app import App
from .sansio.scaffold import _sentinel
from .sessions import SecureCookieSessionInterface
from .sessions import SessionInterface
from .signals import appcontext_tearing_down
@ -295,7 +290,7 @@ class Flask(App):
.. versionadded:: 0.9
"""
value = current_app.config["SEND_FILE_MAX_AGE_DEFAULT"]
value = self.config["SEND_FILE_MAX_AGE_DEFAULT"]
if value is None:
return None
@ -517,8 +512,8 @@ class Flask(App):
names: t.Iterable[str | None] = (None,)
# A template may be rendered outside a request context.
if request:
names = chain(names, reversed(request.blueprints))
if (ctx := _cv_app.get(None)) is not None and ctx.has_request:
names = chain(names, reversed(ctx.request.blueprints))
# The values passed to render_template take precedence. Keep a
# copy to re-apply after all context functions.
@ -886,7 +881,8 @@ class Flask(App):
This no longer does the exception handling, this code was
moved to the new :meth:`full_dispatch_request`.
"""
req = request_ctx.request
req = _cv_app.get().request
if req.routing_exception is not None:
self.raise_routing_exception(req)
rule: Rule = req.url_rule # type: ignore[assignment]
@ -957,7 +953,7 @@ class Flask(App):
.. versionadded:: 0.7
"""
adapter = request_ctx.url_adapter
adapter = _cv_app.get().url_adapter
methods = adapter.allowed_methods() # type: ignore[union-attr]
rv = self.response_class()
rv.allow.update(methods)
@ -1057,11 +1053,9 @@ class Flask(App):
.. versionadded:: 2.2
Moved from ``flask.url_for``, which calls this method.
"""
req_ctx = _cv_request.get(None)
if req_ctx is not None:
url_adapter = req_ctx.url_adapter
blueprint_name = req_ctx.request.blueprint
if (ctx := _cv_app.get(None)) is not None and ctx.has_request:
url_adapter = ctx.url_adapter
blueprint_name = ctx.request.blueprint
# If the endpoint starts with "." and the request matches a
# blueprint, the endpoint is relative to the blueprint.
@ -1076,13 +1070,11 @@ class Flask(App):
if _external is None:
_external = _scheme is not None
else:
app_ctx = _cv_app.get(None)
# If called by helpers.url_for, an app context is active,
# use its url_adapter. Otherwise, app.url_for was called
# directly, build an adapter.
if app_ctx is not None:
url_adapter = app_ctx.url_adapter
if ctx is not None:
url_adapter = ctx.url_adapter
else:
url_adapter = self.create_url_adapter(None)
@ -1278,12 +1270,13 @@ class Flask(App):
value is handled as if it was the return value from the view, and
further request handling is stopped.
"""
names = (None, *reversed(request.blueprints))
req = _cv_app.get().request
names = (None, *reversed(req.blueprints))
for name in names:
if name in self.url_value_preprocessors:
for url_func in self.url_value_preprocessors[name]:
url_func(request.endpoint, request.view_args)
url_func(req.endpoint, req.view_args)
for name in names:
if name in self.before_request_funcs:
@ -1308,12 +1301,12 @@ class Flask(App):
:return: a new response object or the same, has to be an
instance of :attr:`response_class`.
"""
ctx = request_ctx._get_current_object() # type: ignore[attr-defined]
ctx = _cv_app.get()
for func in ctx._after_request_functions:
response = self.ensure_sync(func)(response)
for name in chain(request.blueprints, (None,)):
for name in chain(ctx.request.blueprints, (None,)):
if name in self.after_request_funcs:
for func in reversed(self.after_request_funcs[name]):
response = self.ensure_sync(func)(response)
@ -1323,77 +1316,57 @@ class Flask(App):
return response
def do_teardown_request(
self,
exc: BaseException | None = _sentinel, # type: ignore[assignment]
) -> None:
"""Called after the request is dispatched and the response is
returned, right before the request context is popped.
def do_teardown_request(self, exc: BaseException | None = None) -> None:
"""Called after the request is dispatched and the response is finalized,
right before the request context is popped. Called by
:meth:`.AppContext.pop`.
This calls all functions decorated with
:meth:`teardown_request`, and :meth:`Blueprint.teardown_request`
if a blueprint handled the request. Finally, the
:data:`request_tearing_down` signal is sent.
This calls all functions decorated with :meth:`teardown_request`, and
:meth:`Blueprint.teardown_request` if a blueprint handled the request.
Finally, the :data:`request_tearing_down` signal is sent.
This is called by
:meth:`RequestContext.pop() <flask.ctx.RequestContext.pop>`,
which may be delayed during testing to maintain access to
resources.
:param exc: An unhandled exception raised while dispatching the
request. Detected from the current exception information if
not passed. Passed to each teardown function.
:param exc: An unhandled exception raised while dispatching the request.
Passed to each teardown function.
.. versionchanged:: 0.9
Added the ``exc`` argument.
"""
if exc is _sentinel:
exc = sys.exc_info()[1]
req = _cv_app.get().request
for name in chain(request.blueprints, (None,)):
for name in chain(req.blueprints, (None,)):
if name in self.teardown_request_funcs:
for func in reversed(self.teardown_request_funcs[name]):
self.ensure_sync(func)(exc)
request_tearing_down.send(self, _async_wrapper=self.ensure_sync, exc=exc)
def do_teardown_appcontext(
self,
exc: BaseException | None = _sentinel, # type: ignore[assignment]
) -> None:
"""Called right before the application context is popped.
def do_teardown_appcontext(self, exc: BaseException | None = None) -> None:
"""Called right before the application context is popped. Called by
:meth:`.AppContext.pop`.
When handling a request, the application context is popped
after the request context. See :meth:`do_teardown_request`.
This calls all functions decorated with :meth:`teardown_appcontext`.
Then the :data:`appcontext_tearing_down` signal is sent.
This calls all functions decorated with
:meth:`teardown_appcontext`. Then the
:data:`appcontext_tearing_down` signal is sent.
This is called by
:meth:`AppContext.pop() <flask.ctx.AppContext.pop>`.
:param exc: An unhandled exception raised while the context was active.
Passed to each teardown function.
.. versionadded:: 0.9
"""
if exc is _sentinel:
exc = sys.exc_info()[1]
for func in reversed(self.teardown_appcontext_funcs):
self.ensure_sync(func)(exc)
appcontext_tearing_down.send(self, _async_wrapper=self.ensure_sync, exc=exc)
def app_context(self) -> AppContext:
"""Create an :class:`~flask.ctx.AppContext`. Use as a ``with``
block to push the context, which will make :data:`current_app`
point at this application.
"""Create an :class:`.AppContext`. When the context is pushed,
:data:`.current_app` and :data:`.g` become available.
An application context is automatically pushed by
:meth:`RequestContext.push() <flask.ctx.RequestContext.push>`
when handling a request, and when running a CLI command. Use
this to manually create a context outside of these situations.
A context is automatically pushed when handling each request, and when
running any ``flask`` CLI command. Use this as a ``with`` block to
manually push a context outside of those situations, such as during
setup or testing.
::
.. code-block:: python
with app.app_context():
init_db()
@ -1404,44 +1377,37 @@ class Flask(App):
"""
return AppContext(self)
def request_context(self, environ: WSGIEnvironment) -> RequestContext:
"""Create a :class:`~flask.ctx.RequestContext` representing a
WSGI environment. Use a ``with`` block to push the context,
which will make :data:`request` point at this request.
def request_context(self, environ: WSGIEnvironment) -> AppContext:
"""Create an :class:`.AppContext` with request information representing
the given WSGI environment. A context is automatically pushed when
handling each request. When the context is pushed, :data:`.request`,
:data:`.session`, :data:`g:, and :data:`.current_app` become available.
See :doc:`/reqcontext`.
This method should not be used in your own code. Creating a valid WSGI
environ is not trivial. Use :meth:`test_request_context` to correctly
create a WSGI environ and request context instead.
Typically you should not call this from your own code. A request
context is automatically pushed by the :meth:`wsgi_app` when
handling a request. Use :meth:`test_request_context` to create
an environment and context instead of this method.
See :doc:`/appcontext`.
:param environ: a WSGI environment
:param environ: A WSGI environment.
"""
return RequestContext(self, environ)
return AppContext.from_environ(self, environ)
def test_request_context(self, *args: t.Any, **kwargs: t.Any) -> RequestContext:
"""Create a :class:`~flask.ctx.RequestContext` for a WSGI
environment created from the given values. This is mostly useful
during testing, where you may want to run a function that uses
request data without dispatching a full request.
def test_request_context(self, *args: t.Any, **kwargs: t.Any) -> AppContext:
"""Create an :class:`.AppContext` with request information created from
the given arguments. When the context is pushed, :data:`.request`,
:data:`.session`, :data:`g:, and :data:`.current_app` become available.
See :doc:`/reqcontext`.
This is useful during testing to run a function that uses request data
without dispatching a full request. Use this as a ``with`` block to push
a context.
Use a ``with`` block to push the context, which will make
:data:`request` point at the request for the created
environment. ::
.. code-block:: python
with app.test_request_context(...):
generate_report()
When using the shell, it may be easier to push and pop the
context manually to avoid indentation. ::
ctx = app.test_request_context(...)
ctx.push()
...
ctx.pop()
See :doc:`/appcontext`.
Takes the same arguments as Werkzeug's
:class:`~werkzeug.test.EnvironBuilder`, with some defaults from
@ -1451,20 +1417,18 @@ class Flask(App):
:param path: URL path being requested.
:param base_url: Base URL where the app is being served, which
``path`` is relative to. If not given, built from
:data:`PREFERRED_URL_SCHEME`, ``subdomain``,
:data:`SERVER_NAME`, and :data:`APPLICATION_ROOT`.
:param subdomain: Subdomain name to append to
:data:`SERVER_NAME`.
:data:`PREFERRED_URL_SCHEME`, ``subdomain``, :data:`SERVER_NAME`,
and :data:`APPLICATION_ROOT`.
:param subdomain: Subdomain name to prepend to :data:`SERVER_NAME`.
:param url_scheme: Scheme to use instead of
:data:`PREFERRED_URL_SCHEME`.
:param data: The request body, either as a string or a dict of
form keys and values.
:param data: The request body text or bytes,or a dict of form data.
:param json: If given, this is serialized as JSON and passed as
``data``. Also defaults ``content_type`` to
``application/json``.
:param args: other positional arguments passed to
:param args: Other positional arguments passed to
:class:`~werkzeug.test.EnvironBuilder`.
:param kwargs: other keyword arguments passed to
:param kwargs: Other keyword arguments passed to
:class:`~werkzeug.test.EnvironBuilder`.
"""
from .testing import EnvironBuilder
@ -1472,10 +1436,12 @@ class Flask(App):
builder = EnvironBuilder(self, *args, **kwargs)
try:
return self.request_context(builder.get_environ())
environ = builder.get_environ()
finally:
builder.close()
return self.request_context(environ)
def wsgi_app(
self, environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable[bytes]:
@ -1496,7 +1462,6 @@ class Flask(App):
Teardown events for the request and app contexts are called
even if an unhandled error occurs. Other events may not be
called depending on when an error occurs during dispatch.
See :ref:`callbacks-and-errors`.
:param environ: A WSGI environment.
:param start_response: A callable accepting a status code,
@ -1519,7 +1484,6 @@ class Flask(App):
finally:
if "werkzeug.debug.preserve_context" in environ:
environ["werkzeug.debug.preserve_context"](_cv_app.get())
environ["werkzeug.debug.preserve_context"](_cv_request.get())
if error is not None and self.should_ignore_error(error):
error = None

2
src/flask/cli.py
View file

@ -628,7 +628,7 @@ class FlaskGroup(AppGroup):
# Push an app context for the loaded app unless it is already
# active somehow. This makes the context available to parameter
# and command callbacks without needing @with_appcontext.
if not current_app or current_app._get_current_object() is not app: # type: ignore[attr-defined]
if not current_app or current_app._get_current_object() is not app:
ctx.with_resource(app.app_context())
return app.cli.get_command(ctx, name)

461
src/flask/ctx.py
View file

@ -1,20 +1,20 @@
from __future__ import annotations
import contextvars
import sys
import typing as t
from functools import update_wrapper
from types import TracebackType
from werkzeug.exceptions import HTTPException
from werkzeug.routing import MapAdapter
from . import typing as ft
from .globals import _cv_app
from .globals import _cv_request
from .signals import appcontext_popped
from .signals import appcontext_pushed
if t.TYPE_CHECKING: # pragma: no cover
if t.TYPE_CHECKING:
import typing_extensions as te
from _typeshed.wsgi import WSGIEnvironment
from .app import Flask
@ -31,7 +31,7 @@ class _AppCtxGlobals:
application context.
Creating an app context automatically creates this object, which is
made available as the :data:`g` proxy.
made available as the :data:`.g` proxy.
.. describe:: 'key' in g
@ -117,29 +117,27 @@ class _AppCtxGlobals:
def after_this_request(
f: ft.AfterRequestCallable[t.Any],
) -> ft.AfterRequestCallable[t.Any]:
"""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.
"""Decorate a function to run after the current request. The behavior is the
same as :meth:`.Flask.after_request`, except it only applies to the current
request, rather than every request. Therefore, it must be used within a
request context, rather than during setup.
Example::
.. code-block:: python
@app.route('/')
@app.route("/")
def index():
@after_this_request
def add_header(response):
response.headers['X-Foo'] = 'Parachute'
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.
return "Hello, World!"
.. versionadded:: 0.9
"""
ctx = _cv_request.get(None)
ctx = _cv_app.get(None)
if ctx is None:
if ctx is None or not ctx.has_request:
raise RuntimeError(
"'after_this_request' can only be used when a request"
" context is active, such as in a view function."
@ -153,13 +151,27 @@ F = t.TypeVar("F", bound=t.Callable[..., t.Any])
def copy_current_request_context(f: F) -> F:
"""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.
"""Decorate a function to run inside the current request context. This can
be used when starting a background task, otherwise it will not see the app
and request objects that were active in the parent.
Example::
.. warning::
Due to the following caveats, it is often safer (and simpler) to pass
the data you need when starting the task, rather than using this and
relying on the context objects.
In order to avoid execution switching partially though reading data, either
read the request body (access ``form``, ``json``, ``data``, etc) before
starting the task, or use a lock. This can be an issue when using threading,
but shouldn't be an issue when using greenlet/gevent or asyncio.
If the task will access ``session``, be sure to do so in the parent as well
so that the ``Vary: cookie`` header will be set. Modifying ``session`` in
the task should be avoided, as it may execute after the response cookie has
already been written.
.. code-block:: python
import gevent
from flask import copy_current_request_context
@ -176,7 +188,7 @@ def copy_current_request_context(f: F) -> F:
.. versionadded:: 0.10
"""
ctx = _cv_request.get(None)
ctx = _cv_app.get(None)
if ctx is None:
raise RuntimeError(
@ -194,41 +206,50 @@ def copy_current_request_context(f: F) -> 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.
"""Test if an app context is active and if it has request information.
::
.. code-block:: python
class User(db.Model):
from flask import has_request_context, request
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
if has_request_context():
remote_addr = request.remote_addr
Alternatively you can also just test any of the context bound objects
(such as :class:`request` or :class:`g`) for truthness::
If a request context is active, the :data:`.request` and :data:`.session`
context proxies will available and ``True``, otherwise ``False``. You can
use that to test the data you use, rather than using this function.
class User(db.Model):
.. code-block:: python
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
from flask import request
if request:
remote_addr = request.remote_addr
.. versionadded:: 0.7
"""
return _cv_request.get(None) is not None
return (ctx := _cv_app.get(None)) is not None and ctx.has_request
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.
"""Test if an app context is active. Unlike :func:`has_request_context`
this can be true outside a request, such as in a CLI command.
.. code-block:: python
from flask import has_app_context, g
if has_app_context():
g.cached_data = ...
If an app context is active, the :data:`.g` and :data:`.current_app` context
proxies will available and ``True``, otherwise ``False``. You can use that
to test the data you use, rather than using this function.
from flask import g
if g:
g.cached_data = ...
.. versionadded:: 0.9
"""
@ -236,214 +257,260 @@ def has_app_context() -> bool:
class AppContext:
"""The app context contains application-specific information. An app
context is created and pushed at the beginning of each request if
one is not already active. An app context is also pushed when
running CLI commands.
"""
"""An app context contains information about an app, and about the request
when handling a request. A context is pushed at the beginning of each
request and CLI command, and popped at the end. The context is referred to
as a "request context" if it has request information, and an "app context"
if not.
def __init__(self, app: Flask) -> None:
self.app = app
self.url_adapter = app.create_url_adapter(None)
self.g: _AppCtxGlobals = app.app_ctx_globals_class()
self._cv_tokens: list[contextvars.Token[AppContext]] = []
Do not use this class directly. Use :meth:`.Flask.app_context` to create an
app context if needed during setup, and :meth:`.Flask.test_request_context`
to create a request context if needed during tests.
def push(self) -> None:
"""Binds the app context to the current context."""
self._cv_tokens.append(_cv_app.set(self))
appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync)
When the context is popped, it will evaluate all the teardown functions
registered with :meth:`~flask.Flask.teardown_request` (if handling a
request) then :meth:`.Flask.teardown_appcontext`.
def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore
"""Pops the app context."""
try:
if len(self._cv_tokens) == 1:
if exc is _sentinel:
exc = sys.exc_info()[1]
self.app.do_teardown_appcontext(exc)
finally:
ctx = _cv_app.get()
_cv_app.reset(self._cv_tokens.pop())
When using the interactive debugger, the context will be restored so
``request`` is still accessible. Similarly, the test client can preserve the
context after the request ends. However, teardown functions may already have
closed some resources such as database connections, and will run again when
the restored context is popped.
if ctx is not self:
raise AssertionError(
f"Popped wrong app context. ({ctx!r} instead of {self!r})"
)
:param app: The application this context represents.
:param request: The request data this context represents.
:param session: The session data this context represents. If not given,
loaded from the request on first access.
appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync)
.. versionchanged:: 3.2
Merged with ``RequestContext``. The ``RequestContext`` alias will be
removed in Flask 4.0.
def __enter__(self) -> AppContext:
self.push()
return self
.. versionchanged:: 3.2
A combined app and request context is pushed for every request and CLI
command, rather than trying to detect if an app context is already
pushed.
def __exit__(
self,
exc_type: type | None,
exc_value: BaseException | None,
tb: TracebackType | None,
) -> None:
self.pop(exc_value)
class RequestContext:
"""The request context contains per-request information. The Flask
app creates and pushes it at the beginning of the request, then pops
it at the end of the request. 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. When using the interactive debugger, the context will be
restored so ``request`` is still accessible. Similarly, the test
client can preserve the context after the request ends. However,
teardown functions may already have closed some resources such as
database connections.
.. versionchanged:: 3.2
The session is loaded the first time it is accessed, rather than when
the context is pushed.
"""
def __init__(
self,
app: Flask,
environ: WSGIEnvironment,
*,
request: Request | None = None,
session: SessionMixin | None = None,
) -> None:
self.app = app
if request is None:
request = app.request_class(environ)
request.json_module = app.json
self.request: 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: list[tuple[str, str]] | None = None
self.session: SessionMixin | None = session
# Functions that should be executed after the request on the response
# object. These will be called before the regular "after_request"
# functions.
"""The application represented by this context. Accessed through
:data:`.current_app`.
"""
self.g: _AppCtxGlobals = app.app_ctx_globals_class()
"""The global data for this context. Accessed through :data:`.g`."""
self.url_adapter: MapAdapter | None = None
"""The URL adapter bound to the request, or the app if not in a request.
May be ``None`` if binding failed.
"""
self._request: Request | None = request
self._session: SessionMixin | None = session
self._flashes: list[tuple[str, str]] | None = None
self._after_request_functions: list[ft.AfterRequestCallable[t.Any]] = []
self._cv_tokens: list[
tuple[contextvars.Token[RequestContext], AppContext | None]
] = []
try:
self.url_adapter = app.create_url_adapter(self._request)
except HTTPException as e:
if self._request is not None:
self._request.routing_exception = e
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.
self._cv_token: contextvars.Token[AppContext] | None = None
"""The previous state to restore when popping."""
.. versionadded:: 0.10
self._push_count: int = 0
"""Track nested pushes of this context. Cleanup will only run once the
original push has been popped.
"""
@classmethod
def from_environ(cls, app: Flask, environ: WSGIEnvironment, /) -> te.Self:
"""Create an app context with request data from the given WSGI environ.
:param app: The application this context represents.
:param environ: The request data this context represents.
"""
request = app.request_class(environ)
request.json_module = app.json
return cls(app, request=request)
@property
def has_request(self) -> bool:
"""True if this context was created with request data."""
return self._request is not None
def copy(self) -> te.Self:
"""Create a new context with the same data objects as this context. See
:func:`.copy_current_request_context`.
.. 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.
The current session data is used instead of reloading the original data.
.. versionadded:: 0.10
"""
return self.__class__(
self.app,
environ=self.request.environ,
request=self.request,
session=self.session,
request=self._request,
session=self._session,
)
@property
def request(self) -> Request:
"""The request object associated with this context. Accessed through
:data:`.request`. Only available in request contexts, otherwise raises
:exc:`RuntimeError`.
"""
if self._request is None:
raise RuntimeError("There is no request in this context.")
return self._request
@property
def session(self) -> SessionMixin:
"""The session object associated with this context. Accessed through
:data:`.session`. Only available in request contexts, otherwise raises
:exc:`RuntimeError`.
"""
if self._request is None:
raise RuntimeError("There is no request in this context.")
if self._session is None:
si = self.app.session_interface
self._session = si.open_session(self.app, self.request)
if self._session is None:
self._session = si.make_null_session(self.app)
return self._session
def match_request(self) -> None:
"""Can be overridden by a subclass to hook into the matching
of the request.
"""Apply routing to the current request, storing either the matched
endpoint and args, or a routing exception.
"""
try:
result = self.url_adapter.match(return_rule=True) # type: ignore
self.request.url_rule, self.request.view_args = result # type: ignore
result = self.url_adapter.match(return_rule=True) # type: ignore[union-attr]
except HTTPException as e:
self.request.routing_exception = e
self._request.routing_exception = e # type: ignore[union-attr]
else:
self._request.url_rule, self._request.view_args = result # type: ignore[union-attr]
def push(self) -> None:
# Before we push the request context we have to ensure that there
# is an application context.
app_ctx = _cv_app.get(None)
"""Push this context so that it is the active context. If this is a
request context, calls :meth:`match_request` to perform routing with
the context active.
if app_ctx is None or app_ctx.app is not self.app:
app_ctx = self.app.app_context()
app_ctx.push()
else:
app_ctx = None
Typically, this is not used directly. Instead, use a ``with`` block
to manage the context.
self._cv_tokens.append((_cv_request.set(self), app_ctx))
In some situations, such as streaming or testing, the context may be
pushed multiple times. It will only trigger matching and signals if it
is not currently pushed.
"""
self._push_count += 1
# 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._cv_token is not None:
return
if self.session is None:
self.session = session_interface.make_null_session(self.app)
self._cv_token = _cv_app.set(self)
appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync)
# 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:
if self._request is not None and self.url_adapter is not None:
self.match_request()
def pop(self, exc: BaseException | None = _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.
def pop(self, exc: BaseException | None = None) -> None:
"""Pop this context so that it is no longer the active context. Then
call teardown functions and signals.
Typically, this is not used directly. Instead, use a ``with`` block
to manage the context.
This context must currently be the active context, otherwise a
:exc:`RuntimeError` is raised. In some situations, such as streaming or
testing, the context may have been pushed multiple times. It will only
trigger cleanup once it has been popped as many times as it was pushed.
Until then, it will remain the active context.
:param exc: An unhandled exception that was raised while the context was
active. Passed to teardown functions.
.. versionchanged:: 0.9
Added the `exc` argument.
Added the ``exc`` argument.
"""
clear_request = len(self._cv_tokens) == 1
if self._cv_token is None:
raise RuntimeError(f"Cannot pop this context ({self!r}), it is not pushed.")
ctx = _cv_app.get(None)
if ctx is None or self._cv_token is None:
raise RuntimeError(
f"Cannot pop this context ({self!r}), there is no active context."
)
if ctx is not self:
raise RuntimeError(
f"Cannot pop this context ({self!r}), it is not the active"
f" context ({ctx!r})."
)
self._push_count -= 1
if self._push_count > 0:
return
try:
if clear_request:
if exc is _sentinel:
exc = sys.exc_info()[1]
if self._request is not None:
self.app.do_teardown_request(exc)
request_close = getattr(self.request, "close", None)
if request_close is not None:
request_close()
self._request.close()
finally:
ctx = _cv_request.get()
token, app_ctx = self._cv_tokens.pop()
_cv_request.reset(token)
self.app.do_teardown_appcontext(exc)
_cv_app.reset(self._cv_token)
self._cv_token = None
appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync)
# 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:
ctx.request.environ["werkzeug.request"] = None
if app_ctx is not None:
app_ctx.pop(exc)
if ctx is not self:
raise AssertionError(
f"Popped wrong request context. ({ctx!r} instead of {self!r})"
)
def __enter__(self) -> RequestContext:
def __enter__(self) -> te.Self:
self.push()
return self
def __exit__(
self,
exc_type: type | None,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
tb: TracebackType | None,
) -> None:
self.pop(exc_value)
def __repr__(self) -> str:
return (
f"<{type(self).__name__} {self.request.url!r}"
f" [{self.request.method}] of {self.app.name}>"
if self._request is not None:
return (
f"<{type(self).__name__} {id(self)} of {self.app.name},"
f" {self.request.method} {self.request.url!r}>"
)
return f"<{type(self).__name__} {id(self)} of {self.app.name}>"
def __getattr__(name: str) -> t.Any:
import warnings
if name == "RequestContext":
warnings.warn(
"'RequestContext' has merged with 'AppContext', and will be removed"
" in Flask 4.0. Use 'AppContext' instead.",
DeprecationWarning,
stacklevel=2,
)
return AppContext
raise AttributeError(name)

7
src/flask/debughelpers.py
View file

@ -6,7 +6,7 @@ from jinja2.loaders import BaseLoader
from werkzeug.routing import RequestRedirect
from .blueprints import Blueprint
from .globals import request_ctx
from .globals import _cv_app
from .sansio.app import App
if t.TYPE_CHECKING:
@ -136,8 +136,9 @@ def explain_template_loading_attempts(
info = [f"Locating template {template!r}:"]
total_found = 0
blueprint = None
if request_ctx and request_ctx.request.blueprint is not None:
blueprint = request_ctx.request.blueprint
if (ctx := _cv_app.get(None)) is not None and ctx.has_request:
blueprint = ctx.request.blueprint
for idx, (loader, srcobj, triple) in enumerate(attempts):
if isinstance(srcobj, App):

62
src/flask/globals.py
View file

@ -9,43 +9,69 @@ if t.TYPE_CHECKING: # pragma: no cover
from .app import Flask
from .ctx import _AppCtxGlobals
from .ctx import AppContext
from .ctx import RequestContext
from .sessions import SessionMixin
from .wrappers import Request
T = t.TypeVar("T", covariant=True)
class ProxyMixin(t.Protocol[T]):
def _get_current_object(self) -> T: ...
# These subclasses inform type checkers that the proxy objects look like the
# proxied type along with the _get_current_object method.
class FlaskProxy(ProxyMixin[Flask], Flask): ...
class AppContextProxy(ProxyMixin[AppContext], AppContext): ...
class _AppCtxGlobalsProxy(ProxyMixin[_AppCtxGlobals], _AppCtxGlobals): ...
class RequestProxy(ProxyMixin[Request], Request): ...
class SessionMixinProxy(ProxyMixin[SessionMixin], SessionMixin): ...
_no_app_msg = """\
Working outside of application context.
This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.\
Attempted to use functionality that expected a current application to be set. To
solve this, set up an app context using 'with app.app_context()'. See the
documentation on app context for more information.\
"""
_cv_app: ContextVar[AppContext] = ContextVar("flask.app_ctx")
app_ctx: AppContext = LocalProxy( # type: ignore[assignment]
app_ctx: AppContextProxy = LocalProxy( # type: ignore[assignment]
_cv_app, unbound_message=_no_app_msg
)
current_app: Flask = LocalProxy( # type: ignore[assignment]
current_app: FlaskProxy = LocalProxy( # type: ignore[assignment]
_cv_app, "app", unbound_message=_no_app_msg
)
g: _AppCtxGlobals = LocalProxy( # type: ignore[assignment]
g: _AppCtxGlobalsProxy = LocalProxy( # type: ignore[assignment]
_cv_app, "g", unbound_message=_no_app_msg
)
_no_req_msg = """\
Working outside of request context.
This typically means that you attempted to use functionality that needed
an active HTTP request. Consult the documentation on testing for
information about how to avoid this problem.\
Attempted to use functionality that expected an active HTTP request. See the
documentation on request context for more information.\
"""
_cv_request: ContextVar[RequestContext] = ContextVar("flask.request_ctx")
request_ctx: RequestContext = LocalProxy( # type: ignore[assignment]
_cv_request, unbound_message=_no_req_msg
request: RequestProxy = LocalProxy( # type: ignore[assignment]
_cv_app, "request", unbound_message=_no_req_msg
)
request: Request = LocalProxy( # type: ignore[assignment]
_cv_request, "request", unbound_message=_no_req_msg
)
session: SessionMixin = LocalProxy( # type: ignore[assignment]
_cv_request, "session", unbound_message=_no_req_msg
session: SessionMixinProxy = LocalProxy( # type: ignore[assignment]
_cv_app, "session", unbound_message=_no_req_msg
)
def __getattr__(name: str) -> t.Any:
import warnings
if name == "request_ctx":
warnings.warn(
"'request_ctx' has merged with 'app_ctx', and will be removed"
" in Flask 4.0. Use 'app_ctx' instead.",
DeprecationWarning,
stacklevel=2,
)
return app_ctx
raise AttributeError(name)

50
src/flask/helpers.py
View file

@ -14,10 +14,9 @@ from werkzeug.utils import redirect as _wz_redirect
from werkzeug.wrappers import Response as BaseResponse
from .globals import _cv_app
from .globals import _cv_request
from .globals import app_ctx
from .globals import current_app
from .globals import request
from .globals import request_ctx
from .globals import session
from .signals import message_flashed
@ -64,7 +63,7 @@ def stream_with_context(
generator_or_function: t.Iterator[t.AnyStr] | t.Callable[..., t.Iterator[t.AnyStr]],
) -> t.Iterator[t.AnyStr] | t.Callable[[t.Iterator[t.AnyStr]], t.Iterator[t.AnyStr]]:
"""Wrap a response generator function so that it runs inside the current
request context. This keeps :data:`request`, :data:`session`, and :data:`g`
request context. This keeps :data:`.request`, :data:`.session`, and :data:`.g`
available, even though at the point the generator runs the request context
will typically have ended.
@ -112,22 +111,15 @@ def stream_with_context(
return update_wrapper(decorator, generator_or_function) # type: ignore[arg-type]
def generator() -> t.Iterator[t.AnyStr]:
if (req_ctx := _cv_request.get(None)) is None:
if (ctx := _cv_app.get(None)) is None:
raise RuntimeError(
"'stream_with_context' can only be used when a request"
" context is active, such as in a view function."
)
app_ctx = _cv_app.get()
# Setup code below will run the generator to this point, so that the
# current contexts are recorded. The contexts must be pushed after,
# otherwise their ContextVar will record the wrong event loop during
# async view functions.
yield None # type: ignore[misc]
with ctx:
yield None # type: ignore[misc]
# Push the app context first, so that the request context does not
# automatically create and push a different app context.
with app_ctx, req_ctx:
try:
yield from gen
finally:
@ -135,9 +127,9 @@ def stream_with_context(
if hasattr(gen, "close"):
gen.close()
# Execute the generator to the sentinel value. This ensures the context is
# preserved in the generator's state. Further iteration will push the
# context and yield from the original iterator.
# Execute the generator to the sentinel value. This captures the current
# context and pushes it to preserve it. Further iteration will yield from
# the original iterator.
wrapped_g = generator()
next(wrapped_g)
return wrapped_g
@ -264,8 +256,8 @@ def redirect(
Calls ``current_app.redirect`` if available instead of always
using Werkzeug's default ``redirect``.
"""
if current_app:
return current_app.redirect(location, code=code)
if (ctx := _cv_app.get(None)) is not None:
return ctx.app.redirect(location, code=code)
return _wz_redirect(location, code=code, Response=Response)
@ -287,8 +279,8 @@ def abort(code: int | BaseResponse, *args: t.Any, **kwargs: t.Any) -> t.NoReturn
Calls ``current_app.aborter`` if available instead of always
using Werkzeug's default ``abort``.
"""
if current_app:
current_app.aborter(code, *args, **kwargs)
if (ctx := _cv_app.get(None)) is not None:
ctx.app.aborter(code, *args, **kwargs)
_wz_abort(code, *args, **kwargs)
@ -340,7 +332,7 @@ def flash(message: str, category: str = "message") -> None:
flashes = session.get("_flashes", [])
flashes.append((category, message))
session["_flashes"] = flashes
app = current_app._get_current_object() # type: ignore
app = current_app._get_current_object()
message_flashed.send(
app,
_async_wrapper=app.ensure_sync,
@ -380,10 +372,10 @@ def get_flashed_messages(
:param category_filter: filter of categories to limit return values. Only
categories in the list will be returned.
"""
flashes = request_ctx.flashes
flashes = app_ctx._flashes
if flashes is None:
flashes = session.pop("_flashes") if "_flashes" in session else []
request_ctx.flashes = flashes
app_ctx._flashes = flashes
if category_filter:
flashes = list(filter(lambda f: f[0] in category_filter, flashes))
if not with_categories:
@ -392,14 +384,16 @@ def get_flashed_messages(
def _prepare_send_file_kwargs(**kwargs: t.Any) -> dict[str, t.Any]:
ctx = app_ctx._get_current_object()
if kwargs.get("max_age") is None:
kwargs["max_age"] = current_app.get_send_file_max_age
kwargs["max_age"] = ctx.app.get_send_file_max_age
kwargs.update(
environ=request.environ,
use_x_sendfile=current_app.config["USE_X_SENDFILE"],
response_class=current_app.response_class,
_root_path=current_app.root_path,
environ=ctx.request.environ,
use_x_sendfile=ctx.app.config["USE_X_SENDFILE"],
response_class=ctx.app.response_class,
_root_path=ctx.app.root_path,
)
return kwargs

2
src/flask/json/__init__.py
View file

@ -141,7 +141,7 @@ def jsonify(*args: t.Any, **kwargs: t.Any) -> Response:
mimetype. A dict or list returned from a view will be converted to a
JSON response automatically without needing to call this.
This requires an active request or application context, and calls
This requires an active app context, and calls
:meth:`app.json.response() <flask.json.provider.JSONProvider.response>`.
In debug mode, the output is formatted with indentation to make it

16
src/flask/sansio/app.py
View file

@ -177,11 +177,8 @@ class App(Scaffold):
#: 3. Return None instead of AttributeError on unexpected attributes.
#: 4. Raise exception if an unexpected attr is set, a "controlled" flask.g.
#:
#: In Flask 0.9 this property was called `request_globals_class` but it
#: was changed in 0.10 to :attr:`app_ctx_globals_class` because the
#: flask.g object is now application context scoped.
#:
#: .. versionadded:: 0.10
#: Renamed from ``request_globals_class`.
app_ctx_globals_class = _AppCtxGlobals
#: The class that is used for the ``config`` attribute of this app.
@ -825,10 +822,9 @@ class App(Scaffold):
@setupmethod
def teardown_appcontext(self, f: T_teardown) -> T_teardown:
"""Registers a function to be called when the application
context is popped. The application context is typically popped
after the request context for each request, at the end of CLI
commands, or after a manually pushed context ends.
"""Registers a function to be called when the app context is popped. The
context is popped at the end of a request, CLI command, or manual ``with``
block.
.. code-block:: python
@ -837,9 +833,7 @@ class App(Scaffold):
When the ``with`` block exits (or ``ctx.pop()`` is called), the
teardown functions are called just before the app context is
made inactive. Since a request context typically also manages an
application context it would also be called when you pop a
request context.
made inactive.
When a teardown function was called because of an unhandled
exception it will be passed an error object. If an

4
src/flask/sansio/scaffold.py
View file

@ -507,8 +507,8 @@ class Scaffold:
@setupmethod
def teardown_request(self, f: T_teardown) -> T_teardown:
"""Register a function to be called when the request context is
popped. Typically this happens at the end of each request, but
contexts may be pushed manually as well during testing.
popped. Typically, this happens at the end of each request, but
contexts may be pushed manually during testing.
.. code-block:: python

34
src/flask/templating.py
View file

@ -8,9 +8,7 @@ from jinja2 import Template
from jinja2 import TemplateNotFound
from .globals import _cv_app
from .globals import _cv_request
from .globals import current_app
from .globals import request
from .helpers import stream_with_context
from .signals import before_render_template
from .signals import template_rendered
@ -25,14 +23,16 @@ def _default_template_ctx_processor() -> dict[str, t.Any]:
"""Default template context processor. Injects `request`,
`session` and `g`.
"""
appctx = _cv_app.get(None)
reqctx = _cv_request.get(None)
ctx = _cv_app.get(None)
rv: dict[str, t.Any] = {}
if appctx is not None:
rv["g"] = appctx.g
if reqctx is not None:
rv["request"] = reqctx.request
rv["session"] = reqctx.session
if ctx is not None:
rv["g"] = ctx.g
if ctx.has_request:
rv["request"] = ctx.request
rv["session"] = ctx.session
return rv
@ -145,7 +145,7 @@ def render_template(
a list is given, the first name to exist will be rendered.
:param context: The variables to make available in the template.
"""
app = current_app._get_current_object() # type: ignore[attr-defined]
app = current_app._get_current_object()
template = app.jinja_env.get_or_select_template(template_name_or_list)
return _render(app, template, context)
@ -157,7 +157,7 @@ def render_template_string(source: str, **context: t.Any) -> str:
:param source: The source code of the template to render.
:param context: The variables to make available in the template.
"""
app = current_app._get_current_object() # type: ignore[attr-defined]
app = current_app._get_current_object()
template = app.jinja_env.from_string(source)
return _render(app, template, context)
@ -176,13 +176,7 @@ def _stream(
app, _async_wrapper=app.ensure_sync, template=template, context=context
)
rv = generate()
# If a request context is active, keep it while generating.
if request:
rv = stream_with_context(rv)
return rv
return stream_with_context(generate())
def stream_template(
@ -199,7 +193,7 @@ def stream_template(
.. versionadded:: 2.2
"""
app = current_app._get_current_object() # type: ignore[attr-defined]
app = current_app._get_current_object()
template = app.jinja_env.get_or_select_template(template_name_or_list)
return _stream(app, template, context)
@ -214,6 +208,6 @@ def stream_template_string(source: str, **context: t.Any) -> t.Iterator[str]:
.. versionadded:: 2.2
"""
app = current_app._get_current_object() # type: ignore[attr-defined]
app = current_app._get_current_object()
template = app.jinja_env.from_string(source)
return _stream(app, template, context)

8
src/flask/testing.py
View file

@ -107,10 +107,10 @@ def _get_werkzeug_version() -> str:
class FlaskClient(Client):
"""Works like a regular Werkzeug test client but has knowledge about
Flask's contexts to defer the cleanup of the request context until
the end of a ``with`` block. For general information about how to
use this class refer to :class:`werkzeug.test.Client`.
"""Works like a regular Werkzeug test client, with additional behavior for
Flask. Can defer the cleanup of the request context until the end of a
``with`` block. For general information about how to use this class refer to
:class:`werkzeug.test.Client`.
.. versionchanged:: 0.12
`app.test_client()` includes preset default environment, which can be