back to Reference (Gold) summary
Reference (Gold): flask
Pytest Summary for test tests
| status | count |
|---|---|
| passed | 477 |
| failed | 5 |
| skipped | 2 |
| total | 484 |
| collected | 484 |
Failed pytests:
test_logging.py::test_logger
test_logging.py::test_logger
app =def test_logger(app): assert app.logger.name == "flask_test" assert app.logger.level == logging.NOTSET > assert app.logger.handlers == [default_handler] E AssertionError: assert [] == [ (NOTSET)>] E E Right contains one more item: (NOTSET)> E Use -v to get more diff tests/test_logging.py:39: AssertionError
test_logging.py::test_logger_debug
test_logging.py::test_logger_debug
app =def test_logger_debug(app): app.debug = True assert app.logger.level == logging.DEBUG > assert app.logger.handlers == [default_handler] E AssertionError: assert [] == [ (NOTSET)>] E E Right contains one more item: (NOTSET)> E Use -v to get more diff tests/test_logging.py:45: AssertionError
test_logging.py::test_wsgi_errors_stream
test_logging.py::test_wsgi_errors_stream
app =, client = > def test_wsgi_errors_stream(app, client): @app.route("/") def index(): app.logger.error("test") return "" stream = StringIO() client.get("/", errors_stream=stream) > assert "ERROR in test_logging: test" in stream.getvalue() E AssertionError: assert 'ERROR in test_logging: test' in '' E + where '' = () E + where = <_io.StringIO object at 0x7f3a6e95f340>.getvalue tests/test_logging.py:62: AssertionError
test_logging.py::test_has_level_handler
test_logging.py::test_has_level_handler
def test_has_level_handler():
logger = logging.getLogger("flask.app")
> assert not has_level_handler(logger)
E assert not True
E + where True = has_level_handler()
tests/test_logging.py:72: AssertionError
test_logging.py::test_log_view_exception
test_logging.py::test_log_view_exception
app =, client = > def test_log_view_exception(app, client): @app.route("/") def index(): raise Exception("test") app.testing = False stream = StringIO() rv = client.get("/", errors_stream=stream) assert rv.status_code == 500 assert rv.data err = stream.getvalue() > assert "Exception on / [GET]" in err E AssertionError: assert 'Exception on / [GET]' in '' tests/test_logging.py:97: AssertionError
Patch diff
diff --git a/src/flask/app.py b/src/flask/app.py
index 3e76b0ba..7622b5e8 100644
--- a/src/flask/app.py
+++ b/src/flask/app.py
@@ -1,4 +1,5 @@
from __future__ import annotations
+
import collections.abc as cabc
import os
import sys
@@ -9,6 +10,7 @@ from inspect import iscoroutinefunction
from itertools import chain
from types import TracebackType
from urllib.parse import quote as _url_quote
+
import click
from werkzeug.datastructures import Headers
from werkzeug.datastructures import ImmutableDict
@@ -22,6 +24,7 @@ from werkzeug.routing import RoutingException
from werkzeug.routing import Rule
from werkzeug.serving import is_running_from_reloader
from werkzeug.wrappers import Response as BaseResponse
+
from . import cli
from . import typing as ft
from .ctx import AppContext
@@ -49,19 +52,28 @@ from .signals import request_tearing_down
from .templating import Environment
from .wrappers import Request
from .wrappers import Response
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from _typeshed.wsgi import StartResponse
from _typeshed.wsgi import WSGIEnvironment
+
from .testing import FlaskClient
from .testing import FlaskCliRunner
-T_shell_context_processor = t.TypeVar('T_shell_context_processor', bound=ft
- .ShellContextProcessorCallable)
-T_teardown = t.TypeVar('T_teardown', bound=ft.TeardownCallable)
-T_template_filter = t.TypeVar('T_template_filter', bound=ft.
- TemplateFilterCallable)
-T_template_global = t.TypeVar('T_template_global', bound=ft.
- TemplateGlobalCallable)
-T_template_test = t.TypeVar('T_template_test', bound=ft.TemplateTestCallable)
+
+T_shell_context_processor = t.TypeVar(
+ "T_shell_context_processor", bound=ft.ShellContextProcessorCallable
+)
+T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable)
+T_template_filter = t.TypeVar("T_template_filter", bound=ft.TemplateFilterCallable)
+T_template_global = t.TypeVar("T_template_global", bound=ft.TemplateGlobalCallable)
+T_template_test = t.TypeVar("T_template_test", bound=ft.TemplateTestCallable)
+
+
+def _make_timedelta(value: timedelta | int | None) -> timedelta | None:
+ if value is None or isinstance(value, timedelta):
+ return value
+
+ return timedelta(seconds=value)
class Flask(App):
@@ -160,45 +172,105 @@ class Flask(App):
This should only be set manually when it can't be detected
automatically, such as for namespace packages.
"""
- default_config = ImmutableDict({'DEBUG': None, 'TESTING': False,
- 'PROPAGATE_EXCEPTIONS': None, 'SECRET_KEY': None,
- 'PERMANENT_SESSION_LIFETIME': timedelta(days=31), 'USE_X_SENDFILE':
- False, 'SERVER_NAME': None, 'APPLICATION_ROOT': '/',
- 'SESSION_COOKIE_NAME': 'session', 'SESSION_COOKIE_DOMAIN': None,
- 'SESSION_COOKIE_PATH': None, 'SESSION_COOKIE_HTTPONLY': True,
- 'SESSION_COOKIE_SECURE': False, 'SESSION_COOKIE_SAMESITE': None,
- 'SESSION_REFRESH_EACH_REQUEST': True, 'MAX_CONTENT_LENGTH': None,
- 'SEND_FILE_MAX_AGE_DEFAULT': None, 'TRAP_BAD_REQUEST_ERRORS': None,
- 'TRAP_HTTP_EXCEPTIONS': False, 'EXPLAIN_TEMPLATE_LOADING': False,
- 'PREFERRED_URL_SCHEME': 'http', 'TEMPLATES_AUTO_RELOAD': None,
- 'MAX_COOKIE_SIZE': 4093})
+
+ default_config = ImmutableDict(
+ {
+ "DEBUG": None,
+ "TESTING": False,
+ "PROPAGATE_EXCEPTIONS": None,
+ "SECRET_KEY": None,
+ "PERMANENT_SESSION_LIFETIME": timedelta(days=31),
+ "USE_X_SENDFILE": False,
+ "SERVER_NAME": None,
+ "APPLICATION_ROOT": "/",
+ "SESSION_COOKIE_NAME": "session",
+ "SESSION_COOKIE_DOMAIN": None,
+ "SESSION_COOKIE_PATH": None,
+ "SESSION_COOKIE_HTTPONLY": True,
+ "SESSION_COOKIE_SECURE": False,
+ "SESSION_COOKIE_SAMESITE": None,
+ "SESSION_REFRESH_EACH_REQUEST": True,
+ "MAX_CONTENT_LENGTH": None,
+ "SEND_FILE_MAX_AGE_DEFAULT": None,
+ "TRAP_BAD_REQUEST_ERRORS": None,
+ "TRAP_HTTP_EXCEPTIONS": False,
+ "EXPLAIN_TEMPLATE_LOADING": False,
+ "PREFERRED_URL_SCHEME": "http",
+ "TEMPLATES_AUTO_RELOAD": None,
+ "MAX_COOKIE_SIZE": 4093,
+ }
+ )
+
+ #: The class that is used for request objects. See :class:`~flask.Request`
+ #: for more information.
request_class: type[Request] = Request
+
+ #: The class that is used for response objects. See
+ #: :class:`~flask.Response` for more information.
response_class: type[Response] = Response
+
+ #: the session interface to use. By default an instance of
+ #: :class:`~flask.sessions.SecureCookieSessionInterface` is used here.
+ #:
+ #: .. versionadded:: 0.8
session_interface: SessionInterface = SecureCookieSessionInterface()
- def __init__(self, import_name: str, static_url_path: (str | None)=None,
- static_folder: (str | os.PathLike[str] | None)='static',
- static_host: (str | None)=None, host_matching: bool=False,
- subdomain_matching: bool=False, template_folder: (str | os.PathLike
- [str] | None)='templates', instance_path: (str | None)=None,
- instance_relative_config: bool=False, root_path: (str | None)=None):
- super().__init__(import_name=import_name, static_url_path=
- static_url_path, static_folder=static_folder, static_host=
- static_host, host_matching=host_matching, subdomain_matching=
- subdomain_matching, template_folder=template_folder,
- instance_path=instance_path, instance_relative_config=
- instance_relative_config, root_path=root_path)
+ def __init__(
+ self,
+ import_name: str,
+ static_url_path: str | None = None,
+ static_folder: str | os.PathLike[str] | None = "static",
+ static_host: str | None = None,
+ host_matching: bool = False,
+ subdomain_matching: bool = False,
+ template_folder: str | os.PathLike[str] | None = "templates",
+ instance_path: str | None = None,
+ instance_relative_config: bool = False,
+ root_path: str | None = None,
+ ):
+ super().__init__(
+ import_name=import_name,
+ static_url_path=static_url_path,
+ static_folder=static_folder,
+ static_host=static_host,
+ host_matching=host_matching,
+ subdomain_matching=subdomain_matching,
+ template_folder=template_folder,
+ instance_path=instance_path,
+ instance_relative_config=instance_relative_config,
+ root_path=root_path,
+ )
+
+ #: The Click command group for registering CLI commands for this
+ #: object. The commands are available from the ``flask`` command
+ #: once the application has been discovered and blueprints have
+ #: been registered.
self.cli = cli.AppGroup()
+
+ # Set the name of the Click group in case someone wants to add
+ # the app's commands to another CLI tool.
self.cli.name = self.name
+
+ # Add a static route using the provided static_url_path, static_host,
+ # and static_folder if there is a configured static_folder.
+ # Note we do this without checking if static_folder exists.
+ # For one, it might be created while the server is running (e.g. during
+ # development). Also, Google App Engine stores static files somewhere
if self.has_static_folder:
- assert bool(static_host
- ) == host_matching, 'Invalid static_host/host_matching combination'
+ assert (
+ bool(static_host) == host_matching
+ ), "Invalid static_host/host_matching combination"
+ # Use a weakref to avoid creating a reference cycle between the app
+ # and the view function (see #3761).
self_ref = weakref.ref(self)
- self.add_url_rule(f'{self.static_url_path}/<path:filename>',
- endpoint='static', host=static_host, view_func=lambda **kw:
- self_ref().send_static_file(**kw))
-
- def get_send_file_max_age(self, filename: (str | None)) ->(int | None):
+ self.add_url_rule(
+ f"{self.static_url_path}/<path:filename>",
+ endpoint="static",
+ host=static_host,
+ view_func=lambda **kw: self_ref().send_static_file(**kw), # type: ignore # noqa: B950
+ )
+
+ def get_send_file_max_age(self, filename: str | None) -> int | None:
"""Used by :func:`send_file` to determine the ``max_age`` cache
value for a given file path if it wasn't passed.
@@ -215,9 +287,17 @@ class Flask(App):
.. versionadded:: 0.9
"""
- pass
+ value = current_app.config["SEND_FILE_MAX_AGE_DEFAULT"]
+
+ if value is None:
+ return None
+
+ if isinstance(value, timedelta):
+ return int(value.total_seconds())
- def send_static_file(self, filename: str) ->Response:
+ return value # type: ignore[no-any-return]
+
+ def send_static_file(self, filename: str) -> Response:
"""The view function used to serve files from
:attr:`static_folder`. A route is automatically registered for
this view at :attr:`static_url_path` if :attr:`static_folder` is
@@ -229,9 +309,17 @@ class Flask(App):
.. versionadded:: 0.5
"""
- pass
+ if not self.has_static_folder:
+ raise RuntimeError("'static_folder' must be set to serve static_files.")
+
+ # send_file only knows to call get_send_file_max_age on the app,
+ # call it here so it works for blueprints too.
+ max_age = self.get_send_file_max_age(filename)
+ return send_from_directory(
+ t.cast(str, self.static_folder), filename, max_age=max_age
+ )
- def open_resource(self, resource: str, mode: str='rb') ->t.IO[t.AnyStr]:
+ def open_resource(self, resource: str, mode: str = "rb") -> t.IO[t.AnyStr]:
"""Open a resource file relative to :attr:`root_path` for
reading.
@@ -253,10 +341,12 @@ class Flask(App):
class.
"""
- pass
+ if mode not in {"r", "rt", "rb"}:
+ raise ValueError("Resources can only be opened for reading.")
- def open_instance_resource(self, resource: str, mode: str='rb') ->t.IO[t
- .AnyStr]:
+ return open(os.path.join(self.root_path, resource), mode)
+
+ def open_instance_resource(self, resource: str, mode: str = "rb") -> t.IO[t.AnyStr]:
"""Opens a resource from the application's instance folder
(:attr:`instance_path`). Otherwise works like
:meth:`open_resource`. Instance resources can also be opened for
@@ -266,9 +356,9 @@ class Flask(App):
subfolders use forward slashes as separator.
:param mode: resource file opening mode, default is 'rb'.
"""
- pass
+ return open(os.path.join(self.instance_path, resource), mode)
- def create_jinja_environment(self) ->Environment:
+ def create_jinja_environment(self) -> Environment:
"""Create the Jinja environment based on :attr:`jinja_options`
and the various Jinja-related methods of the app. Changing
:attr:`jinja_options` after this will have no effect. Also adds
@@ -280,10 +370,35 @@ class Flask(App):
.. versionadded:: 0.5
"""
- pass
-
- def create_url_adapter(self, request: (Request | None)) ->(MapAdapter |
- None):
+ options = dict(self.jinja_options)
+
+ if "autoescape" not in options:
+ options["autoescape"] = self.select_jinja_autoescape
+
+ if "auto_reload" not in options:
+ auto_reload = self.config["TEMPLATES_AUTO_RELOAD"]
+
+ if auto_reload is None:
+ auto_reload = self.debug
+
+ options["auto_reload"] = auto_reload
+
+ rv = self.jinja_environment(self, **options)
+ rv.globals.update(
+ url_for=self.url_for,
+ get_flashed_messages=get_flashed_messages,
+ config=self.config,
+ # request, session and g are normally added with the
+ # context processor for efficiency reasons but for imported
+ # templates we also want the proxies in there.
+ request=request,
+ session=session,
+ g=g,
+ )
+ rv.policies["json.dumps_function"] = self.json.dumps
+ return rv
+
+ def create_url_adapter(self, request: Request | None) -> MapAdapter | None:
"""Creates a URL adapter for the given request. The URL adapter
is created at a point where the request context is not yet set
up so the request is passed explicitly.
@@ -298,9 +413,32 @@ class Flask(App):
:data:`SERVER_NAME` no longer implicitly enables subdomain
matching. Use :attr:`subdomain_matching` instead.
"""
- pass
-
- def raise_routing_exception(self, request: Request) ->t.NoReturn:
+ if request is not None:
+ # If subdomain matching is disabled (the default), use the
+ # default subdomain in all cases. This should be the default
+ # in Werkzeug but it currently does not have that feature.
+ if not self.subdomain_matching:
+ subdomain = self.url_map.default_subdomain or None
+ else:
+ subdomain = None
+
+ return self.url_map.bind_to_environ(
+ request.environ,
+ server_name=self.config["SERVER_NAME"],
+ subdomain=subdomain,
+ )
+ # We need at the very least the server name to be set for this
+ # to work.
+ if self.config["SERVER_NAME"] is not None:
+ return self.url_map.bind(
+ self.config["SERVER_NAME"],
+ script_name=self.config["APPLICATION_ROOT"],
+ url_scheme=self.config["PREFERRED_URL_SCHEME"],
+ )
+
+ return None
+
+ def raise_routing_exception(self, request: Request) -> t.NoReturn:
"""Intercept routing exceptions and possibly do something else.
In debug mode, intercept a routing redirect and replace it with
@@ -316,9 +454,19 @@ class Flask(App):
:meta private:
:internal:
"""
- pass
+ if (
+ not self.debug
+ or not isinstance(request.routing_exception, RequestRedirect)
+ or request.routing_exception.code in {307, 308}
+ or request.method in {"GET", "HEAD", "OPTIONS"}
+ ):
+ raise request.routing_exception # type: ignore[misc]
+
+ from .debughelpers import FormDataRoutingRedirect
- def update_template_context(self, context: dict[str, t.Any]) ->None:
+ raise FormDataRoutingRedirect(request)
+
+ def update_template_context(self, context: dict[str, t.Any]) -> None:
"""Update the template context with some commonly used variables.
This injects request, session, config and g into the template
context as well as everything template context processors want
@@ -329,19 +477,43 @@ class Flask(App):
:param context: the context as a dictionary that is updated in place
to add extra variables.
"""
- pass
+ names: t.Iterable[str | None] = (None,)
+
+ # A template may be rendered outside a request context.
+ if request:
+ names = chain(names, reversed(request.blueprints))
- def make_shell_context(self) ->dict[str, t.Any]:
+ # The values passed to render_template take precedence. Keep a
+ # copy to re-apply after all context functions.
+ orig_ctx = context.copy()
+
+ for name in names:
+ if name in self.template_context_processors:
+ for func in self.template_context_processors[name]:
+ context.update(self.ensure_sync(func)())
+
+ context.update(orig_ctx)
+
+ def make_shell_context(self) -> dict[str, t.Any]:
"""Returns the shell context for an interactive shell for this
application. This runs all the registered shell context
processors.
.. versionadded:: 0.11
"""
- pass
-
- def run(self, host: (str | None)=None, port: (int | None)=None, debug:
- (bool | None)=None, load_dotenv: bool=True, **options: t.Any) ->None:
+ rv = {"app": self, "g": g}
+ for processor in self.shell_context_processors:
+ rv.update(processor())
+ return rv
+
+ def run(
+ self,
+ host: str | None = None,
+ port: int | None = None,
+ debug: bool | None = None,
+ load_dotenv: bool = True,
+ **options: t.Any,
+ ) -> None:
"""Runs the application on a local development server.
Do not use ``run()`` in a production setting. It is not intended to
@@ -397,10 +569,67 @@ class Flask(App):
The default port is now picked from the ``SERVER_NAME``
variable.
"""
- pass
-
- def test_client(self, use_cookies: bool=True, **kwargs: t.Any
- ) ->FlaskClient:
+ # Ignore this call so that it doesn't start another server if
+ # the 'flask run' command is used.
+ if os.environ.get("FLASK_RUN_FROM_CLI") == "true":
+ if not is_running_from_reloader():
+ click.secho(
+ " * Ignoring a call to 'app.run()' that would block"
+ " the current 'flask' CLI command.\n"
+ " Only call 'app.run()' in an 'if __name__ =="
+ ' "__main__"\' guard.',
+ fg="red",
+ )
+
+ return
+
+ if get_load_dotenv(load_dotenv):
+ cli.load_dotenv()
+
+ # if set, env var overrides existing value
+ if "FLASK_DEBUG" in os.environ:
+ self.debug = get_debug_flag()
+
+ # debug passed to method overrides all other sources
+ if debug is not None:
+ self.debug = bool(debug)
+
+ server_name = self.config.get("SERVER_NAME")
+ sn_host = sn_port = None
+
+ if server_name:
+ sn_host, _, sn_port = server_name.partition(":")
+
+ if not host:
+ if sn_host:
+ host = sn_host
+ else:
+ host = "127.0.0.1"
+
+ if port or port == 0:
+ port = int(port)
+ elif sn_port:
+ port = int(sn_port)
+ else:
+ port = 5000
+
+ options.setdefault("use_reloader", self.debug)
+ options.setdefault("use_debugger", self.debug)
+ options.setdefault("threaded", True)
+
+ cli.show_server_banner(self.debug, self.name)
+
+ from werkzeug.serving import run_simple
+
+ try:
+ run_simple(t.cast(str, host), port, self, **options)
+ finally:
+ # reset the first request information if the development server
+ # reset normally. This makes it possible to restart the server
+ # without reloader and that stuff from an interactive shell.
+ self._got_first_request = False
+
+ def test_client(self, use_cookies: bool = True, **kwargs: t.Any) -> FlaskClient:
"""Creates a test client for this application. For information
about unit testing head over to :doc:`/testing`.
@@ -451,9 +680,14 @@ class Flask(App):
Added `**kwargs` to support passing additional keyword arguments to
the constructor of :attr:`test_client_class`.
"""
- pass
-
- def test_cli_runner(self, **kwargs: t.Any) ->FlaskCliRunner:
+ cls = self.test_client_class
+ if cls is None:
+ from .testing import FlaskClient as cls
+ return cls( # type: ignore
+ self, self.response_class, use_cookies=use_cookies, **kwargs
+ )
+
+ def test_cli_runner(self, **kwargs: t.Any) -> FlaskCliRunner:
"""Create a CLI runner for testing CLI commands.
See :ref:`testing-cli`.
@@ -463,10 +697,16 @@ class Flask(App):
.. versionadded:: 1.0
"""
- pass
+ cls = self.test_cli_runner_class
+
+ if cls is None:
+ from .testing import FlaskCliRunner as cls
+
+ return cls(self, **kwargs) # type: ignore
- def handle_http_exception(self, e: HTTPException) ->(HTTPException | ft
- .ResponseReturnValue):
+ def handle_http_exception(
+ self, e: HTTPException
+ ) -> HTTPException | ft.ResponseReturnValue:
"""Handles an HTTP exception. By default this will invoke the
registered error handlers and fall back to returning the
exception as response.
@@ -483,10 +723,25 @@ class Flask(App):
.. versionadded:: 0.3
"""
- pass
-
- def handle_user_exception(self, e: Exception) ->(HTTPException | ft.
- ResponseReturnValue):
+ # Proxy exceptions don't have error codes. We want to always return
+ # those unchanged as errors
+ if e.code is None:
+ return e
+
+ # RoutingExceptions are used internally to trigger routing
+ # actions, such as slash redirects raising RequestRedirect. They
+ # are not raised or handled in user code.
+ if isinstance(e, RoutingException):
+ return e
+
+ handler = self._find_error_handler(e, request.blueprints)
+ if handler is None:
+ return e
+ return self.ensure_sync(handler)(e) # type: ignore[no-any-return]
+
+ def handle_user_exception(
+ self, e: Exception
+ ) -> HTTPException | ft.ResponseReturnValue:
"""This method is called whenever an exception occurs that
should be handled. A special case is :class:`~werkzeug
.exceptions.HTTPException` which is forwarded to the
@@ -501,9 +756,22 @@ class Flask(App):
.. versionadded:: 0.7
"""
- pass
+ if isinstance(e, BadRequestKeyError) and (
+ self.debug or self.config["TRAP_BAD_REQUEST_ERRORS"]
+ ):
+ e.show_exception = True
+
+ if isinstance(e, HTTPException) and not self.trap_http_exception(e):
+ return self.handle_http_exception(e)
+
+ handler = self._find_error_handler(e, request.blueprints)
+
+ if handler is None:
+ raise
- def handle_exception(self, e: Exception) ->Response:
+ return self.ensure_sync(handler)(e) # type: ignore[no-any-return]
+
+ def handle_exception(self, e: Exception) -> Response:
"""Handle an exception that did not have an error handler
associated with it, or that was raised from an error handler.
This always causes a 500 ``InternalServerError``.
@@ -531,10 +799,35 @@ class Flask(App):
.. versionadded:: 0.3
"""
- pass
+ exc_info = sys.exc_info()
+ got_request_exception.send(self, _async_wrapper=self.ensure_sync, exception=e)
+ propagate = self.config["PROPAGATE_EXCEPTIONS"]
+
+ if propagate is None:
+ propagate = self.testing or self.debug
+
+ if propagate:
+ # Re-raise if called with an active exception, otherwise
+ # raise the passed in exception.
+ if exc_info[1] is e:
+ raise
+
+ raise e
- def log_exception(self, exc_info: (tuple[type, BaseException,
- TracebackType] | tuple[None, None, None])) ->None:
+ self.log_exception(exc_info)
+ server_error: InternalServerError | ft.ResponseReturnValue
+ server_error = InternalServerError(original_exception=e)
+ handler = self._find_error_handler(server_error, request.blueprints)
+
+ if handler is not None:
+ server_error = self.ensure_sync(handler)(server_error)
+
+ return self.finalize_request(server_error, from_error_handler=True)
+
+ def log_exception(
+ self,
+ exc_info: (tuple[type, BaseException, TracebackType] | tuple[None, None, None]),
+ ) -> None:
"""Logs an exception. This is called by :meth:`handle_exception`
if debugging is disabled and right before the handler is called.
The default implementation logs the exception as error on the
@@ -542,9 +835,11 @@ class Flask(App):
.. versionadded:: 0.8
"""
- pass
+ self.logger.error(
+ f"Exception on {request.path} [{request.method}]", exc_info=exc_info
+ )
- def dispatch_request(self) ->ft.ResponseReturnValue:
+ def dispatch_request(self) -> ft.ResponseReturnValue:
"""Does the request dispatching. Matches the URL and returns the
return value of the view or error handler. This does not have to
be a response object. In order to convert the return value to a
@@ -554,19 +849,44 @@ class Flask(App):
This no longer does the exception handling, this code was
moved to the new :meth:`full_dispatch_request`.
"""
- pass
-
- def full_dispatch_request(self) ->Response:
+ req = request_ctx.request
+ if req.routing_exception is not None:
+ self.raise_routing_exception(req)
+ rule: Rule = req.url_rule # type: ignore[assignment]
+ # if we provide automatic options for this URL and the
+ # request came with the OPTIONS method, reply automatically
+ if (
+ getattr(rule, "provide_automatic_options", False)
+ and req.method == "OPTIONS"
+ ):
+ return self.make_default_options_response()
+ # otherwise dispatch to the handler for that endpoint
+ view_args: dict[str, t.Any] = req.view_args # type: ignore[assignment]
+ return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args) # type: ignore[no-any-return]
+
+ def full_dispatch_request(self) -> Response:
"""Dispatches the request and on top of that performs request
pre and postprocessing as well as HTTP exception catching and
error handling.
.. versionadded:: 0.7
"""
- pass
-
- def finalize_request(self, rv: (ft.ResponseReturnValue | HTTPException),
- from_error_handler: bool=False) ->Response:
+ self._got_first_request = True
+
+ try:
+ request_started.send(self, _async_wrapper=self.ensure_sync)
+ rv = self.preprocess_request()
+ if rv is None:
+ rv = self.dispatch_request()
+ except Exception as e:
+ rv = self.handle_user_exception(e)
+ return self.finalize_request(rv)
+
+ def finalize_request(
+ self,
+ rv: ft.ResponseReturnValue | HTTPException,
+ from_error_handler: bool = False,
+ ) -> Response:
"""Given the return value from a view function this finalizes
the request by converting it into a response and invoking the
postprocessing functions. This is invoked for both normal
@@ -579,19 +899,34 @@ class Flask(App):
:internal:
"""
- pass
-
- def make_default_options_response(self) ->Response:
+ response = self.make_response(rv)
+ try:
+ response = self.process_response(response)
+ request_finished.send(
+ self, _async_wrapper=self.ensure_sync, response=response
+ )
+ except Exception:
+ if not from_error_handler:
+ raise
+ self.logger.exception(
+ "Request finalizing failed with an error while handling an error"
+ )
+ return response
+
+ def make_default_options_response(self) -> Response:
"""This method is called to create the default ``OPTIONS`` response.
This can be changed through subclassing to change the default
behavior of ``OPTIONS`` responses.
.. versionadded:: 0.7
"""
- pass
+ adapter = request_ctx.url_adapter
+ methods = adapter.allowed_methods() # type: ignore[union-attr]
+ rv = self.response_class()
+ rv.allow.update(methods)
+ return rv
- def ensure_sync(self, func: t.Callable[..., t.Any]) ->t.Callable[..., t.Any
- ]:
+ def ensure_sync(self, func: t.Callable[..., t.Any]) -> t.Callable[..., t.Any]:
"""Ensure that the function is synchronous for WSGI workers.
Plain ``def`` functions are returned as-is. ``async def``
functions are wrapped to run and wait for the response.
@@ -600,10 +935,14 @@ class Flask(App):
.. versionadded:: 2.0
"""
- pass
+ if iscoroutinefunction(func):
+ return self.async_to_sync(func)
- def async_to_sync(self, func: t.Callable[..., t.Coroutine[t.Any, t.Any,
- t.Any]]) ->t.Callable[..., t.Any]:
+ return func
+
+ def async_to_sync(
+ self, func: t.Callable[..., t.Coroutine[t.Any, t.Any, t.Any]]
+ ) -> t.Callable[..., t.Any]:
"""Return a sync function that will run the coroutine function.
.. code-block:: python
@@ -615,11 +954,26 @@ class Flask(App):
.. versionadded:: 2.0
"""
- pass
-
- def url_for(self, /, endpoint: str, *, _anchor: (str | None)=None,
- _method: (str | None)=None, _scheme: (str | None)=None, _external:
- (bool | None)=None, **values: t.Any) ->str:
+ try:
+ from asgiref.sync import async_to_sync as asgiref_async_to_sync
+ except ImportError:
+ raise RuntimeError(
+ "Install Flask with the 'async' extra in order to use async views."
+ ) from None
+
+ return asgiref_async_to_sync(func)
+
+ def url_for(
+ self,
+ /,
+ endpoint: str,
+ *,
+ _anchor: str | None = None,
+ _method: str | None = None,
+ _scheme: str | None = None,
+ _external: bool | None = None,
+ **values: t.Any,
+ ) -> str:
"""Generate a URL to the given endpoint with the given values.
This is called by :func:`flask.url_for`, and can be called
@@ -666,9 +1020,76 @@ class Flask(App):
.. versionadded:: 2.2
Moved from ``flask.url_for``, which calls this method.
"""
- pass
-
- def make_response(self, rv: ft.ResponseReturnValue) ->Response:
+ 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 the endpoint starts with "." and the request matches a
+ # blueprint, the endpoint is relative to the blueprint.
+ if endpoint[:1] == ".":
+ if blueprint_name is not None:
+ endpoint = f"{blueprint_name}{endpoint}"
+ else:
+ endpoint = endpoint[1:]
+
+ # When in a request, generate a URL without scheme and
+ # domain by default, unless a scheme is given.
+ 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
+ else:
+ url_adapter = self.create_url_adapter(None)
+
+ if url_adapter is None:
+ raise RuntimeError(
+ "Unable to build URLs outside an active request"
+ " without 'SERVER_NAME' configured. Also configure"
+ " 'APPLICATION_ROOT' and 'PREFERRED_URL_SCHEME' as"
+ " needed."
+ )
+
+ # When outside a request, generate a URL with scheme and
+ # domain by default.
+ if _external is None:
+ _external = True
+
+ # It is an error to set _scheme when _external=False, in order
+ # to avoid accidental insecure URLs.
+ if _scheme is not None and not _external:
+ raise ValueError("When specifying '_scheme', '_external' must be True.")
+
+ self.inject_url_defaults(endpoint, values)
+
+ try:
+ rv = url_adapter.build( # type: ignore[union-attr]
+ endpoint,
+ values,
+ method=_method,
+ url_scheme=_scheme,
+ force_external=_external,
+ )
+ except BuildError as error:
+ values.update(
+ _anchor=_anchor, _method=_method, _scheme=_scheme, _external=_external
+ )
+ return self.handle_url_build_error(error, endpoint, values)
+
+ if _anchor is not None:
+ _anchor = _url_quote(_anchor, safe="%!#$&'()*+,/:;=?@")
+ rv = f"{rv}#{_anchor}"
+
+ return rv
+
+ def make_response(self, rv: ft.ResponseReturnValue) -> Response:
"""Convert the return value from a view function to an instance of
:attr:`response_class`.
@@ -724,9 +1145,92 @@ class Flask(App):
Previously a tuple was interpreted as the arguments for the
response object.
"""
- pass
- def preprocess_request(self) ->(ft.ResponseReturnValue | None):
+ status = headers = None
+
+ # unpack tuple returns
+ if isinstance(rv, tuple):
+ len_rv = len(rv)
+
+ # a 3-tuple is unpacked directly
+ if len_rv == 3:
+ rv, status, headers = rv # type: ignore[misc]
+ # decide if a 2-tuple has status or headers
+ elif len_rv == 2:
+ if isinstance(rv[1], (Headers, dict, tuple, list)):
+ rv, headers = rv
+ else:
+ rv, status = rv # type: ignore[assignment,misc]
+ # other sized tuples are not allowed
+ else:
+ raise TypeError(
+ "The view function did not return a valid response tuple."
+ " The tuple must have the form (body, status, headers),"
+ " (body, status), or (body, headers)."
+ )
+
+ # the body must not be None
+ if rv is None:
+ raise TypeError(
+ f"The view function for {request.endpoint!r} did not"
+ " return a valid response. The function either returned"
+ " None or ended without a return statement."
+ )
+
+ # make sure the body is an instance of the response class
+ if not isinstance(rv, self.response_class):
+ if isinstance(rv, (str, bytes, bytearray)) or isinstance(rv, cabc.Iterator):
+ # let the response class set the status and headers instead of
+ # waiting to do it manually, so that the class can handle any
+ # special logic
+ rv = self.response_class(
+ rv,
+ status=status,
+ headers=headers, # type: ignore[arg-type]
+ )
+ status = headers = None
+ elif isinstance(rv, (dict, list)):
+ rv = self.json.response(rv)
+ elif isinstance(rv, BaseResponse) or callable(rv):
+ # evaluate a WSGI callable, or coerce a different response
+ # class to the correct type
+ try:
+ rv = self.response_class.force_type(
+ rv, # type: ignore[arg-type]
+ request.environ,
+ )
+ except TypeError as e:
+ raise TypeError(
+ f"{e}\nThe view function did not return a valid"
+ " response. The return type must be a string,"
+ " dict, list, tuple with headers or status,"
+ " Response instance, or WSGI callable, but it"
+ f" was a {type(rv).__name__}."
+ ).with_traceback(sys.exc_info()[2]) from None
+ else:
+ raise TypeError(
+ "The view function did not return a valid"
+ " response. The return type must be a string,"
+ " dict, list, tuple with headers or status,"
+ " Response instance, or WSGI callable, but it was a"
+ f" {type(rv).__name__}."
+ )
+
+ rv = t.cast(Response, rv)
+ # prefer the status if it was provided
+ if status is not None:
+ if isinstance(status, (str, bytes, bytearray)):
+ rv.status = status
+ else:
+ rv.status_code = status
+
+ # extend existing headers with provided headers
+ if headers:
+ rv.headers.update(headers) # type: ignore[arg-type]
+
+ return rv
+
+ def preprocess_request(self) -> ft.ResponseReturnValue | None:
"""Called before the request is dispatched. Calls
:attr:`url_value_preprocessors` registered with the app and the
current blueprint (if any). Then calls :attr:`before_request_funcs`
@@ -736,9 +1240,24 @@ class Flask(App):
value is handled as if it was the return value from the view, and
further request handling is stopped.
"""
- pass
+ names = (None, *reversed(request.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)
+
+ for name in names:
+ if name in self.before_request_funcs:
+ for before_func in self.before_request_funcs[name]:
+ rv = self.ensure_sync(before_func)()
+
+ if rv is not None:
+ return rv # type: ignore[no-any-return]
- def process_response(self, response: Response) ->Response:
+ return None
+
+ def process_response(self, response: Response) -> Response:
"""Can be overridden in order to modify the response object
before it's sent to the WSGI server. By default this will
call all the :meth:`after_request` decorated functions.
@@ -751,10 +1270,25 @@ class Flask(App):
:return: a new response object or the same, has to be an
instance of :attr:`response_class`.
"""
- pass
+ ctx = request_ctx._get_current_object() # type: ignore[attr-defined]
+
+ for func in ctx._after_request_functions:
+ response = self.ensure_sync(func)(response)
+
+ for name in chain(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)
+
+ if not self.session_interface.is_null_session(ctx.session):
+ self.session_interface.save_session(self, ctx.session, response)
- def do_teardown_request(self, exc: (BaseException | None)=_sentinel
- ) ->None:
+ 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.
@@ -775,10 +1309,20 @@ class Flask(App):
.. versionchanged:: 0.9
Added the ``exc`` argument.
"""
- pass
+ if exc is _sentinel:
+ exc = sys.exc_info()[1]
+
+ for name in chain(request.blueprints, (None,)):
+ if name in self.teardown_request_funcs:
+ for func in reversed(self.teardown_request_funcs[name]):
+ self.ensure_sync(func)(exc)
- def do_teardown_appcontext(self, exc: (BaseException | None)=_sentinel
- ) ->None:
+ 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.
When handling a request, the application context is popped
@@ -793,9 +1337,15 @@ class Flask(App):
.. versionadded:: 0.9
"""
- pass
+ if exc is _sentinel:
+ exc = sys.exc_info()[1]
+
+ for func in reversed(self.teardown_appcontext_funcs):
+ self.ensure_sync(func)(exc)
- def app_context(self) ->AppContext:
+ 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.
@@ -814,9 +1364,9 @@ class Flask(App):
.. versionadded:: 0.9
"""
- pass
+ return AppContext(self)
- def request_context(self, environ: WSGIEnvironment) ->RequestContext:
+ 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.
@@ -830,10 +1380,9 @@ class Flask(App):
:param environ: a WSGI environment
"""
- pass
+ return RequestContext(self, environ)
- def test_request_context(self, *args: t.Any, **kwargs: t.Any
- ) ->RequestContext:
+ 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
@@ -880,10 +1429,18 @@ class Flask(App):
:param kwargs: other keyword arguments passed to
:class:`~werkzeug.test.EnvironBuilder`.
"""
- pass
+ from .testing import EnvironBuilder
+
+ builder = EnvironBuilder(self, *args, **kwargs)
- def wsgi_app(self, environ: WSGIEnvironment, start_response: StartResponse
- ) ->cabc.Iterable[bytes]:
+ try:
+ return self.request_context(builder.get_environ())
+ finally:
+ builder.close()
+
+ def wsgi_app(
+ self, environ: WSGIEnvironment, start_response: StartResponse
+ ) -> cabc.Iterable[bytes]:
"""The actual WSGI application. This is not implemented in
:meth:`__call__` so that middlewares can be applied without
losing a reference to the app object. Instead of doing this::
@@ -908,10 +1465,32 @@ class Flask(App):
a list of headers, and an optional exception context to
start the response.
"""
- pass
-
- def __call__(self, environ: WSGIEnvironment, start_response: StartResponse
- ) ->cabc.Iterable[bytes]:
+ ctx = self.request_context(environ)
+ error: BaseException | None = None
+ try:
+ try:
+ ctx.push()
+ response = self.full_dispatch_request()
+ except Exception as e:
+ error = e
+ response = self.handle_exception(e)
+ except: # noqa: B001
+ error = sys.exc_info()[1]
+ raise
+ return response(environ, start_response)
+ 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
+
+ ctx.pop(error)
+
+ def __call__(
+ self, environ: WSGIEnvironment, start_response: StartResponse
+ ) -> cabc.Iterable[bytes]:
"""The WSGI server calls the Flask application object as the
WSGI application. This calls :meth:`wsgi_app`, which can be
wrapped to apply middleware.
diff --git a/src/flask/blueprints.py b/src/flask/blueprints.py
index 446e7185..aa9eacf2 100644
--- a/src/flask/blueprints.py
+++ b/src/flask/blueprints.py
@@ -1,32 +1,58 @@
from __future__ import annotations
+
import os
import typing as t
from datetime import timedelta
+
from .cli import AppGroup
from .globals import current_app
from .helpers import send_from_directory
from .sansio.blueprints import Blueprint as SansioBlueprint
-from .sansio.blueprints import BlueprintSetupState as BlueprintSetupState
+from .sansio.blueprints import BlueprintSetupState as BlueprintSetupState # noqa
from .sansio.scaffold import _sentinel
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from .wrappers import Response
class Blueprint(SansioBlueprint):
-
- def __init__(self, name: str, import_name: str, static_folder: (str |
- os.PathLike[str] | None)=None, static_url_path: (str | None)=None,
- template_folder: (str | os.PathLike[str] | None)=None, url_prefix:
- (str | None)=None, subdomain: (str | None)=None, url_defaults: (
- dict[str, t.Any] | None)=None, root_path: (str | None)=None,
- cli_group: (str | None)=_sentinel) ->None:
- super().__init__(name, import_name, static_folder, static_url_path,
- template_folder, url_prefix, subdomain, url_defaults, root_path,
- cli_group)
+ def __init__(
+ self,
+ name: str,
+ import_name: str,
+ static_folder: str | os.PathLike[str] | None = None,
+ static_url_path: str | None = None,
+ template_folder: str | os.PathLike[str] | None = None,
+ url_prefix: str | None = None,
+ subdomain: str | None = None,
+ url_defaults: dict[str, t.Any] | None = None,
+ root_path: str | None = None,
+ cli_group: str | None = _sentinel, # type: ignore
+ ) -> None:
+ super().__init__(
+ name,
+ import_name,
+ static_folder,
+ static_url_path,
+ template_folder,
+ url_prefix,
+ subdomain,
+ url_defaults,
+ root_path,
+ cli_group,
+ )
+
+ #: The Click command group for registering CLI commands for this
+ #: object. The commands are available from the ``flask`` command
+ #: once the application has been discovered and blueprints have
+ #: been registered.
self.cli = AppGroup()
+
+ # Set the name of the Click group in case someone wants to add
+ # the app's commands to another CLI tool.
self.cli.name = self.name
- def get_send_file_max_age(self, filename: (str | None)) ->(int | None):
+ def get_send_file_max_age(self, filename: str | None) -> int | None:
"""Used by :func:`send_file` to determine the ``max_age`` cache
value for a given file path if it wasn't passed.
@@ -43,9 +69,17 @@ class Blueprint(SansioBlueprint):
.. versionadded:: 0.9
"""
- pass
+ value = current_app.config["SEND_FILE_MAX_AGE_DEFAULT"]
- def send_static_file(self, filename: str) ->Response:
+ if value is None:
+ return None
+
+ if isinstance(value, timedelta):
+ return int(value.total_seconds())
+
+ return value # type: ignore[no-any-return]
+
+ def send_static_file(self, filename: str) -> Response:
"""The view function used to serve files from
:attr:`static_folder`. A route is automatically registered for
this view at :attr:`static_url_path` if :attr:`static_folder` is
@@ -57,9 +91,17 @@ class Blueprint(SansioBlueprint):
.. versionadded:: 0.5
"""
- pass
+ if not self.has_static_folder:
+ raise RuntimeError("'static_folder' must be set to serve static_files.")
- def open_resource(self, resource: str, mode: str='rb') ->t.IO[t.AnyStr]:
+ # send_file only knows to call get_send_file_max_age on the app,
+ # call it here so it works for blueprints too.
+ max_age = self.get_send_file_max_age(filename)
+ return send_from_directory(
+ t.cast(str, self.static_folder), filename, max_age=max_age
+ )
+
+ def open_resource(self, resource: str, mode: str = "rb") -> t.IO[t.AnyStr]:
"""Open a resource file relative to :attr:`root_path` for
reading.
@@ -81,4 +123,7 @@ class Blueprint(SansioBlueprint):
class.
"""
- pass
+ if mode not in {"r", "rt", "rb"}:
+ raise ValueError("Resources can only be opened for reading.")
+
+ return open(os.path.join(self.root_path, resource), mode)
diff --git a/src/flask/cli.py b/src/flask/cli.py
index f98c3984..ecb292a0 100644
--- a/src/flask/cli.py
+++ b/src/flask/cli.py
@@ -1,4 +1,5 @@
from __future__ import annotations
+
import ast
import collections.abc as cabc
import importlib.metadata
@@ -12,19 +13,24 @@ import typing as t
from functools import update_wrapper
from operator import itemgetter
from types import ModuleType
+
import click
from click.core import ParameterSource
from werkzeug import run_simple
from werkzeug.serving import is_running_from_reloader
from werkzeug.utils import import_string
+
from .globals import current_app
from .helpers import get_debug_flag
from .helpers import get_load_dotenv
+
if t.TYPE_CHECKING:
import ssl
+
from _typeshed.wsgi import StartResponse
from _typeshed.wsgi import WSGIApplication
from _typeshed.wsgi import WSGIEnvironment
+
from .app import Flask
@@ -32,14 +38,60 @@ class NoAppException(click.UsageError):
"""Raised if an application cannot be found or loaded."""
-def find_best_app(module: ModuleType) ->Flask:
+def find_best_app(module: ModuleType) -> Flask:
"""Given a module instance this tries to find the best possible
application in the module or raises an exception.
"""
- pass
+ from . import Flask
+
+ # Search for the most common names first.
+ for attr_name in ("app", "application"):
+ app = getattr(module, attr_name, None)
+
+ if isinstance(app, Flask):
+ return app
+
+ # Otherwise find the only object that is a Flask instance.
+ matches = [v for v in module.__dict__.values() if isinstance(v, Flask)]
+
+ if len(matches) == 1:
+ return matches[0]
+ elif len(matches) > 1:
+ raise NoAppException(
+ "Detected multiple Flask applications in module"
+ f" '{module.__name__}'. Use '{module.__name__}:name'"
+ " to specify the correct one."
+ )
+
+ # Search for app factory functions.
+ for attr_name in ("create_app", "make_app"):
+ app_factory = getattr(module, attr_name, None)
+
+ if inspect.isfunction(app_factory):
+ try:
+ app = app_factory()
+
+ if isinstance(app, Flask):
+ return app
+ except TypeError as e:
+ if not _called_with_wrong_args(app_factory):
+ raise
+
+ raise NoAppException(
+ f"Detected factory '{attr_name}' in module '{module.__name__}',"
+ " but could not call it without arguments. Use"
+ f" '{module.__name__}:{attr_name}(args)'"
+ " to specify arguments."
+ ) from e
+
+ raise NoAppException(
+ "Failed to find Flask application or factory in module"
+ f" '{module.__name__}'. Use '{module.__name__}:name'"
+ " to specify one."
+ )
-def _called_with_wrong_args(f: t.Callable[..., Flask]) ->bool:
+def _called_with_wrong_args(f: t.Callable[..., Flask]) -> bool:
"""Check whether calling a function raised a ``TypeError`` because
the call failed or because something in the factory raised the
error.
@@ -47,25 +99,195 @@ def _called_with_wrong_args(f: t.Callable[..., Flask]) ->bool:
:param f: The function that was called.
:return: ``True`` if the call failed.
"""
- pass
+ tb = sys.exc_info()[2]
+ try:
+ while tb is not None:
+ if tb.tb_frame.f_code is f.__code__:
+ # In the function, it was called successfully.
+ return False
-def find_app_by_string(module: ModuleType, app_name: str) ->Flask:
+ tb = tb.tb_next
+
+ # Didn't reach the function.
+ return True
+ finally:
+ # Delete tb to break a circular reference.
+ # https://docs.python.org/2/library/sys.html#sys.exc_info
+ del tb
+
+
+def find_app_by_string(module: ModuleType, app_name: str) -> Flask:
"""Check if the given string is a variable name or a function. Call
a function to get the app instance, or return the variable directly.
"""
- pass
+ from . import Flask
+
+ # Parse app_name as a single expression to determine if it's a valid
+ # attribute name or function call.
+ try:
+ expr = ast.parse(app_name.strip(), mode="eval").body
+ except SyntaxError:
+ raise NoAppException(
+ f"Failed to parse {app_name!r} as an attribute name or function call."
+ ) from None
+
+ if isinstance(expr, ast.Name):
+ name = expr.id
+ args = []
+ kwargs = {}
+ elif isinstance(expr, ast.Call):
+ # Ensure the function name is an attribute name only.
+ if not isinstance(expr.func, ast.Name):
+ raise NoAppException(
+ f"Function reference must be a simple name: {app_name!r}."
+ )
+
+ name = expr.func.id
+
+ # Parse the positional and keyword arguments as literals.
+ try:
+ args = [ast.literal_eval(arg) for arg in expr.args]
+ kwargs = {
+ kw.arg: ast.literal_eval(kw.value)
+ for kw in expr.keywords
+ if kw.arg is not None
+ }
+ except ValueError:
+ # literal_eval gives cryptic error messages, show a generic
+ # message with the full expression instead.
+ raise NoAppException(
+ f"Failed to parse arguments as literal values: {app_name!r}."
+ ) from None
+ else:
+ raise NoAppException(
+ f"Failed to parse {app_name!r} as an attribute name or function call."
+ )
+
+ try:
+ attr = getattr(module, name)
+ except AttributeError as e:
+ raise NoAppException(
+ f"Failed to find attribute {name!r} in {module.__name__!r}."
+ ) from e
+
+ # If the attribute is a function, call it with any args and kwargs
+ # to get the real application.
+ if inspect.isfunction(attr):
+ try:
+ app = attr(*args, **kwargs)
+ except TypeError as e:
+ if not _called_with_wrong_args(attr):
+ raise
+
+ raise NoAppException(
+ f"The factory {app_name!r} in module"
+ f" {module.__name__!r} could not be called with the"
+ " specified arguments."
+ ) from e
+ else:
+ app = attr
+
+ if isinstance(app, Flask):
+ return app
+
+ raise NoAppException(
+ "A valid Flask application was not obtained from"
+ f" '{module.__name__}:{app_name}'."
+ )
-def prepare_import(path: str) ->str:
+def prepare_import(path: str) -> str:
"""Given a filename this will try to calculate the python path, add it
to the search path and return the actual module name that is expected.
"""
- pass
+ path = os.path.realpath(path)
+
+ fname, ext = os.path.splitext(path)
+ if ext == ".py":
+ path = fname
+
+ if os.path.basename(path) == "__init__":
+ path = os.path.dirname(path)
+
+ module_name = []
+
+ # move up until outside package structure (no __init__.py)
+ while True:
+ path, name = os.path.split(path)
+ module_name.append(name)
+
+ if not os.path.exists(os.path.join(path, "__init__.py")):
+ break
+
+ if sys.path[0] != path:
+ sys.path.insert(0, path)
+
+ return ".".join(module_name[::-1])
+
+
+@t.overload
+def locate_app(
+ module_name: str, app_name: str | None, raise_if_not_found: t.Literal[True] = True
+) -> Flask: ...
+
+
+@t.overload
+def locate_app(
+ module_name: str, app_name: str | None, raise_if_not_found: t.Literal[False] = ...
+) -> Flask | None: ...
+
+
+def locate_app(
+ module_name: str, app_name: str | None, raise_if_not_found: bool = True
+) -> Flask | None:
+ try:
+ __import__(module_name)
+ except ImportError:
+ # Reraise the ImportError if it occurred within the imported module.
+ # Determine this by checking whether the trace has a depth > 1.
+ if sys.exc_info()[2].tb_next: # type: ignore[union-attr]
+ raise NoAppException(
+ f"While importing {module_name!r}, an ImportError was"
+ f" raised:\n\n{traceback.format_exc()}"
+ ) from None
+ elif raise_if_not_found:
+ raise NoAppException(f"Could not import {module_name!r}.") from None
+ else:
+ return None
+
+ module = sys.modules[module_name]
+ if app_name is None:
+ return find_best_app(module)
+ else:
+ return find_app_by_string(module, app_name)
-version_option = click.Option(['--version'], help='Show the Flask version.',
- expose_value=False, callback=get_version, is_flag=True, is_eager=True)
+
+def get_version(ctx: click.Context, param: click.Parameter, value: t.Any) -> None:
+ if not value or ctx.resilient_parsing:
+ return
+
+ flask_version = importlib.metadata.version("flask")
+ werkzeug_version = importlib.metadata.version("werkzeug")
+
+ click.echo(
+ f"Python {platform.python_version()}\n"
+ f"Flask {flask_version}\n"
+ f"Werkzeug {werkzeug_version}",
+ color=ctx.color,
+ )
+ ctx.exit()
+
+
+version_option = click.Option(
+ ["--version"],
+ help="Show the Flask version.",
+ expose_value=False,
+ callback=get_version,
+ is_flag=True,
+ is_eager=True,
+)
class ScriptInfo:
@@ -77,27 +299,71 @@ class ScriptInfo:
onwards as click object.
"""
- def __init__(self, app_import_path: (str | None)=None, create_app: (t.
- Callable[..., Flask] | None)=None, set_debug_flag: bool=True) ->None:
+ def __init__(
+ self,
+ app_import_path: str | None = None,
+ create_app: t.Callable[..., Flask] | None = None,
+ set_debug_flag: bool = True,
+ ) -> None:
+ #: Optionally the import path for the Flask application.
self.app_import_path = app_import_path
+ #: Optionally a function that is passed the script info to create
+ #: the instance of the application.
self.create_app = create_app
+ #: A dictionary with arbitrary data that can be associated with
+ #: this script info.
self.data: dict[t.Any, t.Any] = {}
self.set_debug_flag = set_debug_flag
self._loaded_app: Flask | None = None
- def load_app(self) ->Flask:
+ def load_app(self) -> Flask:
"""Loads the Flask app (if not yet loaded) and returns it. Calling
this multiple times will just result in the already loaded app to
be returned.
"""
- pass
+ if self._loaded_app is not None:
+ return self._loaded_app
+
+ if self.create_app is not None:
+ app: Flask | None = self.create_app()
+ else:
+ if self.app_import_path:
+ path, name = (
+ re.split(r":(?![\\/])", self.app_import_path, maxsplit=1) + [None]
+ )[:2]
+ import_name = prepare_import(path)
+ app = locate_app(import_name, name)
+ else:
+ for path in ("wsgi.py", "app.py"):
+ import_name = prepare_import(path)
+ app = locate_app(import_name, None, raise_if_not_found=False)
+
+ if app is not None:
+ break
+
+ if app is None:
+ raise NoAppException(
+ "Could not locate a Flask application. Use the"
+ " 'flask --app' option, 'FLASK_APP' environment"
+ " variable, or a 'wsgi.py' or 'app.py' file in the"
+ " current directory."
+ )
+
+ if self.set_debug_flag:
+ # Update the app's debug flag through the descriptor so that
+ # other values repopulate as well.
+ app.debug = get_debug_flag()
+
+ self._loaded_app = app
+ return app
pass_script_info = click.make_pass_decorator(ScriptInfo, ensure=True)
-F = t.TypeVar('F', bound=t.Callable[..., t.Any])
+F = t.TypeVar("F", bound=t.Callable[..., t.Any])
-def with_appcontext(f: F) ->F:
+
+def with_appcontext(f: F) -> F:
"""Wraps a callback so that it's guaranteed to be executed with the
script's application context.
@@ -110,7 +376,16 @@ def with_appcontext(f: F) ->F:
decorated callback. The app context is always available to
``app.cli`` command and parameter callbacks.
"""
- pass
+
+ @click.pass_context
+ def decorator(ctx: click.Context, /, *args: t.Any, **kwargs: t.Any) -> t.Any:
+ if not current_app:
+ app = ctx.ensure_object(ScriptInfo).load_app()
+ ctx.with_resource(app.app_context())
+
+ return ctx.invoke(f, *args, **kwargs)
+
+ return update_wrapper(decorator, f) # type: ignore[return-value]
class AppGroup(click.Group):
@@ -121,32 +396,119 @@ class AppGroup(click.Group):
Not to be confused with :class:`FlaskGroup`.
"""
- def command(self, *args: t.Any, **kwargs: t.Any) ->t.Callable[[t.
- Callable[..., t.Any]], click.Command]:
+ def command( # type: ignore[override]
+ self, *args: t.Any, **kwargs: t.Any
+ ) -> t.Callable[[t.Callable[..., t.Any]], click.Command]:
"""This works exactly like the method of the same name on a regular
:class:`click.Group` but it wraps callbacks in :func:`with_appcontext`
unless it's disabled by passing ``with_appcontext=False``.
"""
- pass
+ wrap_for_ctx = kwargs.pop("with_appcontext", True)
+
+ def decorator(f: t.Callable[..., t.Any]) -> click.Command:
+ if wrap_for_ctx:
+ f = with_appcontext(f)
+ return super(AppGroup, self).command(*args, **kwargs)(f) # type: ignore[no-any-return]
- def group(self, *args: t.Any, **kwargs: t.Any) ->t.Callable[[t.Callable
- [..., t.Any]], click.Group]:
+ return decorator
+
+ def group( # type: ignore[override]
+ self, *args: t.Any, **kwargs: t.Any
+ ) -> t.Callable[[t.Callable[..., t.Any]], click.Group]:
"""This works exactly like the method of the same name on a regular
:class:`click.Group` but it defaults the group class to
:class:`AppGroup`.
"""
- pass
-
-
-_app_option = click.Option(['-A', '--app'], metavar='IMPORT', help=
- "The Flask application or factory function to load, in the form 'module:name'. Module can be a dotted import or file path. Name is not required if it is 'app', 'application', 'create_app', or 'make_app', and can be 'name(args)' to pass arguments."
- , is_eager=True, expose_value=False, callback=_set_app)
-_debug_option = click.Option(['--debug/--no-debug'], help='Set debug mode.',
- expose_value=False, callback=_set_debug)
-_env_file_option = click.Option(['-e', '--env-file'], type=click.Path(
- exists=True, dir_okay=False), help=
- 'Load environment variables from this file. python-dotenv must be installed.'
- , is_eager=True, expose_value=False, callback=_env_file_callback)
+ kwargs.setdefault("cls", AppGroup)
+ return super().group(*args, **kwargs) # type: ignore[no-any-return]
+
+
+def _set_app(ctx: click.Context, param: click.Option, value: str | None) -> str | None:
+ if value is None:
+ return None
+
+ info = ctx.ensure_object(ScriptInfo)
+ info.app_import_path = value
+ return value
+
+
+# This option is eager so the app will be available if --help is given.
+# --help is also eager, so --app must be before it in the param list.
+# no_args_is_help bypasses eager processing, so this option must be
+# processed manually in that case to ensure FLASK_APP gets picked up.
+_app_option = click.Option(
+ ["-A", "--app"],
+ metavar="IMPORT",
+ help=(
+ "The Flask application or factory function to load, in the form 'module:name'."
+ " Module can be a dotted import or file path. Name is not required if it is"
+ " 'app', 'application', 'create_app', or 'make_app', and can be 'name(args)' to"
+ " pass arguments."
+ ),
+ is_eager=True,
+ expose_value=False,
+ callback=_set_app,
+)
+
+
+def _set_debug(ctx: click.Context, param: click.Option, value: bool) -> bool | None:
+ # If the flag isn't provided, it will default to False. Don't use
+ # that, let debug be set by env in that case.
+ source = ctx.get_parameter_source(param.name) # type: ignore[arg-type]
+
+ if source is not None and source in (
+ ParameterSource.DEFAULT,
+ ParameterSource.DEFAULT_MAP,
+ ):
+ return None
+
+ # Set with env var instead of ScriptInfo.load so that it can be
+ # accessed early during a factory function.
+ os.environ["FLASK_DEBUG"] = "1" if value else "0"
+ return value
+
+
+_debug_option = click.Option(
+ ["--debug/--no-debug"],
+ help="Set debug mode.",
+ expose_value=False,
+ callback=_set_debug,
+)
+
+
+def _env_file_callback(
+ ctx: click.Context, param: click.Option, value: str | None
+) -> str | None:
+ if value is None:
+ return None
+
+ import importlib
+
+ try:
+ importlib.import_module("dotenv")
+ except ImportError:
+ raise click.BadParameter(
+ "python-dotenv must be installed to load an env file.",
+ ctx=ctx,
+ param=param,
+ ) from None
+
+ # Don't check FLASK_SKIP_DOTENV, that only disables automatically
+ # loading .env and .flaskenv files.
+ load_dotenv(value)
+ return value
+
+
+# This option is eager so env vars are loaded as early as possible to be
+# used by other options.
+_env_file_option = click.Option(
+ ["-e", "--env-file"],
+ type=click.Path(exists=True, dir_okay=False),
+ help="Load environment variables from this file. python-dotenv must be installed.",
+ is_eager=True,
+ expose_value=False,
+ callback=_env_file_callback,
+)
class FlaskGroup(AppGroup):
@@ -178,36 +540,151 @@ class FlaskGroup(AppGroup):
from :file:`.env` and :file:`.flaskenv` files.
"""
- def __init__(self, add_default_commands: bool=True, create_app: (t.
- Callable[..., Flask] | None)=None, add_version_option: bool=True,
- load_dotenv: bool=True, set_debug_flag: bool=True, **extra: t.Any
- ) ->None:
- params = list(extra.pop('params', None) or ())
+ def __init__(
+ self,
+ add_default_commands: bool = True,
+ create_app: t.Callable[..., Flask] | None = None,
+ add_version_option: bool = True,
+ load_dotenv: bool = True,
+ set_debug_flag: bool = True,
+ **extra: t.Any,
+ ) -> None:
+ params = list(extra.pop("params", None) or ())
+ # Processing is done with option callbacks instead of a group
+ # callback. This allows users to make a custom group callback
+ # without losing the behavior. --env-file must come first so
+ # that it is eagerly evaluated before --app.
params.extend((_env_file_option, _app_option, _debug_option))
+
if add_version_option:
params.append(version_option)
- if 'context_settings' not in extra:
- extra['context_settings'] = {}
- extra['context_settings'].setdefault('auto_envvar_prefix', 'FLASK')
+
+ if "context_settings" not in extra:
+ extra["context_settings"] = {}
+
+ extra["context_settings"].setdefault("auto_envvar_prefix", "FLASK")
+
super().__init__(params=params, **extra)
+
self.create_app = create_app
self.load_dotenv = load_dotenv
self.set_debug_flag = set_debug_flag
+
if add_default_commands:
self.add_command(run_command)
self.add_command(shell_command)
self.add_command(routes_command)
- self._loaded_plugin_commands = False
+ self._loaded_plugin_commands = False
-def _path_is_ancestor(path: str, other: str) ->bool:
+ def _load_plugin_commands(self) -> None:
+ if self._loaded_plugin_commands:
+ return
+
+ if sys.version_info >= (3, 10):
+ from importlib import metadata
+ else:
+ # Use a backport on Python < 3.10. We technically have
+ # importlib.metadata on 3.8+, but the API changed in 3.10,
+ # so use the backport for consistency.
+ import importlib_metadata as metadata
+
+ for ep in metadata.entry_points(group="flask.commands"):
+ self.add_command(ep.load(), ep.name)
+
+ self._loaded_plugin_commands = True
+
+ def get_command(self, ctx: click.Context, name: str) -> click.Command | None:
+ self._load_plugin_commands()
+ # Look up built-in and plugin commands, which should be
+ # available even if the app fails to load.
+ rv = super().get_command(ctx, name)
+
+ if rv is not None:
+ return rv
+
+ info = ctx.ensure_object(ScriptInfo)
+
+ # Look up commands provided by the app, showing an error and
+ # continuing if the app couldn't be loaded.
+ try:
+ app = info.load_app()
+ except NoAppException as e:
+ click.secho(f"Error: {e.format_message()}\n", err=True, fg="red")
+ return None
+
+ # 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]
+ ctx.with_resource(app.app_context())
+
+ return app.cli.get_command(ctx, name)
+
+ def list_commands(self, ctx: click.Context) -> list[str]:
+ self._load_plugin_commands()
+ # Start with the built-in and plugin commands.
+ rv = set(super().list_commands(ctx))
+ info = ctx.ensure_object(ScriptInfo)
+
+ # Add commands provided by the app, showing an error and
+ # continuing if the app couldn't be loaded.
+ try:
+ rv.update(info.load_app().cli.list_commands(ctx))
+ except NoAppException as e:
+ # When an app couldn't be loaded, show the error message
+ # without the traceback.
+ click.secho(f"Error: {e.format_message()}\n", err=True, fg="red")
+ except Exception:
+ # When any other errors occurred during loading, show the
+ # full traceback.
+ click.secho(f"{traceback.format_exc()}\n", err=True, fg="red")
+
+ return sorted(rv)
+
+ def make_context(
+ self,
+ info_name: str | None,
+ args: list[str],
+ parent: click.Context | None = None,
+ **extra: t.Any,
+ ) -> click.Context:
+ # Set a flag to tell app.run to become a no-op. If app.run was
+ # not in a __name__ == __main__ guard, it would start the server
+ # when importing, blocking whatever command is being called.
+ os.environ["FLASK_RUN_FROM_CLI"] = "true"
+
+ # Attempt to load .env and .flask env files. The --env-file
+ # option can cause another file to be loaded.
+ if get_load_dotenv(self.load_dotenv):
+ load_dotenv()
+
+ if "obj" not in extra and "obj" not in self.context_settings:
+ extra["obj"] = ScriptInfo(
+ create_app=self.create_app, set_debug_flag=self.set_debug_flag
+ )
+
+ return super().make_context(info_name, args, parent=parent, **extra)
+
+ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
+ if not args and self.no_args_is_help:
+ # Attempt to load --env-file and --app early in case they
+ # were given as env vars. Otherwise no_args_is_help will not
+ # see commands from app.cli.
+ _env_file_option.handle_parse_result(ctx, {}, [])
+ _app_option.handle_parse_result(ctx, {}, [])
+
+ return super().parse_args(ctx, args)
+
+
+def _path_is_ancestor(path: str, other: str) -> bool:
"""Take ``other`` and remove the length of ``path`` from it. Then join it
to ``path``. If it is the original value, ``path`` is an ancestor of
``other``."""
- pass
+ return os.path.join(path, other[len(path) :].lstrip(os.sep)) == other
-def load_dotenv(path: (str | os.PathLike[str] | None)=None) ->bool:
+def load_dotenv(path: str | os.PathLike[str] | None = None) -> bool:
"""Load "dotenv" files in order of precedence to set environment variables.
If an env var is already set it is not overwritten, so earlier files in the
@@ -233,14 +710,53 @@ def load_dotenv(path: (str | os.PathLike[str] | None)=None) ->bool:
.. versionadded:: 1.0
"""
- pass
+ try:
+ import dotenv
+ except ImportError:
+ if path or os.path.isfile(".env") or os.path.isfile(".flaskenv"):
+ click.secho(
+ " * Tip: There are .env or .flaskenv files present."
+ ' Do "pip install python-dotenv" to use them.',
+ fg="yellow",
+ err=True,
+ )
+ return False
-def show_server_banner(debug: bool, app_import_path: (str | None)) ->None:
+ # Always return after attempting to load a given path, don't load
+ # the default files.
+ if path is not None:
+ if os.path.isfile(path):
+ return dotenv.load_dotenv(path, encoding="utf-8")
+
+ return False
+
+ loaded = False
+
+ for name in (".env", ".flaskenv"):
+ path = dotenv.find_dotenv(name, usecwd=True)
+
+ if not path:
+ continue
+
+ dotenv.load_dotenv(path, encoding="utf-8")
+ loaded = True
+
+ return loaded # True if at least one file was located and loaded.
+
+
+def show_server_banner(debug: bool, app_import_path: str | None) -> None:
"""Show extra startup messages the first time the server is run,
ignoring the reloader.
"""
- pass
+ if is_running_from_reloader():
+ return
+
+ if app_import_path is not None:
+ click.echo(f" * Serving Flask app '{app_import_path}'")
+
+ if debug is not None:
+ click.echo(f" * Debug mode: {'on' if debug else 'off'}")
class CertParamType(click.ParamType):
@@ -248,19 +764,86 @@ class CertParamType(click.ParamType):
existing file, the string ``'adhoc'``, or an import for a
:class:`~ssl.SSLContext` object.
"""
- name = 'path'
- def __init__(self) ->None:
- self.path_type = click.Path(exists=True, dir_okay=False,
- resolve_path=True)
+ name = "path"
+
+ def __init__(self) -> None:
+ self.path_type = click.Path(exists=True, dir_okay=False, resolve_path=True)
+
+ def convert(
+ self, value: t.Any, param: click.Parameter | None, ctx: click.Context | None
+ ) -> t.Any:
+ try:
+ import ssl
+ except ImportError:
+ raise click.BadParameter(
+ 'Using "--cert" requires Python to be compiled with SSL support.',
+ ctx,
+ param,
+ ) from None
+
+ try:
+ return self.path_type(value, param, ctx)
+ except click.BadParameter:
+ value = click.STRING(value, param, ctx).lower()
+ if value == "adhoc":
+ try:
+ import cryptography # noqa: F401
+ except ImportError:
+ raise click.BadParameter(
+ "Using ad-hoc certificates requires the cryptography library.",
+ ctx,
+ param,
+ ) from None
-def _validate_key(ctx: click.Context, param: click.Parameter, value: t.Any
- ) ->t.Any:
+ return value
+
+ obj = import_string(value, silent=True)
+
+ if isinstance(obj, ssl.SSLContext):
+ return obj
+
+ raise
+
+
+def _validate_key(ctx: click.Context, param: click.Parameter, value: t.Any) -> t.Any:
"""The ``--key`` option must be specified when ``--cert`` is a file.
Modifies the ``cert`` param to be a ``(cert, key)`` pair if needed.
"""
- pass
+ cert = ctx.params.get("cert")
+ is_adhoc = cert == "adhoc"
+
+ try:
+ import ssl
+ except ImportError:
+ is_context = False
+ else:
+ is_context = isinstance(cert, ssl.SSLContext)
+
+ if value is not None:
+ if is_adhoc:
+ raise click.BadParameter(
+ 'When "--cert" is "adhoc", "--key" is not used.', ctx, param
+ )
+
+ if is_context:
+ raise click.BadParameter(
+ 'When "--cert" is an SSLContext object, "--key" is not used.',
+ ctx,
+ param,
+ )
+
+ if not cert:
+ raise click.BadParameter('"--cert" must also be specified.', ctx, param)
+
+ ctx.params["cert"] = cert, value
+
+ else:
+ if cert and not (is_adhoc or is_context):
+ raise click.BadParameter('Required when using "--cert".', ctx, param)
+
+ return value
class SeparatedPathType(click.Path):
@@ -269,37 +852,79 @@ class SeparatedPathType(click.Path):
validated as a :class:`click.Path` type.
"""
-
-@click.command('run', short_help='Run a development server.')
-@click.option('--host', '-h', default='127.0.0.1', help=
- 'The interface to bind to.')
-@click.option('--port', '-p', default=5000, help='The port to bind to.')
-@click.option('--cert', type=CertParamType(), help=
- 'Specify a certificate file to use HTTPS.', is_eager=True)
-@click.option('--key', type=click.Path(exists=True, dir_okay=False,
- resolve_path=True), callback=_validate_key, expose_value=False, help=
- 'The key file to use when specifying a certificate.')
-@click.option('--reload/--no-reload', default=None, help=
- 'Enable or disable the reloader. By default the reloader is active if debug is enabled.'
- )
-@click.option('--debugger/--no-debugger', default=None, help=
- 'Enable or disable the debugger. By default the debugger is active if debug is enabled.'
- )
-@click.option('--with-threads/--without-threads', default=True, help=
- 'Enable or disable multithreading.')
-@click.option('--extra-files', default=None, type=SeparatedPathType(), help
- =
- f'Extra files that trigger a reload on change. Multiple paths are separated by {os.path.pathsep!r}.'
- )
-@click.option('--exclude-patterns', default=None, type=SeparatedPathType(),
- help=
- f'Files matching these fnmatch patterns will not trigger a reload on change. Multiple patterns are separated by {os.path.pathsep!r}.'
- )
+ def convert(
+ self, value: t.Any, param: click.Parameter | None, ctx: click.Context | None
+ ) -> t.Any:
+ items = self.split_envvar_value(value)
+ # can't call no-arg super() inside list comprehension until Python 3.12
+ super_convert = super().convert
+ return [super_convert(item, param, ctx) for item in items]
+
+
+@click.command("run", short_help="Run a development server.")
+@click.option("--host", "-h", default="127.0.0.1", help="The interface to bind to.")
+@click.option("--port", "-p", default=5000, help="The port to bind to.")
+@click.option(
+ "--cert",
+ type=CertParamType(),
+ help="Specify a certificate file to use HTTPS.",
+ is_eager=True,
+)
+@click.option(
+ "--key",
+ type=click.Path(exists=True, dir_okay=False, resolve_path=True),
+ callback=_validate_key,
+ expose_value=False,
+ help="The key file to use when specifying a certificate.",
+)
+@click.option(
+ "--reload/--no-reload",
+ default=None,
+ help="Enable or disable the reloader. By default the reloader "
+ "is active if debug is enabled.",
+)
+@click.option(
+ "--debugger/--no-debugger",
+ default=None,
+ help="Enable or disable the debugger. By default the debugger "
+ "is active if debug is enabled.",
+)
+@click.option(
+ "--with-threads/--without-threads",
+ default=True,
+ help="Enable or disable multithreading.",
+)
+@click.option(
+ "--extra-files",
+ default=None,
+ type=SeparatedPathType(),
+ help=(
+ "Extra files that trigger a reload on change. Multiple paths"
+ f" are separated by {os.path.pathsep!r}."
+ ),
+)
+@click.option(
+ "--exclude-patterns",
+ default=None,
+ type=SeparatedPathType(),
+ help=(
+ "Files matching these fnmatch patterns will not trigger a reload"
+ " on change. Multiple patterns are separated by"
+ f" {os.path.pathsep!r}."
+ ),
+)
@pass_script_info
-def run_command(info: ScriptInfo, host: str, port: int, reload: bool,
- debugger: bool, with_threads: bool, cert: (ssl.SSLContext | tuple[str,
- str | None] | t.Literal['adhoc'] | None), extra_files: (list[str] |
- None), exclude_patterns: (list[str] | None)) ->None:
+def run_command(
+ info: ScriptInfo,
+ host: str,
+ port: int,
+ reload: bool,
+ debugger: bool,
+ with_threads: bool,
+ cert: ssl.SSLContext | tuple[str, str | None] | t.Literal["adhoc"] | None,
+ extra_files: list[str] | None,
+ exclude_patterns: list[str] | None,
+) -> None:
"""Run a local development server.
This server is for development purposes only. It does not provide
@@ -308,15 +933,54 @@ def run_command(info: ScriptInfo, host: str, port: int, reload: bool,
The reloader and debugger are enabled by default with the '--debug'
option.
"""
- pass
+ try:
+ app: WSGIApplication = info.load_app()
+ except Exception as e:
+ if is_running_from_reloader():
+ # When reloading, print out the error immediately, but raise
+ # it later so the debugger or server can handle it.
+ traceback.print_exc()
+ err = e
+
+ def app(
+ environ: WSGIEnvironment, start_response: StartResponse
+ ) -> cabc.Iterable[bytes]:
+ raise err from None
+
+ else:
+ # When not reloading, raise the error immediately so the
+ # command fails.
+ raise e from None
+
+ debug = get_debug_flag()
+
+ if reload is None:
+ reload = debug
+
+ if debugger is None:
+ debugger = debug
+
+ show_server_banner(debug, info.app_import_path)
+
+ run_simple(
+ host,
+ port,
+ app,
+ use_reloader=reload,
+ use_debugger=debugger,
+ threaded=with_threads,
+ ssl_context=cert,
+ extra_files=extra_files,
+ exclude_patterns=exclude_patterns,
+ )
run_command.params.insert(0, _debug_option)
-@click.command('shell', short_help='Run a shell in the app context.')
+@click.command("shell", short_help="Run a shell in the app context.")
@with_appcontext
-def shell_command() ->None:
+def shell_command() -> None:
"""Run an interactive Python shell in the context of a given
Flask application. The application will populate the default
namespace of this shell according to its configuration.
@@ -324,29 +988,122 @@ def shell_command() ->None:
This is useful for executing small snippets of management code
without having to manually configure the application.
"""
- pass
+ import code
-
-@click.command('routes', short_help='Show the routes for the app.')
-@click.option('--sort', '-s', type=click.Choice(('endpoint', 'methods',
- 'domain', 'rule', 'match')), default='endpoint', help=
- "Method to sort routes by. 'match' is the order that Flask will match routes when dispatching a request."
+ banner = (
+ f"Python {sys.version} on {sys.platform}\n"
+ f"App: {current_app.import_name}\n"
+ f"Instance: {current_app.instance_path}"
)
-@click.option('--all-methods', is_flag=True, help=
- 'Show HEAD and OPTIONS methods.')
+ ctx: dict[str, t.Any] = {}
+
+ # Support the regular Python interpreter startup script if someone
+ # is using it.
+ startup = os.environ.get("PYTHONSTARTUP")
+ if startup and os.path.isfile(startup):
+ with open(startup) as f:
+ eval(compile(f.read(), startup, "exec"), ctx)
+
+ ctx.update(current_app.make_shell_context())
+
+ # Site, customize, or startup script can set a hook to call when
+ # entering interactive mode. The default one sets up readline with
+ # tab and history completion.
+ interactive_hook = getattr(sys, "__interactivehook__", None)
+
+ if interactive_hook is not None:
+ try:
+ import readline
+ from rlcompleter import Completer
+ except ImportError:
+ pass
+ else:
+ # rlcompleter uses __main__.__dict__ by default, which is
+ # flask.__main__. Use the shell context instead.
+ readline.set_completer(Completer(ctx).complete)
+
+ interactive_hook()
+
+ code.interact(banner=banner, local=ctx)
+
+
+@click.command("routes", short_help="Show the routes for the app.")
+@click.option(
+ "--sort",
+ "-s",
+ type=click.Choice(("endpoint", "methods", "domain", "rule", "match")),
+ default="endpoint",
+ help=(
+ "Method to sort routes by. 'match' is the order that Flask will match routes"
+ " when dispatching a request."
+ ),
+)
+@click.option("--all-methods", is_flag=True, help="Show HEAD and OPTIONS methods.")
@with_appcontext
-def routes_command(sort: str, all_methods: bool) ->None:
+def routes_command(sort: str, all_methods: bool) -> None:
"""Show all registered routes with endpoints and methods."""
- pass
+ rules = list(current_app.url_map.iter_rules())
+
+ if not rules:
+ click.echo("No routes were registered.")
+ return
+
+ ignored_methods = set() if all_methods else {"HEAD", "OPTIONS"}
+ host_matching = current_app.url_map.host_matching
+ has_domain = any(rule.host if host_matching else rule.subdomain for rule in rules)
+ rows = []
+
+ for rule in rules:
+ row = [
+ rule.endpoint,
+ ", ".join(sorted((rule.methods or set()) - ignored_methods)),
+ ]
+
+ if has_domain:
+ row.append((rule.host if host_matching else rule.subdomain) or "")
+
+ row.append(rule.rule)
+ rows.append(row)
+ headers = ["Endpoint", "Methods"]
+ sorts = ["endpoint", "methods"]
-cli = FlaskGroup(name='flask', help=
- """A general utility script for Flask applications.
+ if has_domain:
+ headers.append("Host" if host_matching else "Subdomain")
+ sorts.append("domain")
+
+ headers.append("Rule")
+ sorts.append("rule")
+
+ try:
+ rows.sort(key=itemgetter(sorts.index(sort)))
+ except ValueError:
+ pass
+
+ rows.insert(0, headers)
+ widths = [max(len(row[i]) for row in rows) for i in range(len(headers))]
+ rows.insert(1, ["-" * w for w in widths])
+ template = " ".join(f"{{{i}:<{w}}}" for i, w in enumerate(widths))
+
+ for row in rows:
+ click.echo(template.format(*row))
+
+
+cli = FlaskGroup(
+ name="flask",
+ help="""\
+A general utility script for Flask applications.
An application to load must be given with the '--app' option,
'FLASK_APP' environment variable, or with a 'wsgi.py' or 'app.py' file
in the current directory.
-"""
- )
-if __name__ == '__main__':
+""",
+)
+
+
+def main() -> None:
+ cli.main()
+
+
+if __name__ == "__main__":
main()
diff --git a/src/flask/config.py b/src/flask/config.py
index 917b25e6..7e3ba179 100644
--- a/src/flask/config.py
+++ b/src/flask/config.py
@@ -1,46 +1,53 @@
from __future__ import annotations
+
import errno
import json
import os
import types
import typing as t
+
from werkzeug.utils import import_string
+
if t.TYPE_CHECKING:
import typing_extensions as te
+
from .sansio.app import App
-T = t.TypeVar('T')
+
+
+T = t.TypeVar("T")
class ConfigAttribute(t.Generic[T]):
"""Makes an attribute forward to the config"""
- def __init__(self, name: str, get_converter: (t.Callable[[t.Any], T] |
- None)=None) ->None:
+ def __init__(
+ self, name: str, get_converter: t.Callable[[t.Any], T] | None = None
+ ) -> None:
self.__name__ = name
self.get_converter = get_converter
@t.overload
- def __get__(self, obj: None, owner: None) ->te.Self:
- ...
+ def __get__(self, obj: None, owner: None) -> te.Self: ...
@t.overload
- def __get__(self, obj: App, owner: type[App]) ->T:
- ...
+ def __get__(self, obj: App, owner: type[App]) -> T: ...
- def __get__(self, obj: (App | None), owner: (type[App] | None)=None) ->(T |
- te.Self):
+ def __get__(self, obj: App | None, owner: type[App] | None = None) -> T | te.Self:
if obj is None:
return self
+
rv = obj.config[self.__name__]
+
if self.get_converter is not None:
rv = self.get_converter(rv)
- return rv
- def __set__(self, obj: App, value: t.Any) ->None:
+ return rv # type: ignore[no-any-return]
+
+ def __set__(self, obj: App, value: t.Any) -> None:
obj.config[self.__name__] = value
-class Config(dict):
+class Config(dict): # type: ignore[type-arg]
"""Works exactly like a dict but provides ways to fill it from files
or special dictionaries. There are two common patterns to populate the
config.
@@ -84,12 +91,15 @@ class Config(dict):
:param defaults: an optional dictionary of default values
"""
- def __init__(self, root_path: (str | os.PathLike[str]), defaults: (dict
- [str, t.Any] | None)=None) ->None:
+ def __init__(
+ self,
+ root_path: str | os.PathLike[str],
+ defaults: dict[str, t.Any] | None = None,
+ ) -> None:
super().__init__(defaults or {})
self.root_path = root_path
- def from_envvar(self, variable_name: str, silent: bool=False) ->bool:
+ def from_envvar(self, variable_name: str, silent: bool = False) -> bool:
"""Loads a configuration from an environment variable pointing to
a configuration file. This is basically just a shortcut with nicer
error messages for this line of code::
@@ -101,10 +111,21 @@ class Config(dict):
files.
:return: ``True`` if the file was loaded successfully.
"""
- pass
-
- def from_prefixed_env(self, prefix: str='FLASK', *, loads: t.Callable[[
- str], t.Any]=json.loads) ->bool:
+ rv = os.environ.get(variable_name)
+ if not rv:
+ if silent:
+ return False
+ raise RuntimeError(
+ f"The environment variable {variable_name!r} is not set"
+ " and as such configuration could not be loaded. Set"
+ " this variable and make it point to a configuration"
+ " file"
+ )
+ return self.from_pyfile(rv, silent=silent)
+
+ def from_prefixed_env(
+ self, prefix: str = "FLASK", *, loads: t.Callable[[str], t.Any] = json.loads
+ ) -> bool:
"""Load any environment variables that start with ``FLASK_``,
dropping the prefix from the env key for the config key. Values
are passed through a loading function to attempt to convert them
@@ -128,10 +149,47 @@ class Config(dict):
.. versionadded:: 2.1
"""
- pass
+ prefix = f"{prefix}_"
+ len_prefix = len(prefix)
+
+ for key in sorted(os.environ):
+ if not key.startswith(prefix):
+ continue
+
+ value = os.environ[key]
+
+ try:
+ value = loads(value)
+ except Exception:
+ # Keep the value as a string if loading failed.
+ pass
- def from_pyfile(self, filename: (str | os.PathLike[str]), silent: bool=
- False) ->bool:
+ # Change to key.removeprefix(prefix) on Python >= 3.9.
+ key = key[len_prefix:]
+
+ if "__" not in key:
+ # A non-nested key, set directly.
+ self[key] = value
+ continue
+
+ # Traverse nested dictionaries with keys separated by "__".
+ current = self
+ *parts, tail = key.split("__")
+
+ for part in parts:
+ # If an intermediate dict does not exist, create it.
+ if part not in current:
+ current[part] = {}
+
+ current = current[part]
+
+ current[tail] = value
+
+ return True
+
+ def from_pyfile(
+ self, filename: str | os.PathLike[str], silent: bool = False
+ ) -> bool:
"""Updates the values in the config from a Python file. This function
behaves as if the file was imported as module with the
:meth:`from_object` function.
@@ -146,9 +204,21 @@ class Config(dict):
.. versionadded:: 0.7
`silent` parameter.
"""
- pass
-
- def from_object(self, obj: (object | str)) ->None:
+ filename = os.path.join(self.root_path, filename)
+ d = types.ModuleType("config")
+ d.__file__ = filename
+ try:
+ with open(filename, mode="rb") as config_file:
+ exec(compile(config_file.read(), filename, "exec"), d.__dict__)
+ except OSError as e:
+ if silent and e.errno in (errno.ENOENT, errno.EISDIR, errno.ENOTDIR):
+ return False
+ e.strerror = f"Unable to load configuration file ({e.strerror})"
+ raise
+ self.from_object(d)
+ return True
+
+ def from_object(self, obj: object | str) -> None:
"""Updates the values from the given object. An object can be of one
of the following two types:
@@ -180,11 +250,19 @@ class Config(dict):
:param obj: an import name or object
"""
- pass
-
- def from_file(self, filename: (str | os.PathLike[str]), load: t.
- Callable[[t.IO[t.Any]], t.Mapping[str, t.Any]], silent: bool=False,
- text: bool=True) ->bool:
+ if isinstance(obj, str):
+ obj = import_string(obj)
+ for key in dir(obj):
+ if key.isupper():
+ self[key] = getattr(obj, key)
+
+ def from_file(
+ self,
+ filename: str | os.PathLike[str],
+ load: t.Callable[[t.IO[t.Any]], t.Mapping[str, t.Any]],
+ silent: bool = False,
+ text: bool = True,
+ ) -> bool:
"""Update the values in the config from a file that is loaded
using the ``load`` parameter. The loaded data is passed to the
:meth:`from_mapping` method.
@@ -212,10 +290,23 @@ class Config(dict):
.. versionadded:: 2.0
"""
- pass
+ filename = os.path.join(self.root_path, filename)
+
+ try:
+ with open(filename, "r" if text else "rb") as f:
+ obj = load(f)
+ except OSError as e:
+ if silent and e.errno in (errno.ENOENT, errno.EISDIR):
+ return False
+
+ e.strerror = f"Unable to load configuration file ({e.strerror})"
+ raise
- def from_mapping(self, mapping: (t.Mapping[str, t.Any] | None)=None, **
- kwargs: t.Any) ->bool:
+ return self.from_mapping(obj)
+
+ def from_mapping(
+ self, mapping: t.Mapping[str, t.Any] | None = None, **kwargs: t.Any
+ ) -> bool:
"""Updates the config like :meth:`update` ignoring items with
non-upper keys.
@@ -223,10 +314,18 @@ class Config(dict):
.. versionadded:: 0.11
"""
- pass
-
- def get_namespace(self, namespace: str, lowercase: bool=True,
- trim_namespace: bool=True) ->dict[str, t.Any]:
+ mappings: dict[str, t.Any] = {}
+ if mapping is not None:
+ mappings.update(mapping)
+ mappings.update(kwargs)
+ for key, value in mappings.items():
+ if key.isupper():
+ self[key] = value
+ return True
+
+ def get_namespace(
+ self, namespace: str, lowercase: bool = True, trim_namespace: bool = True
+ ) -> dict[str, t.Any]:
"""Returns a dictionary containing a subset of configuration options
that match the specified namespace/prefix. Example usage::
@@ -254,7 +353,18 @@ class Config(dict):
.. versionadded:: 0.11
"""
- pass
+ rv = {}
+ for k, v in self.items():
+ if not k.startswith(namespace):
+ continue
+ if trim_namespace:
+ key = k[len(namespace) :]
+ else:
+ key = k
+ if lowercase:
+ key = key.lower()
+ rv[key] = v
+ return rv
- def __repr__(self) ->str:
- return f'<{type(self).__name__} {dict.__repr__(self)}>'
+ def __repr__(self) -> str:
+ return f"<{type(self).__name__} {dict.__repr__(self)}>"
diff --git a/src/flask/ctx.py b/src/flask/ctx.py
index ce2683ae..9b164d39 100644
--- a/src/flask/ctx.py
+++ b/src/flask/ctx.py
@@ -1,20 +1,28 @@
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 . 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:
+
+if t.TYPE_CHECKING: # pragma: no cover
from _typeshed.wsgi import WSGIEnvironment
+
from .app import Flask
from .sessions import SessionMixin
from .wrappers import Request
+
+
+# a singleton sentinel value for parameter defaults
_sentinel = object()
@@ -38,22 +46,25 @@ class _AppCtxGlobals:
.. versionadded:: 0.10
"""
- def __getattr__(self, name: str) ->t.Any:
+ # Define attr methods to let mypy know this is a namespace object
+ # that has arbitrary attributes.
+
+ def __getattr__(self, name: str) -> t.Any:
try:
return self.__dict__[name]
except KeyError:
raise AttributeError(name) from None
- def __setattr__(self, name: str, value: t.Any) ->None:
+ def __setattr__(self, name: str, value: t.Any) -> None:
self.__dict__[name] = value
- def __delattr__(self, name: str) ->None:
+ def __delattr__(self, name: str) -> None:
try:
del self.__dict__[name]
except KeyError:
raise AttributeError(name) from None
- def get(self, name: str, default: (t.Any | None)=None) ->t.Any:
+ def get(self, name: str, default: t.Any | None = None) -> t.Any:
"""Get an attribute by name, or a default value. Like
:meth:`dict.get`.
@@ -62,9 +73,9 @@ class _AppCtxGlobals:
.. versionadded:: 0.10
"""
- pass
+ return self.__dict__.get(name, default)
- def pop(self, name: str, default: t.Any=_sentinel) ->t.Any:
+ def pop(self, name: str, default: t.Any = _sentinel) -> t.Any:
"""Get and remove an attribute by name. Like :meth:`dict.pop`.
:param name: Name of attribute to pop.
@@ -73,9 +84,12 @@ class _AppCtxGlobals:
.. versionadded:: 0.11
"""
- pass
+ if default is _sentinel:
+ return self.__dict__.pop(name)
+ else:
+ return self.__dict__.pop(name, default)
- def setdefault(self, name: str, default: t.Any=None) ->t.Any:
+ def setdefault(self, name: str, default: t.Any = None) -> t.Any:
"""Get the value of an attribute if it is present, otherwise
set and return a default value. Like :meth:`dict.setdefault`.
@@ -85,23 +99,24 @@ class _AppCtxGlobals:
.. versionadded:: 0.11
"""
- pass
+ return self.__dict__.setdefault(name, default)
- def __contains__(self, item: str) ->bool:
+ def __contains__(self, item: str) -> bool:
return item in self.__dict__
- def __iter__(self) ->t.Iterator[str]:
+ def __iter__(self) -> t.Iterator[str]:
return iter(self.__dict__)
- def __repr__(self) ->str:
+ def __repr__(self) -> str:
ctx = _cv_app.get(None)
if ctx is not None:
return f"<flask.g of '{ctx.app.name}'>"
return object.__repr__(self)
-def after_this_request(f: ft.AfterRequestCallable[t.Any]
- ) ->ft.AfterRequestCallable[t.Any]:
+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.
@@ -122,13 +137,22 @@ def after_this_request(f: ft.AfterRequestCallable[t.Any]
.. versionadded:: 0.9
"""
- pass
+ ctx = _cv_request.get(None)
+ if ctx is None:
+ raise RuntimeError(
+ "'after_this_request' can only be used when a request"
+ " context is active, such as in a view function."
+ )
-F = t.TypeVar('F', bound=t.Callable[..., t.Any])
+ ctx._after_request_functions.append(f)
+ return f
-def copy_current_request_context(f: F) ->F:
+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
@@ -152,10 +176,24 @@ def copy_current_request_context(f: F) ->F:
.. versionadded:: 0.10
"""
- pass
+ ctx = _cv_request.get(None)
+
+ if ctx is None:
+ raise RuntimeError(
+ "'copy_current_request_context' can only be used when a"
+ " request context is active, such as in a view function."
+ )
+
+ ctx = ctx.copy()
+ def wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any:
+ with ctx: # type: ignore[union-attr]
+ return ctx.app.ensure_sync(f)(*args, **kwargs) # type: ignore[union-attr]
-def has_request_context() ->bool:
+ return update_wrapper(wrapper, f) # type: ignore[return-value]
+
+
+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
@@ -184,17 +222,17 @@ def has_request_context() ->bool:
.. versionadded:: 0.7
"""
- pass
+ return _cv_request.get(None) is not None
-def has_app_context() ->bool:
+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.
.. versionadded:: 0.9
"""
- pass
+ return _cv_app.get(None) is not None
class AppContext:
@@ -204,26 +242,45 @@ class AppContext:
running CLI commands.
"""
- def __init__(self, app: Flask) ->None:
+ 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]] = []
- def push(self) ->None:
+ def push(self) -> None:
"""Binds the app context to the current context."""
- pass
+ self._cv_tokens.append(_cv_app.set(self))
+ appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync)
- def pop(self, exc: (BaseException | None)=_sentinel) ->None:
+ def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore
"""Pops the app context."""
- pass
+ 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())
+
+ if ctx is not self:
+ raise AssertionError(
+ f"Popped wrong app context. ({ctx!r} instead of {self!r})"
+ )
+
+ appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync)
- def __enter__(self) ->AppContext:
+ def __enter__(self) -> AppContext:
self.push()
return self
- def __exit__(self, exc_type: (type | None), exc_value: (BaseException |
- None), tb: (TracebackType | None)) ->None:
+ def __exit__(
+ self,
+ exc_type: type | None,
+ exc_value: BaseException | None,
+ tb: TracebackType | None,
+ ) -> None:
self.pop(exc_value)
@@ -249,8 +306,13 @@ class RequestContext:
database connections.
"""
- def __init__(self, app: Flask, environ: WSGIEnvironment, request: (
- Request | None)=None, session: (SessionMixin | None)=None) ->None:
+ 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)
@@ -263,12 +325,16 @@ class RequestContext:
self.request.routing_exception = e
self.flashes: list[tuple[str, str]] | None = None
self.session: SessionMixin | None = session
- self._after_request_functions: list[ft.AfterRequestCallable[t.Any]] = [
- ]
- self._cv_tokens: list[tuple[contextvars.Token[RequestContext],
- AppContext | None]] = []
+ # Functions that should be executed after the request on the response
+ # object. These will be called before the regular "after_request"
+ # functions.
+ self._after_request_functions: list[ft.AfterRequestCallable[t.Any]] = []
- def copy(self) ->RequestContext:
+ self._cv_tokens: list[
+ tuple[contextvars.Token[RequestContext], AppContext | None]
+ ] = []
+
+ 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
@@ -281,15 +347,53 @@ class RequestContext:
The current session object is used instead of reloading the original
data. This prevents `flask.session` pointing to an out-of-date object.
"""
- pass
-
- def match_request(self) ->None:
+ return self.__class__(
+ self.app,
+ environ=self.request.environ,
+ request=self.request,
+ session=self.session,
+ )
+
+ def match_request(self) -> None:
"""Can be overridden by a subclass to hook into the matching
of the request.
"""
- pass
+ try:
+ result = self.url_adapter.match(return_rule=True) # type: ignore
+ self.request.url_rule, self.request.view_args = result # type: ignore
+ except HTTPException as e:
+ self.request.routing_exception = e
+
+ 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)
- def pop(self, exc: (BaseException | None)=_sentinel) ->None:
+ 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
+
+ self._cv_tokens.append((_cv_request.set(self), app_ctx))
+
+ # 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.session is None:
+ self.session = session_interface.make_null_session(self.app)
+
+ # 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:
+ 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.
@@ -297,17 +401,49 @@ class RequestContext:
.. versionchanged:: 0.9
Added the `exc` argument.
"""
- pass
+ clear_request = len(self._cv_tokens) == 1
- def __enter__(self) ->RequestContext:
+ try:
+ if clear_request:
+ if exc is _sentinel:
+ exc = sys.exc_info()[1]
+ self.app.do_teardown_request(exc)
+
+ request_close = getattr(self.request, "close", None)
+ if request_close is not None:
+ request_close()
+ finally:
+ ctx = _cv_request.get()
+ token, app_ctx = self._cv_tokens.pop()
+ _cv_request.reset(token)
+
+ # 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:
self.push()
return self
- def __exit__(self, exc_type: (type | None), exc_value: (BaseException |
- None), tb: (TracebackType | None)) ->None:
+ def __exit__(
+ self,
+ exc_type: type | None,
+ exc_value: BaseException | None,
+ tb: TracebackType | None,
+ ) -> None:
self.pop(exc_value)
- def __repr__(self) ->str:
+ def __repr__(self) -> str:
return (
- f'<{type(self).__name__} {self.request.url!r} [{self.request.method}] of {self.app.name}>'
- )
+ f"<{type(self).__name__} {self.request.url!r}"
+ f" [{self.request.method}] of {self.app.name}>"
+ )
diff --git a/src/flask/debughelpers.py b/src/flask/debughelpers.py
index ac217f75..2c8c4c48 100644
--- a/src/flask/debughelpers.py
+++ b/src/flask/debughelpers.py
@@ -1,10 +1,14 @@
from __future__ import annotations
+
import typing as t
+
from jinja2.loaders import BaseLoader
from werkzeug.routing import RequestRedirect
+
from .blueprints import Blueprint
from .globals import request_ctx
from .sansio.app import App
+
if t.TYPE_CHECKING:
from .sansio.scaffold import Scaffold
from .wrappers import Request
@@ -21,21 +25,25 @@ class DebugFilesKeyError(KeyError, AssertionError):
provide a better error message than just a generic KeyError/BadRequest.
"""
- def __init__(self, request: Request, key: str) ->None:
+ def __init__(self, request: Request, key: str) -> None:
form_matches = request.form.getlist(key)
buf = [
- f'You tried to access the file {key!r} in the request.files dictionary but it does not exist. The mimetype for the request is {request.mimetype!r} instead of \'multipart/form-data\' which means that no file contents were transmitted. To fix this error you should provide enctype="multipart/form-data" in your form.'
- ]
+ f"You tried to access the file {key!r} in the request.files"
+ " dictionary but it does not exist. The mimetype for the"
+ f" request is {request.mimetype!r} instead of"
+ " 'multipart/form-data' which means that no file contents"
+ " were transmitted. To fix this error you should provide"
+ ' enctype="multipart/form-data" in your form.'
+ ]
if form_matches:
- names = ', '.join(repr(x) for x in form_matches)
+ names = ", ".join(repr(x) for x in form_matches)
buf.append(
- f"""
-
-The browser instead transmitted some file names. This was submitted: {names}"""
- )
- self.msg = ''.join(buf)
+ "\n\nThe browser instead transmitted some file names. "
+ f"This was submitted: {names}"
+ )
+ self.msg = "".join(buf)
- def __str__(self) ->str:
+ def __str__(self) -> str:
return self.msg
@@ -46,36 +54,125 @@ class FormDataRoutingRedirect(AssertionError):
307 or 308.
"""
- def __init__(self, request: Request) ->None:
+ def __init__(self, request: Request) -> None:
exc = request.routing_exception
assert isinstance(exc, RequestRedirect)
buf = [
- f"A request was sent to '{request.url}', but routing issued a redirect to the canonical URL '{exc.new_url}'."
- ]
- if f'{request.base_url}/' == exc.new_url.partition('?')[0]:
- buf.append(
- ' The URL was defined with a trailing slash. Flask will redirect to the URL with a trailing slash if it was accessed without one.'
- )
- buf.append(
- """ Send requests to the canonical URL, or use 307 or 308 for routing redirects. Otherwise, browsers will drop form data.
+ f"A request was sent to '{request.url}', but routing issued"
+ f" a redirect to the canonical URL '{exc.new_url}'."
+ ]
-This exception is only raised in debug mode."""
+ if f"{request.base_url}/" == exc.new_url.partition("?")[0]:
+ buf.append(
+ " The URL was defined with a trailing slash. Flask"
+ " will redirect to the URL with a trailing slash if it"
+ " was accessed without one."
)
- super().__init__(''.join(buf))
+
+ buf.append(
+ " Send requests to the canonical URL, or use 307 or 308 for"
+ " routing redirects. Otherwise, browsers will drop form"
+ " data.\n\n"
+ "This exception is only raised in debug mode."
+ )
+ super().__init__("".join(buf))
-def attach_enctype_error_multidict(request: Request) ->None:
+def attach_enctype_error_multidict(request: Request) -> None:
"""Patch ``request.files.__getitem__`` to raise a descriptive error
about ``enctype=multipart/form-data``.
:param request: The request to patch.
:meta private:
"""
- pass
-
-
-def explain_template_loading_attempts(app: App, template: str, attempts:
- list[tuple[BaseLoader, Scaffold, tuple[str, str | None, t.Callable[[],
- bool] | None] | None]]) ->None:
+ oldcls = request.files.__class__
+
+ class newcls(oldcls): # type: ignore[valid-type, misc]
+ def __getitem__(self, key: str) -> t.Any:
+ try:
+ return super().__getitem__(key)
+ except KeyError as e:
+ if key not in request.form:
+ raise
+
+ raise DebugFilesKeyError(request, key).with_traceback(
+ e.__traceback__
+ ) from None
+
+ newcls.__name__ = oldcls.__name__
+ newcls.__module__ = oldcls.__module__
+ request.files.__class__ = newcls
+
+
+def _dump_loader_info(loader: BaseLoader) -> t.Iterator[str]:
+ yield f"class: {type(loader).__module__}.{type(loader).__name__}"
+ for key, value in sorted(loader.__dict__.items()):
+ if key.startswith("_"):
+ continue
+ if isinstance(value, (tuple, list)):
+ if not all(isinstance(x, str) for x in value):
+ continue
+ yield f"{key}:"
+ for item in value:
+ yield f" - {item}"
+ continue
+ elif not isinstance(value, (str, int, float, bool)):
+ continue
+ yield f"{key}: {value!r}"
+
+
+def explain_template_loading_attempts(
+ app: App,
+ template: str,
+ attempts: list[
+ tuple[
+ BaseLoader,
+ Scaffold,
+ tuple[str, str | None, t.Callable[[], bool] | None] | None,
+ ]
+ ],
+) -> None:
"""This should help developers understand what failed"""
- pass
+ 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
+
+ for idx, (loader, srcobj, triple) in enumerate(attempts):
+ if isinstance(srcobj, App):
+ src_info = f"application {srcobj.import_name!r}"
+ elif isinstance(srcobj, Blueprint):
+ src_info = f"blueprint {srcobj.name!r} ({srcobj.import_name})"
+ else:
+ src_info = repr(srcobj)
+
+ info.append(f"{idx + 1:5}: trying loader of {src_info}")
+
+ for line in _dump_loader_info(loader):
+ info.append(f" {line}")
+
+ if triple is None:
+ detail = "no match"
+ else:
+ detail = f"found ({triple[1] or '<string>'!r})"
+ total_found += 1
+ info.append(f" -> {detail}")
+
+ seems_fishy = False
+ if total_found == 0:
+ info.append("Error: the template could not be found.")
+ seems_fishy = True
+ elif total_found > 1:
+ info.append("Warning: multiple loaders returned a match for the template.")
+ seems_fishy = True
+
+ if blueprint is not None and seems_fishy:
+ info.append(
+ " The template was looked up from an endpoint that belongs"
+ f" to the blueprint {blueprint!r}."
+ )
+ info.append(" Maybe you did not place a template in the right folder?")
+ info.append(" See https://flask.palletsprojects.com/blueprints/#templates")
+
+ app.logger.info("\n".join(info))
diff --git a/src/flask/globals.py b/src/flask/globals.py
index 4a5f4a78..e2c410cc 100644
--- a/src/flask/globals.py
+++ b/src/flask/globals.py
@@ -1,32 +1,51 @@
from __future__ import annotations
+
import typing as t
from contextvars import ContextVar
+
from werkzeug.local import LocalProxy
-if t.TYPE_CHECKING:
+
+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
-_no_app_msg = """Working outside of application context.
+
+
+_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."""
-_cv_app: ContextVar[AppContext] = ContextVar('flask.app_ctx')
-app_ctx: AppContext = LocalProxy(_cv_app, unbound_message=_no_app_msg)
-current_app: Flask = LocalProxy(_cv_app, 'app', unbound_message=_no_app_msg)
-g: _AppCtxGlobals = LocalProxy(_cv_app, 'g', unbound_message=_no_app_msg)
-_no_req_msg = """Working outside of request context.
+with app.app_context(). See the documentation for more information.\
+"""
+_cv_app: ContextVar[AppContext] = ContextVar("flask.app_ctx")
+app_ctx: AppContext = LocalProxy( # type: ignore[assignment]
+ _cv_app, unbound_message=_no_app_msg
+)
+current_app: Flask = LocalProxy( # type: ignore[assignment]
+ _cv_app, "app", unbound_message=_no_app_msg
+)
+g: _AppCtxGlobals = 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."""
-_cv_request: ContextVar[RequestContext] = ContextVar('flask.request_ctx')
-request_ctx: RequestContext = LocalProxy(_cv_request, unbound_message=
- _no_req_msg)
-request: Request = LocalProxy(_cv_request, 'request', unbound_message=
- _no_req_msg)
-session: SessionMixin = LocalProxy(_cv_request, 'session', unbound_message=
- _no_req_msg)
+information about how to avoid this problem.\
+"""
+_cv_request: ContextVar[RequestContext] = ContextVar("flask.request_ctx")
+request_ctx: RequestContext = LocalProxy( # type: ignore[assignment]
+ _cv_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
+)
diff --git a/src/flask/helpers.py b/src/flask/helpers.py
index 2061dbf3..359a842a 100644
--- a/src/flask/helpers.py
+++ b/src/flask/helpers.py
@@ -1,4 +1,5 @@
from __future__ import annotations
+
import importlib.util
import os
import sys
@@ -6,39 +7,49 @@ import typing as t
from datetime import datetime
from functools import lru_cache
from functools import update_wrapper
+
import werkzeug.utils
from werkzeug.exceptions import abort as _wz_abort
from werkzeug.utils import redirect as _wz_redirect
from werkzeug.wrappers import Response as BaseResponse
+
from .globals import _cv_request
from .globals import current_app
from .globals import request
from .globals import request_ctx
from .globals import session
from .signals import message_flashed
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from .wrappers import Response
-def get_debug_flag() ->bool:
+def get_debug_flag() -> bool:
"""Get whether debug mode should be enabled for the app, indicated by the
:envvar:`FLASK_DEBUG` environment variable. The default is ``False``.
"""
- pass
+ val = os.environ.get("FLASK_DEBUG")
+ return bool(val and val.lower() not in {"0", "false", "no"})
-def get_load_dotenv(default: bool=True) ->bool:
+def get_load_dotenv(default: bool = True) -> bool:
"""Get whether the user has disabled loading default dotenv files by
setting :envvar:`FLASK_SKIP_DOTENV`. The default is ``True``, load
the files.
:param default: What to return if the env var isn't set.
"""
- pass
+ val = os.environ.get("FLASK_SKIP_DOTENV")
+ if not val:
+ return default
-def stream_with_context(generator_or_function: (t.Iterator[t.AnyStr] | t.
- Callable[..., t.Iterator[t.AnyStr]])) ->t.Iterator[t.AnyStr]:
+ return val.lower() in ("0", "false", "no")
+
+
+def stream_with_context(
+ generator_or_function: t.Iterator[t.AnyStr] | t.Callable[..., t.Iterator[t.AnyStr]],
+) -> t.Iterator[t.AnyStr]:
"""Request contexts disappear when the response is started on the server.
This is done for efficiency reasons and to make it less likely to encounter
memory leaks with badly written WSGI middlewares. The downside is that if
@@ -72,10 +83,48 @@ def stream_with_context(generator_or_function: (t.Iterator[t.AnyStr] | t.
.. versionadded:: 0.9
"""
- pass
-
-
-def make_response(*args: t.Any) ->Response:
+ try:
+ gen = iter(generator_or_function) # type: ignore[arg-type]
+ except TypeError:
+
+ def decorator(*args: t.Any, **kwargs: t.Any) -> t.Any:
+ gen = generator_or_function(*args, **kwargs) # type: ignore[operator]
+ return stream_with_context(gen)
+
+ return update_wrapper(decorator, generator_or_function) # type: ignore[arg-type]
+
+ def generator() -> t.Iterator[t.AnyStr | None]:
+ ctx = _cv_request.get(None)
+ if ctx is None:
+ raise RuntimeError(
+ "'stream_with_context' can only be used when a request"
+ " context is active, such as in a view function."
+ )
+ with ctx:
+ # Dummy sentinel. Has to be inside the context block or we're
+ # not actually keeping the context around.
+ yield None
+
+ # The try/finally is here so that if someone passes a WSGI level
+ # iterator in we're still running the cleanup logic. Generators
+ # don't need that because they are closed on their destruction
+ # automatically.
+ try:
+ yield from gen
+ finally:
+ if hasattr(gen, "close"):
+ gen.close()
+
+ # The trick is to start the generator. Then the code execution runs until
+ # the first dummy None is yielded at which point the context was already
+ # pushed. This item is discarded. Then when the iteration continues the
+ # real generator is executed.
+ wrapped_g = generator()
+ next(wrapped_g)
+ return wrapped_g # type: ignore[return-value]
+
+
+def make_response(*args: t.Any) -> Response:
"""Sometimes it is necessary to set additional headers in a view. Because
views do not have to return response objects but can return a value that
is converted into a response object by Flask itself, it becomes tricky to
@@ -117,12 +166,22 @@ def make_response(*args: t.Any) ->Response:
.. versionadded:: 0.6
"""
- pass
-
-
-def url_for(endpoint: str, *, _anchor: (str | None)=None, _method: (str |
- None)=None, _scheme: (str | None)=None, _external: (bool | None)=None,
- **values: t.Any) ->str:
+ if not args:
+ return current_app.response_class()
+ if len(args) == 1:
+ args = args[0]
+ return current_app.make_response(args)
+
+
+def url_for(
+ endpoint: str,
+ *,
+ _anchor: str | None = None,
+ _method: str | None = None,
+ _scheme: str | None = None,
+ _external: bool | None = None,
+ **values: t.Any,
+) -> str:
"""Generate a URL to the given endpoint with the given values.
This requires an active request or application context, and calls
@@ -158,11 +217,19 @@ def url_for(endpoint: str, *, _anchor: (str | None)=None, _method: (str |
.. versionchanged:: 0.9
Calls ``app.handle_url_build_error`` on build errors.
"""
- pass
-
-
-def redirect(location: str, code: int=302, Response: (type[BaseResponse] |
- None)=None) ->BaseResponse:
+ return current_app.url_for(
+ endpoint,
+ _anchor=_anchor,
+ _method=_method,
+ _scheme=_scheme,
+ _external=_external,
+ **values,
+ )
+
+
+def redirect(
+ location: str, code: int = 302, Response: type[BaseResponse] | None = None
+) -> BaseResponse:
"""Create a redirect response object.
If :data:`~flask.current_app` is available, it will use its
@@ -178,11 +245,13 @@ def redirect(location: str, code: int=302, Response: (type[BaseResponse] |
Calls ``current_app.redirect`` if available instead of always
using Werkzeug's default ``redirect``.
"""
- pass
+ if current_app:
+ return current_app.redirect(location, code=code)
+
+ return _wz_redirect(location, code=code, Response=Response)
-def abort(code: (int | BaseResponse), *args: t.Any, **kwargs: t.Any
- ) ->t.NoReturn:
+def abort(code: int | BaseResponse, *args: t.Any, **kwargs: t.Any) -> t.NoReturn:
"""Raise an :exc:`~werkzeug.exceptions.HTTPException` for the given
status code.
@@ -199,10 +268,13 @@ def abort(code: (int | BaseResponse), *args: t.Any, **kwargs: t.Any
Calls ``current_app.aborter`` if available instead of always
using Werkzeug's default ``abort``.
"""
- pass
+ if current_app:
+ current_app.aborter(code, *args, **kwargs)
+ _wz_abort(code, *args, **kwargs)
-def get_template_attribute(template_name: str, attribute: str) ->t.Any:
+
+def get_template_attribute(template_name: str, attribute: str) -> t.Any:
"""Loads a macro (or variable) a template exports. This can be used to
invoke a macro from within Python code. If you for example have a
template named :file:`_cider.html` with the following contents:
@@ -221,10 +293,10 @@ def get_template_attribute(template_name: str, attribute: str) ->t.Any:
:param template_name: the name of the template
:param attribute: the name of the variable of macro to access
"""
- pass
+ return getattr(current_app.jinja_env.get_template(template_name).module, attribute)
-def flash(message: str, category: str='message') ->None:
+def flash(message: str, category: str = "message") -> None:
"""Flashes a message to the next request. In order to remove the
flashed message from the session and to display it to the user,
the template has to call :func:`get_flashed_messages`.
@@ -239,11 +311,28 @@ def flash(message: str, category: str='message') ->None:
messages and ``'warning'`` for warnings. However any
kind of string can be used as category.
"""
- pass
-
-
-def get_flashed_messages(with_categories: bool=False, category_filter: t.
- Iterable[str]=()) ->(list[str] | list[tuple[str, str]]):
+ # Original implementation:
+ #
+ # session.setdefault('_flashes', []).append((category, message))
+ #
+ # This assumed that changes made to mutable structures in the session are
+ # always in sync with the session object, which is not true for session
+ # implementations that use external storage for keeping their keys/values.
+ flashes = session.get("_flashes", [])
+ flashes.append((category, message))
+ session["_flashes"] = flashes
+ app = current_app._get_current_object() # type: ignore
+ message_flashed.send(
+ app,
+ _async_wrapper=app.ensure_sync,
+ message=message,
+ category=category,
+ )
+
+
+def get_flashed_messages(
+ with_categories: bool = False, category_filter: t.Iterable[str] = ()
+) -> list[str] | list[tuple[str, str]]:
"""Pulls all flashed messages from the session and returns them.
Further calls in the same request to the function will return
the same messages. By default just the messages are returned,
@@ -272,14 +361,40 @@ def get_flashed_messages(with_categories: bool=False, category_filter: t.
:param category_filter: filter of categories to limit return values. Only
categories in the list will be returned.
"""
- pass
-
-
-def send_file(path_or_file: (os.PathLike[t.AnyStr] | str | t.BinaryIO),
- mimetype: (str | None)=None, as_attachment: bool=False, download_name:
- (str | None)=None, conditional: bool=True, etag: (bool | str)=True,
- last_modified: (datetime | int | float | None)=None, max_age: (None | (
- int | t.Callable[[str | None], int | None]))=None) ->Response:
+ flashes = request_ctx.flashes
+ if flashes is None:
+ flashes = session.pop("_flashes") if "_flashes" in session else []
+ request_ctx.flashes = flashes
+ if category_filter:
+ flashes = list(filter(lambda f: f[0] in category_filter, flashes))
+ if not with_categories:
+ return [x[1] for x in flashes]
+ return flashes
+
+
+def _prepare_send_file_kwargs(**kwargs: t.Any) -> dict[str, t.Any]:
+ if kwargs.get("max_age") is None:
+ kwargs["max_age"] = current_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, # type: ignore
+ )
+ return kwargs
+
+
+def send_file(
+ path_or_file: os.PathLike[t.AnyStr] | str | t.BinaryIO,
+ mimetype: str | None = None,
+ as_attachment: bool = False,
+ download_name: str | None = None,
+ conditional: bool = True,
+ etag: bool | str = True,
+ last_modified: datetime | int | float | None = None,
+ max_age: None | (int | t.Callable[[str | None], int | None]) = None,
+) -> Response:
"""Send the contents of a file to the client.
The first argument can be a file path or a file-like object. Paths
@@ -381,11 +496,26 @@ def send_file(path_or_file: (os.PathLike[t.AnyStr] | str | t.BinaryIO),
.. versionadded:: 0.2
"""
- pass
-
-
-def send_from_directory(directory: (os.PathLike[str] | str), path: (os.
- PathLike[str] | str), **kwargs: t.Any) ->Response:
+ return werkzeug.utils.send_file( # type: ignore[return-value]
+ **_prepare_send_file_kwargs(
+ path_or_file=path_or_file,
+ environ=request.environ,
+ mimetype=mimetype,
+ as_attachment=as_attachment,
+ download_name=download_name,
+ conditional=conditional,
+ etag=etag,
+ last_modified=last_modified,
+ max_age=max_age,
+ )
+ )
+
+
+def send_from_directory(
+ directory: os.PathLike[str] | str,
+ path: os.PathLike[str] | str,
+ **kwargs: t.Any,
+) -> Response:
"""Send a file from within a directory using :func:`send_file`.
.. code-block:: python
@@ -419,10 +549,12 @@ def send_from_directory(directory: (os.PathLike[str] | str), path: (os.
.. versionadded:: 0.5
"""
- pass
+ return werkzeug.utils.send_from_directory( # type: ignore[return-value]
+ directory, path, **_prepare_send_file_kwargs(**kwargs)
+ )
-def get_root_path(import_name: str) ->str:
+def get_root_path(import_name: str) -> str:
"""Find the root path of a package, or the path that contains a
module. If it cannot be found, returns the current working
directory.
@@ -431,4 +563,59 @@ def get_root_path(import_name: str) ->str:
:meta private:
"""
- pass
+ # Module already imported and has a file attribute. Use that first.
+ mod = sys.modules.get(import_name)
+
+ if mod is not None and hasattr(mod, "__file__") and mod.__file__ is not None:
+ return os.path.dirname(os.path.abspath(mod.__file__))
+
+ # Next attempt: check the loader.
+ try:
+ spec = importlib.util.find_spec(import_name)
+
+ if spec is None:
+ raise ValueError
+ except (ImportError, ValueError):
+ loader = None
+ else:
+ loader = spec.loader
+
+ # Loader does not exist or we're referring to an unloaded main
+ # module or a main module without path (interactive sessions), go
+ # with the current working directory.
+ if loader is None:
+ return os.getcwd()
+
+ if hasattr(loader, "get_filename"):
+ filepath = loader.get_filename(import_name)
+ else:
+ # Fall back to imports.
+ __import__(import_name)
+ mod = sys.modules[import_name]
+ filepath = getattr(mod, "__file__", None)
+
+ # If we don't have a file path it might be because it is a
+ # namespace package. In this case pick the root path from the
+ # first module that is contained in the package.
+ if filepath is None:
+ raise RuntimeError(
+ "No root path can be found for the provided module"
+ f" {import_name!r}. This can happen because the module"
+ " came from an import hook that does not provide file"
+ " name information or because it's a namespace package."
+ " In this case the root path needs to be explicitly"
+ " provided."
+ )
+
+ # filepath is import_name.py for a module, or __init__.py for a package.
+ return os.path.dirname(os.path.abspath(filepath)) # type: ignore[no-any-return]
+
+
+@lru_cache(maxsize=None)
+def _split_blueprint_path(name: str) -> list[str]:
+ out: list[str] = [name]
+
+ if "." in name:
+ out.extend(_split_blueprint_path(name.rpartition(".")[0]))
+
+ return out
diff --git a/src/flask/json/provider.py b/src/flask/json/provider.py
index e9123cb0..f9b2e8ff 100644
--- a/src/flask/json/provider.py
+++ b/src/flask/json/provider.py
@@ -1,4 +1,5 @@
from __future__ import annotations
+
import dataclasses
import decimal
import json
@@ -6,9 +7,12 @@ import typing as t
import uuid
import weakref
from datetime import date
+
from werkzeug.http import http_date
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from werkzeug.sansio.response import Response
+
from ..sansio.app import App
@@ -31,18 +35,18 @@ class JSONProvider:
.. versionadded:: 2.2
"""
- def __init__(self, app: App) ->None:
+ def __init__(self, app: App) -> None:
self._app: App = weakref.proxy(app)
- def dumps(self, obj: t.Any, **kwargs: t.Any) ->str:
+ def dumps(self, obj: t.Any, **kwargs: t.Any) -> str:
"""Serialize data as JSON.
:param obj: The data to serialize.
:param kwargs: May be passed to the underlying JSON library.
"""
- pass
+ raise NotImplementedError
- def dump(self, obj: t.Any, fp: t.IO[str], **kwargs: t.Any) ->None:
+ def dump(self, obj: t.Any, fp: t.IO[str], **kwargs: t.Any) -> None:
"""Serialize data as JSON and write to a file.
:param obj: The data to serialize.
@@ -50,25 +54,39 @@ class JSONProvider:
encoding to be valid JSON.
:param kwargs: May be passed to the underlying JSON library.
"""
- pass
+ fp.write(self.dumps(obj, **kwargs))
- def loads(self, s: (str | bytes), **kwargs: t.Any) ->t.Any:
+ def loads(self, s: str | bytes, **kwargs: t.Any) -> t.Any:
"""Deserialize data as JSON.
:param s: Text or UTF-8 bytes.
:param kwargs: May be passed to the underlying JSON library.
"""
- pass
+ raise NotImplementedError
- def load(self, fp: t.IO[t.AnyStr], **kwargs: t.Any) ->t.Any:
+ def load(self, fp: t.IO[t.AnyStr], **kwargs: t.Any) -> t.Any:
"""Deserialize data as JSON read from a file.
:param fp: A file opened for reading text or UTF-8 bytes.
:param kwargs: May be passed to the underlying JSON library.
"""
- pass
+ return self.loads(fp.read(), **kwargs)
+
+ def _prepare_response_obj(
+ self, args: tuple[t.Any, ...], kwargs: dict[str, t.Any]
+ ) -> t.Any:
+ if args and kwargs:
+ raise TypeError("app.json.response() takes either args or kwargs, not both")
+
+ if not args and not kwargs:
+ return None
- def response(self, *args: t.Any, **kwargs: t.Any) ->Response:
+ if len(args) == 1:
+ return args[0]
+
+ return args or kwargs
+
+ def response(self, *args: t.Any, **kwargs: t.Any) -> Response:
"""Serialize the given arguments as JSON, and return a
:class:`~flask.Response` object with the ``application/json``
mimetype.
@@ -83,7 +101,24 @@ class JSONProvider:
treat as a list to serialize.
:param kwargs: Treat as a dict to serialize.
"""
- pass
+ obj = self._prepare_response_obj(args, kwargs)
+ return self._app.response_class(self.dumps(obj), mimetype="application/json")
+
+
+def _default(o: t.Any) -> t.Any:
+ if isinstance(o, date):
+ return http_date(o)
+
+ if isinstance(o, (decimal.Decimal, uuid.UUID)):
+ return str(o)
+
+ if dataclasses and dataclasses.is_dataclass(o):
+ return dataclasses.asdict(o)
+
+ if hasattr(o, "__html__"):
+ return str(o.__html__())
+
+ raise TypeError(f"Object of type {type(o).__name__} is not JSON serializable")
class DefaultJSONProvider(JSONProvider):
@@ -99,31 +134,36 @@ class DefaultJSONProvider(JSONProvider):
- :class:`~markupsafe.Markup` (or any object with a ``__html__``
method) will call the ``__html__`` method to get a string.
"""
- default: t.Callable[[t.Any], t.Any] = staticmethod(_default)
+
+ default: t.Callable[[t.Any], t.Any] = staticmethod(_default) # type: ignore[assignment]
"""Apply this function to any object that :meth:`json.dumps` does
not know how to serialize. It should return a valid JSON type or
raise a ``TypeError``.
"""
+
ensure_ascii = True
"""Replace non-ASCII characters with escape sequences. This may be
more compatible with some clients, but can be disabled for better
performance and size.
"""
+
sort_keys = True
"""Sort the keys in any serialized dicts. This may be useful for
some caching situations, but can be disabled for better performance.
When enabled, keys must all be strings, they are not converted
before sorting.
"""
+
compact: bool | None = None
"""If ``True``, or ``None`` out of debug mode, the :meth:`response`
output will not add indentation, newlines, or spaces. If ``False``,
or ``None`` in debug mode, it will use a non-compact representation.
"""
- mimetype = 'application/json'
+
+ mimetype = "application/json"
"""The mimetype set in :meth:`response`."""
- def dumps(self, obj: t.Any, **kwargs: t.Any) ->str:
+ def dumps(self, obj: t.Any, **kwargs: t.Any) -> str:
"""Serialize data as JSON to a string.
Keyword arguments are passed to :func:`json.dumps`. Sets some
@@ -133,17 +173,20 @@ class DefaultJSONProvider(JSONProvider):
:param obj: The data to serialize.
:param kwargs: Passed to :func:`json.dumps`.
"""
- pass
+ kwargs.setdefault("default", self.default)
+ kwargs.setdefault("ensure_ascii", self.ensure_ascii)
+ kwargs.setdefault("sort_keys", self.sort_keys)
+ return json.dumps(obj, **kwargs)
- def loads(self, s: (str | bytes), **kwargs: t.Any) ->t.Any:
+ def loads(self, s: str | bytes, **kwargs: t.Any) -> t.Any:
"""Deserialize data as JSON from a string or bytes.
:param s: Text or UTF-8 bytes.
:param kwargs: Passed to :func:`json.loads`.
"""
- pass
+ return json.loads(s, **kwargs)
- def response(self, *args: t.Any, **kwargs: t.Any) ->Response:
+ def response(self, *args: t.Any, **kwargs: t.Any) -> Response:
"""Serialize the given arguments as JSON, and return a
:class:`~flask.Response` object with it. The response mimetype
will be "application/json" and can be changed with
@@ -159,4 +202,14 @@ class DefaultJSONProvider(JSONProvider):
treat as a list to serialize.
:param kwargs: Treat as a dict to serialize.
"""
- pass
+ obj = self._prepare_response_obj(args, kwargs)
+ dump_args: dict[str, t.Any] = {}
+
+ if (self.compact is None and self._app.debug) or self.compact is False:
+ dump_args.setdefault("indent", 2)
+ else:
+ dump_args.setdefault("separators", (",", ":"))
+
+ return self._app.response_class(
+ f"{self.dumps(obj, **dump_args)}\n", mimetype=self.mimetype
+ )
diff --git a/src/flask/json/tag.py b/src/flask/json/tag.py
index ded094b2..8dc3629b 100644
--- a/src/flask/json/tag.py
+++ b/src/flask/json/tag.py
@@ -40,46 +40,54 @@ be processed before ``dict``.
app.session_interface.serializer.register(TagOrderedDict, index=0)
"""
+
from __future__ import annotations
+
import typing as t
from base64 import b64decode
from base64 import b64encode
from datetime import datetime
from uuid import UUID
+
from markupsafe import Markup
from werkzeug.http import http_date
from werkzeug.http import parse_date
+
from ..json import dumps
from ..json import loads
class JSONTag:
"""Base class for defining type tags for :class:`TaggedJSONSerializer`."""
- __slots__ = 'serializer',
- key: str = ''
- def __init__(self, serializer: TaggedJSONSerializer) ->None:
+ __slots__ = ("serializer",)
+
+ #: The tag to mark the serialized object with. If empty, this tag is
+ #: only used as an intermediate step during tagging.
+ key: str = ""
+
+ def __init__(self, serializer: TaggedJSONSerializer) -> None:
"""Create a tagger for the given serializer."""
self.serializer = serializer
- def check(self, value: t.Any) ->bool:
+ def check(self, value: t.Any) -> bool:
"""Check if the given value should be tagged by this tag."""
- pass
+ raise NotImplementedError
- def to_json(self, value: t.Any) ->t.Any:
+ def to_json(self, value: t.Any) -> t.Any:
"""Convert the Python object to an object that is a valid JSON type.
The tag will be added later."""
- pass
+ raise NotImplementedError
- def to_python(self, value: t.Any) ->t.Any:
+ def to_python(self, value: t.Any) -> t.Any:
"""Convert the JSON representation back to the correct type. The tag
will already be removed."""
- pass
+ raise NotImplementedError
- def tag(self, value: t.Any) ->dict[str, t.Any]:
+ def tag(self, value: t.Any) -> dict[str, t.Any]:
"""Convert the value to a valid JSON type and add the tag structure
around it."""
- pass
+ return {self.key: self.to_json(value)}
class TagDict(JSONTag):
@@ -88,46 +96,124 @@ class TagDict(JSONTag):
Internally, the dict key is suffixed with `__`, and the suffix is removed
when deserializing.
"""
+
__slots__ = ()
- key = ' di'
+ key = " di"
+
+ def check(self, value: t.Any) -> bool:
+ return (
+ isinstance(value, dict)
+ and len(value) == 1
+ and next(iter(value)) in self.serializer.tags
+ )
+
+ def to_json(self, value: t.Any) -> t.Any:
+ key = next(iter(value))
+ return {f"{key}__": self.serializer.tag(value[key])}
+
+ def to_python(self, value: t.Any) -> t.Any:
+ key = next(iter(value))
+ return {key[:-2]: value[key]}
class PassDict(JSONTag):
__slots__ = ()
+
+ def check(self, value: t.Any) -> bool:
+ return isinstance(value, dict)
+
+ def to_json(self, value: t.Any) -> t.Any:
+ # JSON objects may only have string keys, so don't bother tagging the
+ # key here.
+ return {k: self.serializer.tag(v) for k, v in value.items()}
+
tag = to_json
class TagTuple(JSONTag):
__slots__ = ()
- key = ' t'
+ key = " t"
+
+ def check(self, value: t.Any) -> bool:
+ return isinstance(value, tuple)
+
+ def to_json(self, value: t.Any) -> t.Any:
+ return [self.serializer.tag(item) for item in value]
+
+ def to_python(self, value: t.Any) -> t.Any:
+ return tuple(value)
class PassList(JSONTag):
__slots__ = ()
+
+ def check(self, value: t.Any) -> bool:
+ return isinstance(value, list)
+
+ def to_json(self, value: t.Any) -> t.Any:
+ return [self.serializer.tag(item) for item in value]
+
tag = to_json
class TagBytes(JSONTag):
__slots__ = ()
- key = ' b'
+ key = " b"
+
+ def check(self, value: t.Any) -> bool:
+ return isinstance(value, bytes)
+
+ def to_json(self, value: t.Any) -> t.Any:
+ return b64encode(value).decode("ascii")
+
+ def to_python(self, value: t.Any) -> t.Any:
+ return b64decode(value)
class TagMarkup(JSONTag):
"""Serialize anything matching the :class:`~markupsafe.Markup` API by
having a ``__html__`` method to the result of that method. Always
deserializes to an instance of :class:`~markupsafe.Markup`."""
+
__slots__ = ()
- key = ' m'
+ key = " m"
+
+ def check(self, value: t.Any) -> bool:
+ return callable(getattr(value, "__html__", None))
+
+ def to_json(self, value: t.Any) -> t.Any:
+ return str(value.__html__())
+
+ def to_python(self, value: t.Any) -> t.Any:
+ return Markup(value)
class TagUUID(JSONTag):
__slots__ = ()
- key = ' u'
+ key = " u"
+
+ def check(self, value: t.Any) -> bool:
+ return isinstance(value, UUID)
+
+ def to_json(self, value: t.Any) -> t.Any:
+ return value.hex
+
+ def to_python(self, value: t.Any) -> t.Any:
+ return UUID(value)
class TagDateTime(JSONTag):
__slots__ = ()
- key = ' d'
+ key = " d"
+
+ def check(self, value: t.Any) -> bool:
+ return isinstance(value, datetime)
+
+ def to_json(self, value: t.Any) -> t.Any:
+ return http_date(value)
+
+ def to_python(self, value: t.Any) -> t.Any:
+ return parse_date(value)
class TaggedJSONSerializer:
@@ -144,18 +230,35 @@ class TaggedJSONSerializer:
* :class:`~uuid.UUID`
* :class:`~datetime.datetime`
"""
- __slots__ = 'tags', 'order'
- default_tags = [TagDict, PassDict, TagTuple, PassList, TagBytes,
- TagMarkup, TagUUID, TagDateTime]
- def __init__(self) ->None:
+ __slots__ = ("tags", "order")
+
+ #: Tag classes to bind when creating the serializer. Other tags can be
+ #: added later using :meth:`~register`.
+ default_tags = [
+ TagDict,
+ PassDict,
+ TagTuple,
+ PassList,
+ TagBytes,
+ TagMarkup,
+ TagUUID,
+ TagDateTime,
+ ]
+
+ def __init__(self) -> None:
self.tags: dict[str, JSONTag] = {}
self.order: list[JSONTag] = []
+
for cls in self.default_tags:
self.register(cls)
- def register(self, tag_class: type[JSONTag], force: bool=False, index:
- (int | None)=None) ->None:
+ def register(
+ self,
+ tag_class: type[JSONTag],
+ force: bool = False,
+ index: int | None = None,
+ ) -> None:
"""Register a new tag with this serializer.
:param tag_class: tag class to register. Will be instantiated with this
@@ -169,20 +272,56 @@ class TaggedJSONSerializer:
:raise KeyError: if the tag key is already registered and ``force`` is
not true.
"""
- pass
+ tag = tag_class(self)
+ key = tag.key
+
+ if key:
+ if not force and key in self.tags:
+ raise KeyError(f"Tag '{key}' is already registered.")
+
+ self.tags[key] = tag
- def tag(self, value: t.Any) ->t.Any:
+ if index is None:
+ self.order.append(tag)
+ else:
+ self.order.insert(index, tag)
+
+ def tag(self, value: t.Any) -> t.Any:
"""Convert a value to a tagged representation if necessary."""
- pass
+ for tag in self.order:
+ if tag.check(value):
+ return tag.tag(value)
+
+ return value
- def untag(self, value: dict[str, t.Any]) ->t.Any:
+ def untag(self, value: dict[str, t.Any]) -> t.Any:
"""Convert a tagged representation back to the original type."""
- pass
+ if len(value) != 1:
+ return value
+
+ key = next(iter(value))
+
+ if key not in self.tags:
+ return value
+
+ return self.tags[key].to_python(value[key])
+
+ def _untag_scan(self, value: t.Any) -> t.Any:
+ if isinstance(value, dict):
+ # untag each item recursively
+ value = {k: self._untag_scan(v) for k, v in value.items()}
+ # untag the dict itself
+ value = self.untag(value)
+ elif isinstance(value, list):
+ # untag each item recursively
+ value = [self._untag_scan(item) for item in value]
+
+ return value
- def dumps(self, value: t.Any) ->str:
+ def dumps(self, value: t.Any) -> str:
"""Tag the value and dump it to a compact JSON string."""
- pass
+ return dumps(self.tag(value), separators=(",", ":"))
- def loads(self, value: str) ->t.Any:
+ def loads(self, value: str) -> t.Any:
"""Load data from a JSON string and deserialized any tagged objects."""
- pass
+ return self._untag_scan(loads(value))
diff --git a/src/flask/logging.py b/src/flask/logging.py
index 6fe6f650..0cb8f437 100644
--- a/src/flask/logging.py
+++ b/src/flask/logging.py
@@ -1,15 +1,19 @@
from __future__ import annotations
+
import logging
import sys
import typing as t
+
from werkzeug.local import LocalProxy
+
from .globals import request
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from .sansio.app import App
@LocalProxy
-def wsgi_errors_stream() ->t.TextIO:
+def wsgi_errors_stream() -> t.TextIO:
"""Find the most appropriate error stream for the application. If a request
is active, log to ``wsgi.errors``, otherwise use ``sys.stderr``.
@@ -18,22 +22,40 @@ def wsgi_errors_stream() ->t.TextIO:
can't import this directly, you can refer to it as
``ext://flask.logging.wsgi_errors_stream``.
"""
- pass
+ if request:
+ return request.environ["wsgi.errors"] # type: ignore[no-any-return]
+
+ return sys.stderr
-def has_level_handler(logger: logging.Logger) ->bool:
+def has_level_handler(logger: logging.Logger) -> bool:
"""Check if there is a handler in the logging chain that will handle the
given logger's :meth:`effective level <~logging.Logger.getEffectiveLevel>`.
"""
- pass
+ level = logger.getEffectiveLevel()
+ current = logger
+
+ while current:
+ if any(handler.level <= level for handler in current.handlers):
+ return True
+ if not current.propagate:
+ break
-default_handler = logging.StreamHandler(wsgi_errors_stream)
-default_handler.setFormatter(logging.Formatter(
- '[%(asctime)s] %(levelname)s in %(module)s: %(message)s'))
+ current = current.parent # type: ignore
+ return False
-def create_logger(app: App) ->logging.Logger:
+
+#: Log messages to :func:`~flask.logging.wsgi_errors_stream` with the format
+#: ``[%(asctime)s] %(levelname)s in %(module)s: %(message)s``.
+default_handler = logging.StreamHandler(wsgi_errors_stream) # type: ignore
+default_handler.setFormatter(
+ logging.Formatter("[%(asctime)s] %(levelname)s in %(module)s: %(message)s")
+)
+
+
+def create_logger(app: App) -> logging.Logger:
"""Get the Flask app's logger and configure it if needed.
The logger name will be the same as
@@ -46,4 +68,12 @@ def create_logger(app: App) ->logging.Logger:
:class:`~logging.StreamHandler` for
:func:`~flask.logging.wsgi_errors_stream` with a basic format.
"""
- pass
+ logger = logging.getLogger(app.name)
+
+ if app.debug and not logger.level:
+ logger.setLevel(logging.DEBUG)
+
+ if not has_level_handler(logger):
+ logger.addHandler(default_handler)
+
+ return logger
diff --git a/src/flask/sansio/app.py b/src/flask/sansio/app.py
index 2424fb7f..01fd5dbf 100644
--- a/src/flask/sansio/app.py
+++ b/src/flask/sansio/app.py
@@ -1,10 +1,12 @@
from __future__ import annotations
+
import logging
import os
import sys
import typing as t
from datetime import timedelta
from itertools import chain
+
from werkzeug.exceptions import Aborter
from werkzeug.exceptions import BadRequest
from werkzeug.exceptions import BadRequestKeyError
@@ -14,6 +16,7 @@ from werkzeug.routing import Rule
from werkzeug.sansio.response import Response
from werkzeug.utils import cached_property
from werkzeug.utils import redirect as _wz_redirect
+
from .. import typing as ft
from ..config import Config
from ..config import ConfigAttribute
@@ -29,19 +32,28 @@ from .scaffold import _endpoint_from_view_func
from .scaffold import find_package
from .scaffold import Scaffold
from .scaffold import setupmethod
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from werkzeug.wrappers import Response as BaseResponse
+
from ..testing import FlaskClient
from ..testing import FlaskCliRunner
from .blueprints import Blueprint
-T_shell_context_processor = t.TypeVar('T_shell_context_processor', bound=ft
- .ShellContextProcessorCallable)
-T_teardown = t.TypeVar('T_teardown', bound=ft.TeardownCallable)
-T_template_filter = t.TypeVar('T_template_filter', bound=ft.
- TemplateFilterCallable)
-T_template_global = t.TypeVar('T_template_global', bound=ft.
- TemplateGlobalCallable)
-T_template_test = t.TypeVar('T_template_test', bound=ft.TemplateTestCallable)
+
+T_shell_context_processor = t.TypeVar(
+ "T_shell_context_processor", bound=ft.ShellContextProcessorCallable
+)
+T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable)
+T_template_filter = t.TypeVar("T_template_filter", bound=ft.TemplateFilterCallable)
+T_template_global = t.TypeVar("T_template_global", bound=ft.TemplateGlobalCallable)
+T_template_test = t.TypeVar("T_template_test", bound=ft.TemplateTestCallable)
+
+
+def _make_timedelta(value: timedelta | int | None) -> timedelta | None:
+ if value is None or isinstance(value, timedelta):
+ return value
+
+ return timedelta(seconds=value)
class App(Scaffold):
@@ -140,14 +152,81 @@ class App(Scaffold):
This should only be set manually when it can't be detected
automatically, such as for namespace packages.
"""
+
+ #: The class of the object assigned to :attr:`aborter`, created by
+ #: :meth:`create_aborter`. That object is called by
+ #: :func:`flask.abort` to raise HTTP errors, and can be
+ #: called directly as well.
+ #:
+ #: Defaults to :class:`werkzeug.exceptions.Aborter`.
+ #:
+ #: .. versionadded:: 2.2
aborter_class = Aborter
+
+ #: The class that is used for the Jinja environment.
+ #:
+ #: .. versionadded:: 0.11
jinja_environment = Environment
+
+ #: The class that is used for the :data:`~flask.g` instance.
+ #:
+ #: Example use cases for a custom class:
+ #:
+ #: 1. Store arbitrary attributes on flask.g.
+ #: 2. Add a property for lazy per-request database connectors.
+ #: 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
app_ctx_globals_class = _AppCtxGlobals
+
+ #: The class that is used for the ``config`` attribute of this app.
+ #: Defaults to :class:`~flask.Config`.
+ #:
+ #: Example use cases for a custom class:
+ #:
+ #: 1. Default values for certain config options.
+ #: 2. Access to config values through attributes in addition to keys.
+ #:
+ #: .. versionadded:: 0.11
config_class = Config
- testing = ConfigAttribute[bool]('TESTING')
- secret_key = ConfigAttribute[t.Union[str, bytes, None]]('SECRET_KEY')
+
+ #: The testing flag. Set this to ``True`` to enable the test mode of
+ #: Flask extensions (and in the future probably also Flask itself).
+ #: For example this might activate test helpers that have an
+ #: additional runtime cost which should not be enabled by default.
+ #:
+ #: If this is enabled and PROPAGATE_EXCEPTIONS is not changed from the
+ #: default it's implicitly enabled.
+ #:
+ #: This attribute can also be configured from the config with the
+ #: ``TESTING`` configuration key. Defaults to ``False``.
+ testing = ConfigAttribute[bool]("TESTING")
+
+ #: If a secret key is set, cryptographic components can use this to
+ #: sign cookies and other things. Set this to a complex random value
+ #: when you want to use the secure cookie for instance.
+ #:
+ #: This attribute can also be configured from the config with the
+ #: :data:`SECRET_KEY` configuration key. Defaults to ``None``.
+ secret_key = ConfigAttribute[t.Union[str, bytes, None]]("SECRET_KEY")
+
+ #: A :class:`~datetime.timedelta` which is used to set the expiration
+ #: date of a permanent session. The default is 31 days which makes a
+ #: permanent session survive for roughly one month.
+ #:
+ #: This attribute can also be configured from the config with the
+ #: ``PERMANENT_SESSION_LIFETIME`` configuration key. Defaults to
+ #: ``timedelta(days=31)``
permanent_session_lifetime = ConfigAttribute[timedelta](
- 'PERMANENT_SESSION_LIFETIME', get_converter=_make_timedelta)
+ "PERMANENT_SESSION_LIFETIME",
+ get_converter=_make_timedelta, # type: ignore[arg-type]
+ )
+
json_provider_class: type[JSONProvider] = DefaultJSONProvider
"""A subclass of :class:`~flask.json.provider.JSONProvider`. An
instance is created and assigned to :attr:`app.json` when creating
@@ -159,32 +238,94 @@ class App(Scaffold):
.. versionadded:: 2.2
"""
+
+ #: Options that are passed to the Jinja environment in
+ #: :meth:`create_jinja_environment`. Changing these options after
+ #: the environment is created (accessing :attr:`jinja_env`) will
+ #: have no effect.
+ #:
+ #: .. versionchanged:: 1.1.0
+ #: This is a ``dict`` instead of an ``ImmutableDict`` to allow
+ #: easier configuration.
+ #:
jinja_options: dict[str, t.Any] = {}
+
+ #: The rule object to use for URL rules created. This is used by
+ #: :meth:`add_url_rule`. Defaults to :class:`werkzeug.routing.Rule`.
+ #:
+ #: .. versionadded:: 0.7
url_rule_class = Rule
+
+ #: The map object to use for storing the URL rules and routing
+ #: configuration parameters. Defaults to :class:`werkzeug.routing.Map`.
+ #:
+ #: .. versionadded:: 1.1.0
url_map_class = Map
+
+ #: The :meth:`test_client` method creates an instance of this test
+ #: client class. Defaults to :class:`~flask.testing.FlaskClient`.
+ #:
+ #: .. versionadded:: 0.7
test_client_class: type[FlaskClient] | None = None
+
+ #: The :class:`~click.testing.CliRunner` subclass, by default
+ #: :class:`~flask.testing.FlaskCliRunner` that is used by
+ #: :meth:`test_cli_runner`. Its ``__init__`` method should take a
+ #: Flask app object as the first argument.
+ #:
+ #: .. versionadded:: 1.0
test_cli_runner_class: type[FlaskCliRunner] | None = None
+
default_config: dict[str, t.Any]
response_class: type[Response]
- def __init__(self, import_name: str, static_url_path: (str | None)=None,
- static_folder: (str | os.PathLike[str] | None)='static',
- static_host: (str | None)=None, host_matching: bool=False,
- subdomain_matching: bool=False, template_folder: (str | os.PathLike
- [str] | None)='templates', instance_path: (str | None)=None,
- instance_relative_config: bool=False, root_path: (str | None)=None):
- super().__init__(import_name=import_name, static_folder=
- static_folder, static_url_path=static_url_path, template_folder
- =template_folder, root_path=root_path)
+ def __init__(
+ self,
+ import_name: str,
+ static_url_path: str | None = None,
+ static_folder: str | os.PathLike[str] | None = "static",
+ static_host: str | None = None,
+ host_matching: bool = False,
+ subdomain_matching: bool = False,
+ template_folder: str | os.PathLike[str] | None = "templates",
+ instance_path: str | None = None,
+ instance_relative_config: bool = False,
+ root_path: str | None = None,
+ ):
+ super().__init__(
+ import_name=import_name,
+ static_folder=static_folder,
+ static_url_path=static_url_path,
+ template_folder=template_folder,
+ root_path=root_path,
+ )
+
if instance_path is None:
instance_path = self.auto_find_instance_path()
elif not os.path.isabs(instance_path):
raise ValueError(
- 'If an instance path is provided it must be absolute. A relative path was given instead.'
- )
+ "If an instance path is provided it must be absolute."
+ " A relative path was given instead."
+ )
+
+ #: Holds the path to the instance folder.
+ #:
+ #: .. versionadded:: 0.8
self.instance_path = instance_path
+
+ #: The configuration dictionary as :class:`Config`. This behaves
+ #: exactly like a regular dictionary but supports additional methods
+ #: to load a config from files.
self.config = self.make_config(instance_relative_config)
+
+ #: An instance of :attr:`aborter_class` created by
+ #: :meth:`make_aborter`. This is called by :func:`flask.abort`
+ #: to raise HTTP errors, and can be called directly as well.
+ #:
+ #: .. versionadded:: 2.2
+ #: Moved from ``flask.abort``, which calls this object.
self.aborter = self.make_aborter()
+
self.json: JSONProvider = self.json_provider_class(self)
"""Provides access to JSON methods. Functions in ``flask.json``
will call methods on this provider when the application context
@@ -200,19 +341,89 @@ class App(Scaffold):
.. versionadded:: 2.2
"""
- self.url_build_error_handlers: list[t.Callable[[Exception, str,
- dict[str, t.Any]], str]] = []
+
+ #: A list of functions that are called by
+ #: :meth:`handle_url_build_error` when :meth:`.url_for` raises a
+ #: :exc:`~werkzeug.routing.BuildError`. Each function is called
+ #: with ``error``, ``endpoint`` and ``values``. If a function
+ #: returns ``None`` or raises a ``BuildError``, it is skipped.
+ #: Otherwise, its return value is returned by ``url_for``.
+ #:
+ #: .. versionadded:: 0.9
+ self.url_build_error_handlers: list[
+ t.Callable[[Exception, str, dict[str, t.Any]], str]
+ ] = []
+
+ #: A list of functions that are called when the application context
+ #: is destroyed. Since the application context is also torn down
+ #: if the request ends this is the place to store code that disconnects
+ #: from databases.
+ #:
+ #: .. versionadded:: 0.9
self.teardown_appcontext_funcs: list[ft.TeardownCallable] = []
- self.shell_context_processors: list[ft.ShellContextProcessorCallable
- ] = []
+
+ #: A list of shell context processor functions that should be run
+ #: when a shell context is created.
+ #:
+ #: .. versionadded:: 0.11
+ self.shell_context_processors: list[ft.ShellContextProcessorCallable] = []
+
+ #: Maps registered blueprint names to blueprint objects. The
+ #: dict retains the order the blueprints were registered in.
+ #: Blueprints can be registered multiple times, this dict does
+ #: not track how often they were attached.
+ #:
+ #: .. versionadded:: 0.7
self.blueprints: dict[str, Blueprint] = {}
+
+ #: a place where extensions can store application specific state. For
+ #: example this is where an extension could store database engines and
+ #: similar things.
+ #:
+ #: The key must match the name of the extension module. For example in
+ #: case of a "Flask-Foo" extension in `flask_foo`, the key would be
+ #: ``'foo'``.
+ #:
+ #: .. versionadded:: 0.7
self.extensions: dict[str, t.Any] = {}
+
+ #: The :class:`~werkzeug.routing.Map` for this instance. You can use
+ #: this to change the routing converters after the class was created
+ #: but before any routes are connected. Example::
+ #:
+ #: from werkzeug.routing import BaseConverter
+ #:
+ #: class ListConverter(BaseConverter):
+ #: def to_python(self, value):
+ #: return value.split(',')
+ #: def to_url(self, values):
+ #: return ','.join(super(ListConverter, self).to_url(value)
+ #: for value in values)
+ #:
+ #: app = Flask(__name__)
+ #: app.url_map.converters['list'] = ListConverter
self.url_map = self.url_map_class(host_matching=host_matching)
+
self.subdomain_matching = subdomain_matching
+
+ # tracks internally if the application already handled at least one
+ # request.
self._got_first_request = False
+ def _check_setup_finished(self, f_name: str) -> None:
+ if self._got_first_request:
+ raise AssertionError(
+ f"The setup method '{f_name}' can no longer be called"
+ " on the application. It has already handled its first"
+ " request, any changes will not be applied"
+ " consistently.\n"
+ "Make sure all imports, decorators, functions, etc."
+ " needed to set up the application are done before"
+ " running it."
+ )
+
@cached_property
- def name(self) ->str:
+ def name(self) -> str: # type: ignore
"""The name of the application. This is usually the import name
with the difference that it's guessed from the run file if the
import name is main. This name is used as a display name when
@@ -221,10 +432,15 @@ class App(Scaffold):
.. versionadded:: 0.8
"""
- pass
+ if self.import_name == "__main__":
+ fn: str | None = getattr(sys.modules["__main__"], "__file__", None)
+ if fn is None:
+ return "__main__"
+ return os.path.splitext(os.path.basename(fn))[0]
+ return self.import_name
@cached_property
- def logger(self) ->logging.Logger:
+ def logger(self) -> logging.Logger:
"""A standard Python :class:`~logging.Logger` for the app, with
the same name as :attr:`name`.
@@ -248,19 +464,22 @@ class App(Scaffold):
.. versionadded:: 0.3
"""
- pass
+ return create_logger(self)
@cached_property
- def jinja_env(self) ->Environment:
+ def jinja_env(self) -> Environment:
"""The Jinja environment used to load templates.
The environment is created the first time this property is
accessed. Changing :attr:`jinja_options` after that will have no
effect.
"""
- pass
+ return self.create_jinja_environment()
+
+ def create_jinja_environment(self) -> Environment:
+ raise NotImplementedError()
- def make_config(self, instance_relative: bool=False) ->Config:
+ def make_config(self, instance_relative: bool = False) -> Config:
"""Used to create the config attribute by the Flask constructor.
The `instance_relative` parameter is passed in from the constructor
of Flask (there named `instance_relative_config`) and indicates if
@@ -269,9 +488,14 @@ class App(Scaffold):
.. versionadded:: 0.8
"""
- pass
-
- def make_aborter(self) ->Aborter:
+ root_path = self.root_path
+ if instance_relative:
+ root_path = self.instance_path
+ defaults = dict(self.default_config)
+ defaults["DEBUG"] = get_debug_flag()
+ return self.config_class(root_path, defaults)
+
+ def make_aborter(self) -> Aborter:
"""Create the object to assign to :attr:`aborter`. That object
is called by :func:`flask.abort` to raise HTTP errors, and can
be called directly as well.
@@ -281,9 +505,9 @@ class App(Scaffold):
.. versionadded:: 2.2
"""
- pass
+ return self.aborter_class()
- def auto_find_instance_path(self) ->str:
+ def auto_find_instance_path(self) -> str:
"""Tries to locate the instance path if it was not provided to the
constructor of the application class. It will basically calculate
the path to a folder named ``instance`` next to your main file or
@@ -291,9 +515,12 @@ class App(Scaffold):
.. versionadded:: 0.8
"""
- pass
+ prefix, package_path = find_package(self.import_name)
+ if prefix is None:
+ return os.path.join(package_path, "instance")
+ return os.path.join(prefix, "var", f"{self.name}-instance")
- def create_global_jinja_loader(self) ->DispatchingJinjaLoader:
+ def create_global_jinja_loader(self) -> DispatchingJinjaLoader:
"""Creates the loader for the Jinja2 environment. Can be used to
override just the loader and keeping the rest unchanged. It's
discouraged to override this function. Instead one should override
@@ -304,9 +531,9 @@ class App(Scaffold):
.. versionadded:: 0.7
"""
- pass
+ return DispatchingJinjaLoader(self)
- def select_jinja_autoescape(self, filename: str) ->bool:
+ def select_jinja_autoescape(self, filename: str) -> bool:
"""Returns ``True`` if autoescaping should be active for the given
template name. If no template name is given, returns `True`.
@@ -315,10 +542,12 @@ class App(Scaffold):
.. versionadded:: 0.5
"""
- pass
+ if filename is None:
+ return True
+ return filename.endswith((".html", ".htm", ".xml", ".xhtml", ".svg"))
@property
- def debug(self) ->bool:
+ def debug(self) -> bool:
"""Whether debug mode is enabled. When using ``flask run`` to start the
development server, an interactive debugger will be shown for unhandled
exceptions, and the server will be reloaded when code changes. This maps to the
@@ -328,11 +557,17 @@ class App(Scaffold):
Default: ``False``
"""
- pass
+ return self.config["DEBUG"] # type: ignore[no-any-return]
+
+ @debug.setter
+ def debug(self, value: bool) -> None:
+ self.config["DEBUG"] = value
+
+ if self.config["TEMPLATES_AUTO_RELOAD"] is None:
+ self.jinja_env.auto_reload = value
@setupmethod
- def register_blueprint(self, blueprint: Blueprint, **options: t.Any
- ) ->None:
+ def register_blueprint(self, blueprint: Blueprint, **options: t.Any) -> None:
"""Register a :class:`~flask.Blueprint` on the application. Keyword
arguments passed to this method will override the defaults set on the
blueprint.
@@ -357,18 +592,78 @@ class App(Scaffold):
.. versionadded:: 0.7
"""
- pass
+ blueprint.register(self, options)
- def iter_blueprints(self) ->t.ValuesView[Blueprint]:
+ def iter_blueprints(self) -> t.ValuesView[Blueprint]:
"""Iterates over all blueprints by the order they were registered.
.. versionadded:: 0.11
"""
- pass
+ return self.blueprints.values()
@setupmethod
- def template_filter(self, name: (str | None)=None) ->t.Callable[[
- T_template_filter], T_template_filter]:
+ def add_url_rule(
+ self,
+ rule: str,
+ endpoint: str | None = None,
+ view_func: ft.RouteCallable | None = None,
+ provide_automatic_options: bool | None = None,
+ **options: t.Any,
+ ) -> None:
+ if endpoint is None:
+ endpoint = _endpoint_from_view_func(view_func) # type: ignore
+ options["endpoint"] = endpoint
+ methods = options.pop("methods", None)
+
+ # if the methods are not given and the view_func object knows its
+ # methods we can use that instead. If neither exists, we go with
+ # a tuple of only ``GET`` as default.
+ if methods is None:
+ methods = getattr(view_func, "methods", None) or ("GET",)
+ if isinstance(methods, str):
+ raise TypeError(
+ "Allowed methods must be a list of strings, for"
+ ' example: @app.route(..., methods=["POST"])'
+ )
+ methods = {item.upper() for item in methods}
+
+ # Methods that should always be added
+ required_methods = set(getattr(view_func, "required_methods", ()))
+
+ # starting with Flask 0.8 the view_func object can disable and
+ # force-enable the automatic options handling.
+ if provide_automatic_options is None:
+ provide_automatic_options = getattr(
+ view_func, "provide_automatic_options", None
+ )
+
+ if provide_automatic_options is None:
+ if "OPTIONS" not in methods:
+ provide_automatic_options = True
+ required_methods.add("OPTIONS")
+ else:
+ provide_automatic_options = False
+
+ # Add the required methods now.
+ methods |= required_methods
+
+ rule_obj = self.url_rule_class(rule, methods=methods, **options)
+ rule_obj.provide_automatic_options = provide_automatic_options # type: ignore[attr-defined]
+
+ self.url_map.add(rule_obj)
+ if view_func is not None:
+ old_func = self.view_functions.get(endpoint)
+ if old_func is not None and old_func != view_func:
+ raise AssertionError(
+ "View function mapping is overwriting an existing"
+ f" endpoint function: {endpoint}"
+ )
+ self.view_functions[endpoint] = view_func
+
+ @setupmethod
+ def template_filter(
+ self, name: str | None = None
+ ) -> t.Callable[[T_template_filter], T_template_filter]:
"""A decorator that is used to register custom template filter.
You can specify a name for the filter, otherwise the function
name will be used. Example::
@@ -380,22 +675,29 @@ class App(Scaffold):
:param name: the optional name of the filter, otherwise the
function name will be used.
"""
- pass
+
+ def decorator(f: T_template_filter) -> T_template_filter:
+ self.add_template_filter(f, name=name)
+ return f
+
+ return decorator
@setupmethod
- def add_template_filter(self, f: ft.TemplateFilterCallable, name: (str |
- None)=None) ->None:
+ def add_template_filter(
+ self, f: ft.TemplateFilterCallable, name: str | None = None
+ ) -> None:
"""Register a custom template filter. Works exactly like the
:meth:`template_filter` decorator.
:param name: the optional name of the filter, otherwise the
function name will be used.
"""
- pass
+ self.jinja_env.filters[name or f.__name__] = f
@setupmethod
- def template_test(self, name: (str | None)=None) ->t.Callable[[
- T_template_test], T_template_test]:
+ def template_test(
+ self, name: str | None = None
+ ) -> t.Callable[[T_template_test], T_template_test]:
"""A decorator that is used to register custom template test.
You can specify a name for the test, otherwise the function
name will be used. Example::
@@ -414,11 +716,17 @@ class App(Scaffold):
:param name: the optional name of the test, otherwise the
function name will be used.
"""
- pass
+
+ def decorator(f: T_template_test) -> T_template_test:
+ self.add_template_test(f, name=name)
+ return f
+
+ return decorator
@setupmethod
- def add_template_test(self, f: ft.TemplateTestCallable, name: (str |
- None)=None) ->None:
+ def add_template_test(
+ self, f: ft.TemplateTestCallable, name: str | None = None
+ ) -> None:
"""Register a custom template test. Works exactly like the
:meth:`template_test` decorator.
@@ -427,11 +735,12 @@ class App(Scaffold):
:param name: the optional name of the test, otherwise the
function name will be used.
"""
- pass
+ self.jinja_env.tests[name or f.__name__] = f
@setupmethod
- def template_global(self, name: (str | None)=None) ->t.Callable[[
- T_template_global], T_template_global]:
+ def template_global(
+ self, name: str | None = None
+ ) -> t.Callable[[T_template_global], T_template_global]:
"""A decorator that is used to register a custom template global function.
You can specify a name for the global function, otherwise the function
name will be used. Example::
@@ -445,11 +754,17 @@ class App(Scaffold):
:param name: the optional name of the global function, otherwise the
function name will be used.
"""
- pass
+
+ def decorator(f: T_template_global) -> T_template_global:
+ self.add_template_global(f, name=name)
+ return f
+
+ return decorator
@setupmethod
- def add_template_global(self, f: ft.TemplateGlobalCallable, name: (str |
- None)=None) ->None:
+ def add_template_global(
+ self, f: ft.TemplateGlobalCallable, name: str | None = None
+ ) -> None:
"""Register a custom template global function. Works exactly like the
:meth:`template_global` decorator.
@@ -458,10 +773,10 @@ class App(Scaffold):
:param name: the optional name of the global function, otherwise the
function name will be used.
"""
- pass
+ self.jinja_env.globals[name or f.__name__] = f
@setupmethod
- def teardown_appcontext(self, f: T_teardown) ->T_teardown:
+ 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
@@ -491,27 +806,46 @@ class App(Scaffold):
.. versionadded:: 0.9
"""
- pass
+ self.teardown_appcontext_funcs.append(f)
+ return f
@setupmethod
- def shell_context_processor(self, f: T_shell_context_processor
- ) ->T_shell_context_processor:
+ def shell_context_processor(
+ self, f: T_shell_context_processor
+ ) -> T_shell_context_processor:
"""Registers a shell context processor function.
.. versionadded:: 0.11
"""
- pass
+ self.shell_context_processors.append(f)
+ return f
- def _find_error_handler(self, e: Exception, blueprints: list[str]) ->(ft
- .ErrorHandlerCallable | None):
+ def _find_error_handler(
+ self, e: Exception, blueprints: list[str]
+ ) -> ft.ErrorHandlerCallable | None:
"""Return a registered error handler for an exception in this order:
blueprint handler for a specific code, app handler for a specific code,
blueprint handler for an exception class, app handler for an exception
class, or ``None`` if a suitable handler is not found.
"""
- pass
+ exc_class, code = self._get_exc_class_and_code(type(e))
+ names = (*blueprints, None)
+
+ for c in (code, None) if code is not None else (None,):
+ for name in names:
+ handler_map = self.error_handler_spec[name][c]
+
+ if not handler_map:
+ continue
- def trap_http_exception(self, e: Exception) ->bool:
+ for cls in exc_class.__mro__:
+ handler = handler_map.get(cls)
+
+ if handler is not None:
+ return handler
+ return None
+
+ def trap_http_exception(self, e: Exception) -> bool:
"""Checks if an HTTP exception should be trapped or not. By default
this will return ``False`` for all exceptions except for a bad request
key error if ``TRAP_BAD_REQUEST_ERRORS`` is set to ``True``. It
@@ -528,9 +862,25 @@ class App(Scaffold):
.. versionadded:: 0.8
"""
- pass
+ if self.config["TRAP_HTTP_EXCEPTIONS"]:
+ return True
+
+ trap_bad_request = self.config["TRAP_BAD_REQUEST_ERRORS"]
+
+ # if unset, trap key errors in debug mode
+ if (
+ trap_bad_request is None
+ and self.debug
+ and isinstance(e, BadRequestKeyError)
+ ):
+ return True
- def should_ignore_error(self, error: (BaseException | None)) ->bool:
+ if trap_bad_request:
+ return isinstance(e, BadRequest)
+
+ return False
+
+ def should_ignore_error(self, error: BaseException | None) -> bool:
"""This is called to figure out if an error should be ignored
or not as far as the teardown system is concerned. If this
function returns ``True`` then the teardown handlers will not be
@@ -538,9 +888,9 @@ class App(Scaffold):
.. versionadded:: 0.10
"""
- pass
+ return False
- def redirect(self, location: str, code: int=302) ->BaseResponse:
+ def redirect(self, location: str, code: int = 302) -> BaseResponse:
"""Create a redirect response object.
This is called by :func:`flask.redirect`, and can be called
@@ -552,20 +902,36 @@ class App(Scaffold):
.. versionadded:: 2.2
Moved from ``flask.redirect``, which calls this method.
"""
- pass
+ return _wz_redirect(
+ location,
+ code=code,
+ Response=self.response_class, # type: ignore[arg-type]
+ )
- def inject_url_defaults(self, endpoint: str, values: dict[str, t.Any]
- ) ->None:
+ def inject_url_defaults(self, endpoint: str, values: dict[str, t.Any]) -> None:
"""Injects the URL defaults for the given endpoint directly into
the values dictionary passed. This is used internally and
automatically called on URL building.
.. versionadded:: 0.7
"""
- pass
-
- def handle_url_build_error(self, error: BuildError, endpoint: str,
- values: dict[str, t.Any]) ->str:
+ names: t.Iterable[str | None] = (None,)
+
+ # url_for may be called outside a request context, parse the
+ # passed endpoint instead of using request.blueprints.
+ if "." in endpoint:
+ names = chain(
+ names, reversed(_split_blueprint_path(endpoint.rpartition(".")[0]))
+ )
+
+ for name in names:
+ if name in self.url_default_functions:
+ for func in self.url_default_functions[name]:
+ func(endpoint, values)
+
+ def handle_url_build_error(
+ self, error: BuildError, endpoint: str, values: dict[str, t.Any]
+ ) -> str:
"""Called by :meth:`.url_for` if a
:exc:`~werkzeug.routing.BuildError` was raised. If this returns
a value, it will be returned by ``url_for``, otherwise the error
@@ -580,4 +946,19 @@ class App(Scaffold):
:param endpoint: The endpoint being built.
:param values: The keyword arguments passed to ``url_for``.
"""
- pass
+ for handler in self.url_build_error_handlers:
+ try:
+ rv = handler(error, endpoint, values)
+ except BuildError as e:
+ # make error available outside except block
+ error = e
+ else:
+ if rv is not None:
+ return rv
+
+ # Re-raise if called with an active exception, otherwise raise
+ # the passed in exception.
+ if error is sys.exc_info()[1]:
+ raise
+
+ raise error
diff --git a/src/flask/sansio/blueprints.py b/src/flask/sansio/blueprints.py
index bd3b9de9..4f912cca 100644
--- a/src/flask/sansio/blueprints.py
+++ b/src/flask/sansio/blueprints.py
@@ -1,32 +1,34 @@
from __future__ import annotations
+
import os
import typing as t
from collections import defaultdict
from functools import update_wrapper
+
from .. import typing as ft
from .scaffold import _endpoint_from_view_func
from .scaffold import _sentinel
from .scaffold import Scaffold
from .scaffold import setupmethod
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from .app import App
-DeferredSetupFunction = t.Callable[['BlueprintSetupState'], None]
-T_after_request = t.TypeVar('T_after_request', bound=ft.
- AfterRequestCallable[t.Any])
-T_before_request = t.TypeVar('T_before_request', bound=ft.BeforeRequestCallable
- )
-T_error_handler = t.TypeVar('T_error_handler', bound=ft.ErrorHandlerCallable)
-T_teardown = t.TypeVar('T_teardown', bound=ft.TeardownCallable)
-T_template_context_processor = t.TypeVar('T_template_context_processor',
- bound=ft.TemplateContextProcessorCallable)
-T_template_filter = t.TypeVar('T_template_filter', bound=ft.
- TemplateFilterCallable)
-T_template_global = t.TypeVar('T_template_global', bound=ft.
- TemplateGlobalCallable)
-T_template_test = t.TypeVar('T_template_test', bound=ft.TemplateTestCallable)
-T_url_defaults = t.TypeVar('T_url_defaults', bound=ft.URLDefaultCallable)
-T_url_value_preprocessor = t.TypeVar('T_url_value_preprocessor', bound=ft.
- URLValuePreprocessorCallable)
+
+DeferredSetupFunction = t.Callable[["BlueprintSetupState"], None]
+T_after_request = t.TypeVar("T_after_request", bound=ft.AfterRequestCallable[t.Any])
+T_before_request = t.TypeVar("T_before_request", bound=ft.BeforeRequestCallable)
+T_error_handler = t.TypeVar("T_error_handler", bound=ft.ErrorHandlerCallable)
+T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable)
+T_template_context_processor = t.TypeVar(
+ "T_template_context_processor", bound=ft.TemplateContextProcessorCallable
+)
+T_template_filter = t.TypeVar("T_template_filter", bound=ft.TemplateFilterCallable)
+T_template_global = t.TypeVar("T_template_global", bound=ft.TemplateGlobalCallable)
+T_template_test = t.TypeVar("T_template_test", bound=ft.TemplateTestCallable)
+T_url_defaults = t.TypeVar("T_url_defaults", bound=ft.URLDefaultCallable)
+T_url_value_preprocessor = t.TypeVar(
+ "T_url_value_preprocessor", bound=ft.URLValuePreprocessorCallable
+)
class BlueprintSetupState:
@@ -36,32 +38,82 @@ class BlueprintSetupState:
to all register callback functions.
"""
- def __init__(self, blueprint: Blueprint, app: App, options: t.Any,
- first_registration: bool) ->None:
+ def __init__(
+ self,
+ blueprint: Blueprint,
+ app: App,
+ options: t.Any,
+ first_registration: bool,
+ ) -> None:
+ #: a reference to the current application
self.app = app
+
+ #: a reference to the blueprint that created this setup state.
self.blueprint = blueprint
+
+ #: a dictionary with all options that were passed to the
+ #: :meth:`~flask.Flask.register_blueprint` method.
self.options = options
+
+ #: as blueprints can be registered multiple times with the
+ #: application and not everything wants to be registered
+ #: multiple times on it, this attribute can be used to figure
+ #: out if the blueprint was registered in the past already.
self.first_registration = first_registration
- subdomain = self.options.get('subdomain')
+
+ subdomain = self.options.get("subdomain")
if subdomain is None:
subdomain = self.blueprint.subdomain
+
+ #: The subdomain that the blueprint should be active for, ``None``
+ #: otherwise.
self.subdomain = subdomain
- url_prefix = self.options.get('url_prefix')
+
+ url_prefix = self.options.get("url_prefix")
if url_prefix is None:
url_prefix = self.blueprint.url_prefix
+ #: The prefix that should be used for all URLs defined on the
+ #: blueprint.
self.url_prefix = url_prefix
- self.name = self.options.get('name', blueprint.name)
- self.name_prefix = self.options.get('name_prefix', '')
- self.url_defaults = dict(self.blueprint.url_values_defaults)
- self.url_defaults.update(self.options.get('url_defaults', ()))
- def add_url_rule(self, rule: str, endpoint: (str | None)=None,
- view_func: (ft.RouteCallable | None)=None, **options: t.Any) ->None:
+ self.name = self.options.get("name", blueprint.name)
+ self.name_prefix = self.options.get("name_prefix", "")
+
+ #: A dictionary with URL defaults that is added to each and every
+ #: URL that was defined with the blueprint.
+ self.url_defaults = dict(self.blueprint.url_values_defaults)
+ self.url_defaults.update(self.options.get("url_defaults", ()))
+
+ def add_url_rule(
+ self,
+ rule: str,
+ endpoint: str | None = None,
+ view_func: ft.RouteCallable | None = None,
+ **options: t.Any,
+ ) -> None:
"""A helper method to register a rule (and optionally a view function)
to the application. The endpoint is automatically prefixed with the
blueprint's name.
"""
- pass
+ if self.url_prefix is not None:
+ if rule:
+ rule = "/".join((self.url_prefix.rstrip("/"), rule.lstrip("/")))
+ else:
+ rule = self.url_prefix
+ options.setdefault("subdomain", self.subdomain)
+ if endpoint is None:
+ endpoint = _endpoint_from_view_func(view_func) # type: ignore
+ defaults = self.url_defaults
+ if "defaults" in options:
+ defaults = dict(defaults, **options.pop("defaults"))
+
+ self.app.add_url_rule(
+ rule,
+ f"{self.name_prefix}.{self.name}.{endpoint}".lstrip("."),
+ view_func,
+ defaults=defaults,
+ **options,
+ )
class Blueprint(Scaffold):
@@ -116,60 +168,92 @@ class Blueprint(Scaffold):
.. versionadded:: 0.7
"""
+
_got_registered_once = False
- def __init__(self, name: str, import_name: str, static_folder: (str |
- os.PathLike[str] | None)=None, static_url_path: (str | None)=None,
- template_folder: (str | os.PathLike[str] | None)=None, url_prefix:
- (str | None)=None, subdomain: (str | None)=None, url_defaults: (
- dict[str, t.Any] | None)=None, root_path: (str | None)=None,
- cli_group: (str | None)=_sentinel):
- super().__init__(import_name=import_name, static_folder=
- static_folder, static_url_path=static_url_path, template_folder
- =template_folder, root_path=root_path)
+ def __init__(
+ self,
+ name: str,
+ import_name: str,
+ static_folder: str | os.PathLike[str] | None = None,
+ static_url_path: str | None = None,
+ template_folder: str | os.PathLike[str] | None = None,
+ url_prefix: str | None = None,
+ subdomain: str | None = None,
+ url_defaults: dict[str, t.Any] | None = None,
+ root_path: str | None = None,
+ cli_group: str | None = _sentinel, # type: ignore[assignment]
+ ):
+ super().__init__(
+ import_name=import_name,
+ static_folder=static_folder,
+ static_url_path=static_url_path,
+ template_folder=template_folder,
+ root_path=root_path,
+ )
+
if not name:
raise ValueError("'name' may not be empty.")
- if '.' in name:
+
+ if "." in name:
raise ValueError("'name' may not contain a dot '.' character.")
+
self.name = name
self.url_prefix = url_prefix
self.subdomain = subdomain
self.deferred_functions: list[DeferredSetupFunction] = []
+
if url_defaults is None:
url_defaults = {}
+
self.url_values_defaults = url_defaults
self.cli_group = cli_group
self._blueprints: list[tuple[Blueprint, dict[str, t.Any]]] = []
+ def _check_setup_finished(self, f_name: str) -> None:
+ if self._got_registered_once:
+ raise AssertionError(
+ f"The setup method '{f_name}' can no longer be called on the blueprint"
+ f" '{self.name}'. It has already been registered at least once, any"
+ " changes will not be applied consistently.\n"
+ "Make sure all imports, decorators, functions, etc. needed to set up"
+ " the blueprint are done before registering it."
+ )
+
@setupmethod
- def record(self, func: DeferredSetupFunction) ->None:
+ def record(self, func: DeferredSetupFunction) -> None:
"""Registers a function that is called when the blueprint is
registered on the application. This function is called with the
state as argument as returned by the :meth:`make_setup_state`
method.
"""
- pass
+ self.deferred_functions.append(func)
@setupmethod
- def record_once(self, func: DeferredSetupFunction) ->None:
+ def record_once(self, func: DeferredSetupFunction) -> None:
"""Works like :meth:`record` but wraps the function in another
function that will ensure the function is only called once. If the
blueprint is registered a second time on the application, the
function passed is not called.
"""
- pass
- def make_setup_state(self, app: App, options: dict[str, t.Any],
- first_registration: bool=False) ->BlueprintSetupState:
+ def wrapper(state: BlueprintSetupState) -> None:
+ if state.first_registration:
+ func(state)
+
+ self.record(update_wrapper(wrapper, func))
+
+ def make_setup_state(
+ self, app: App, options: dict[str, t.Any], first_registration: bool = False
+ ) -> BlueprintSetupState:
"""Creates an instance of :meth:`~flask.blueprints.BlueprintSetupState`
object that is later passed to the register callback functions.
Subclasses can override this to return a subclass of the setup state.
"""
- pass
+ return BlueprintSetupState(self, app, options, first_registration)
@setupmethod
- def register_blueprint(self, blueprint: Blueprint, **options: t.Any
- ) ->None:
+ def register_blueprint(self, blueprint: Blueprint, **options: t.Any) -> None:
"""Register a :class:`~flask.Blueprint` on this blueprint. Keyword
arguments passed to this method will override the defaults set
on the blueprint.
@@ -182,9 +266,11 @@ class Blueprint(Scaffold):
.. versionadded:: 2.0
"""
- pass
+ if blueprint is self:
+ raise ValueError("Cannot register a blueprint on itself")
+ self._blueprints.append((blueprint, options))
- def register(self, app: App, options: dict[str, t.Any]) ->None:
+ def register(self, app: App, options: dict[str, t.Any]) -> None:
"""Called by :meth:`Flask.register_blueprint` to register all
views and callbacks registered on the blueprint with the
application. Creates a :class:`.BlueprintSetupState` and calls
@@ -213,35 +299,168 @@ class Blueprint(Scaffold):
blueprint to be registered multiple times with unique names
for ``url_for``.
"""
- pass
+ name_prefix = options.get("name_prefix", "")
+ self_name = options.get("name", self.name)
+ name = f"{name_prefix}.{self_name}".lstrip(".")
+
+ if name in app.blueprints:
+ bp_desc = "this" if app.blueprints[name] is self else "a different"
+ existing_at = f" '{name}'" if self_name != name else ""
+
+ raise ValueError(
+ f"The name '{self_name}' is already registered for"
+ f" {bp_desc} blueprint{existing_at}. Use 'name=' to"
+ f" provide a unique name."
+ )
+
+ first_bp_registration = not any(bp is self for bp in app.blueprints.values())
+ first_name_registration = name not in app.blueprints
+
+ app.blueprints[name] = self
+ self._got_registered_once = True
+ state = self.make_setup_state(app, options, first_bp_registration)
+
+ if self.has_static_folder:
+ state.add_url_rule(
+ f"{self.static_url_path}/<path:filename>",
+ view_func=self.send_static_file, # type: ignore[attr-defined]
+ endpoint="static",
+ )
+
+ # Merge blueprint data into parent.
+ if first_bp_registration or first_name_registration:
+ self._merge_blueprint_funcs(app, name)
+
+ for deferred in self.deferred_functions:
+ deferred(state)
+
+ cli_resolved_group = options.get("cli_group", self.cli_group)
+
+ if self.cli.commands:
+ if cli_resolved_group is None:
+ app.cli.commands.update(self.cli.commands)
+ elif cli_resolved_group is _sentinel:
+ self.cli.name = name
+ app.cli.add_command(self.cli)
+ else:
+ self.cli.name = cli_resolved_group
+ app.cli.add_command(self.cli)
+
+ for blueprint, bp_options in self._blueprints:
+ bp_options = bp_options.copy()
+ bp_url_prefix = bp_options.get("url_prefix")
+ bp_subdomain = bp_options.get("subdomain")
+
+ if bp_subdomain is None:
+ bp_subdomain = blueprint.subdomain
+
+ if state.subdomain is not None and bp_subdomain is not None:
+ bp_options["subdomain"] = bp_subdomain + "." + state.subdomain
+ elif bp_subdomain is not None:
+ bp_options["subdomain"] = bp_subdomain
+ elif state.subdomain is not None:
+ bp_options["subdomain"] = state.subdomain
+
+ if bp_url_prefix is None:
+ bp_url_prefix = blueprint.url_prefix
+
+ if state.url_prefix is not None and bp_url_prefix is not None:
+ bp_options["url_prefix"] = (
+ state.url_prefix.rstrip("/") + "/" + bp_url_prefix.lstrip("/")
+ )
+ elif bp_url_prefix is not None:
+ bp_options["url_prefix"] = bp_url_prefix
+ elif state.url_prefix is not None:
+ bp_options["url_prefix"] = state.url_prefix
+
+ bp_options["name_prefix"] = name
+ blueprint.register(app, bp_options)
+
+ def _merge_blueprint_funcs(self, app: App, name: str) -> None:
+ def extend(
+ bp_dict: dict[ft.AppOrBlueprintKey, list[t.Any]],
+ parent_dict: dict[ft.AppOrBlueprintKey, list[t.Any]],
+ ) -> None:
+ for key, values in bp_dict.items():
+ key = name if key is None else f"{name}.{key}"
+ parent_dict[key].extend(values)
+
+ for key, value in self.error_handler_spec.items():
+ key = name if key is None else f"{name}.{key}"
+ value = defaultdict(
+ dict,
+ {
+ code: {exc_class: func for exc_class, func in code_values.items()}
+ for code, code_values in value.items()
+ },
+ )
+ app.error_handler_spec[key] = value
+
+ for endpoint, func in self.view_functions.items():
+ app.view_functions[endpoint] = func
+
+ extend(self.before_request_funcs, app.before_request_funcs)
+ extend(self.after_request_funcs, app.after_request_funcs)
+ extend(
+ self.teardown_request_funcs,
+ app.teardown_request_funcs,
+ )
+ extend(self.url_default_functions, app.url_default_functions)
+ extend(self.url_value_preprocessors, app.url_value_preprocessors)
+ extend(self.template_context_processors, app.template_context_processors)
@setupmethod
- def add_url_rule(self, rule: str, endpoint: (str | None)=None,
- view_func: (ft.RouteCallable | None)=None,
- provide_automatic_options: (bool | None)=None, **options: t.Any
- ) ->None:
+ def add_url_rule(
+ self,
+ rule: str,
+ endpoint: str | None = None,
+ view_func: ft.RouteCallable | None = None,
+ provide_automatic_options: bool | None = None,
+ **options: t.Any,
+ ) -> None:
"""Register a URL rule with the blueprint. See :meth:`.Flask.add_url_rule` for
full documentation.
The URL rule is prefixed with the blueprint's URL prefix. The endpoint name,
used with :func:`url_for`, is prefixed with the blueprint's name.
"""
- pass
+ if endpoint and "." in endpoint:
+ raise ValueError("'endpoint' may not contain a dot '.' character.")
+
+ if view_func and hasattr(view_func, "__name__") and "." in view_func.__name__:
+ raise ValueError("'view_func' name may not contain a dot '.' character.")
+
+ self.record(
+ lambda s: s.add_url_rule(
+ rule,
+ endpoint,
+ view_func,
+ provide_automatic_options=provide_automatic_options,
+ **options,
+ )
+ )
@setupmethod
- def app_template_filter(self, name: (str | None)=None) ->t.Callable[[
- T_template_filter], T_template_filter]:
+ def app_template_filter(
+ self, name: str | None = None
+ ) -> t.Callable[[T_template_filter], T_template_filter]:
"""Register a template filter, available in any template rendered by the
application. Equivalent to :meth:`.Flask.template_filter`.
:param name: the optional name of the filter, otherwise the
function name will be used.
"""
- pass
+
+ def decorator(f: T_template_filter) -> T_template_filter:
+ self.add_app_template_filter(f, name=name)
+ return f
+
+ return decorator
@setupmethod
- def add_app_template_filter(self, f: ft.TemplateFilterCallable, name: (
- str | None)=None) ->None:
+ def add_app_template_filter(
+ self, f: ft.TemplateFilterCallable, name: str | None = None
+ ) -> None:
"""Register a template filter, available in any template rendered by the
application. Works like the :meth:`app_template_filter` decorator. Equivalent to
:meth:`.Flask.add_template_filter`.
@@ -249,11 +468,16 @@ class Blueprint(Scaffold):
:param name: the optional name of the filter, otherwise the
function name will be used.
"""
- pass
+
+ def register_template(state: BlueprintSetupState) -> None:
+ state.app.jinja_env.filters[name or f.__name__] = f
+
+ self.record_once(register_template)
@setupmethod
- def app_template_test(self, name: (str | None)=None) ->t.Callable[[
- T_template_test], T_template_test]:
+ def app_template_test(
+ self, name: str | None = None
+ ) -> t.Callable[[T_template_test], T_template_test]:
"""Register a template test, available in any template rendered by the
application. Equivalent to :meth:`.Flask.template_test`.
@@ -262,11 +486,17 @@ class Blueprint(Scaffold):
:param name: the optional name of the test, otherwise the
function name will be used.
"""
- pass
+
+ def decorator(f: T_template_test) -> T_template_test:
+ self.add_app_template_test(f, name=name)
+ return f
+
+ return decorator
@setupmethod
- def add_app_template_test(self, f: ft.TemplateTestCallable, name: (str |
- None)=None) ->None:
+ def add_app_template_test(
+ self, f: ft.TemplateTestCallable, name: str | None = None
+ ) -> None:
"""Register a template test, available in any template rendered by the
application. Works like the :meth:`app_template_test` decorator. Equivalent to
:meth:`.Flask.add_template_test`.
@@ -276,11 +506,16 @@ class Blueprint(Scaffold):
:param name: the optional name of the test, otherwise the
function name will be used.
"""
- pass
+
+ def register_template(state: BlueprintSetupState) -> None:
+ state.app.jinja_env.tests[name or f.__name__] = f
+
+ self.record_once(register_template)
@setupmethod
- def app_template_global(self, name: (str | None)=None) ->t.Callable[[
- T_template_global], T_template_global]:
+ def app_template_global(
+ self, name: str | None = None
+ ) -> t.Callable[[T_template_global], T_template_global]:
"""Register a template global, available in any template rendered by the
application. Equivalent to :meth:`.Flask.template_global`.
@@ -289,11 +524,17 @@ class Blueprint(Scaffold):
:param name: the optional name of the global, otherwise the
function name will be used.
"""
- pass
+
+ def decorator(f: T_template_global) -> T_template_global:
+ self.add_app_template_global(f, name=name)
+ return f
+
+ return decorator
@setupmethod
- def add_app_template_global(self, f: ft.TemplateGlobalCallable, name: (
- str | None)=None) ->None:
+ def add_app_template_global(
+ self, f: ft.TemplateGlobalCallable, name: str | None = None
+ ) -> None:
"""Register a template global, available in any template rendered by the
application. Works like the :meth:`app_template_global` decorator. Equivalent to
:meth:`.Flask.add_template_global`.
@@ -303,56 +544,89 @@ class Blueprint(Scaffold):
:param name: the optional name of the global, otherwise the
function name will be used.
"""
- pass
+
+ def register_template(state: BlueprintSetupState) -> None:
+ state.app.jinja_env.globals[name or f.__name__] = f
+
+ self.record_once(register_template)
@setupmethod
- def before_app_request(self, f: T_before_request) ->T_before_request:
+ def before_app_request(self, f: T_before_request) -> T_before_request:
"""Like :meth:`before_request`, but before every request, not only those handled
by the blueprint. Equivalent to :meth:`.Flask.before_request`.
"""
- pass
+ self.record_once(
+ lambda s: s.app.before_request_funcs.setdefault(None, []).append(f)
+ )
+ return f
@setupmethod
- def after_app_request(self, f: T_after_request) ->T_after_request:
+ def after_app_request(self, f: T_after_request) -> T_after_request:
"""Like :meth:`after_request`, but after every request, not only those handled
by the blueprint. Equivalent to :meth:`.Flask.after_request`.
"""
- pass
+ self.record_once(
+ lambda s: s.app.after_request_funcs.setdefault(None, []).append(f)
+ )
+ return f
@setupmethod
- def teardown_app_request(self, f: T_teardown) ->T_teardown:
+ def teardown_app_request(self, f: T_teardown) -> T_teardown:
"""Like :meth:`teardown_request`, but after every request, not only those
handled by the blueprint. Equivalent to :meth:`.Flask.teardown_request`.
"""
- pass
+ self.record_once(
+ lambda s: s.app.teardown_request_funcs.setdefault(None, []).append(f)
+ )
+ return f
@setupmethod
- def app_context_processor(self, f: T_template_context_processor
- ) ->T_template_context_processor:
+ def app_context_processor(
+ self, f: T_template_context_processor
+ ) -> T_template_context_processor:
"""Like :meth:`context_processor`, but for templates rendered by every view, not
only by the blueprint. Equivalent to :meth:`.Flask.context_processor`.
"""
- pass
+ self.record_once(
+ lambda s: s.app.template_context_processors.setdefault(None, []).append(f)
+ )
+ return f
@setupmethod
- def app_errorhandler(self, code: (type[Exception] | int)) ->t.Callable[
- [T_error_handler], T_error_handler]:
+ def app_errorhandler(
+ self, code: type[Exception] | int
+ ) -> t.Callable[[T_error_handler], T_error_handler]:
"""Like :meth:`errorhandler`, but for every request, not only those handled by
the blueprint. Equivalent to :meth:`.Flask.errorhandler`.
"""
- pass
+
+ def decorator(f: T_error_handler) -> T_error_handler:
+ def from_blueprint(state: BlueprintSetupState) -> None:
+ state.app.errorhandler(code)(f)
+
+ self.record_once(from_blueprint)
+ return f
+
+ return decorator
@setupmethod
- def app_url_value_preprocessor(self, f: T_url_value_preprocessor
- ) ->T_url_value_preprocessor:
+ def app_url_value_preprocessor(
+ self, f: T_url_value_preprocessor
+ ) -> T_url_value_preprocessor:
"""Like :meth:`url_value_preprocessor`, but for every request, not only those
handled by the blueprint. Equivalent to :meth:`.Flask.url_value_preprocessor`.
"""
- pass
+ self.record_once(
+ lambda s: s.app.url_value_preprocessors.setdefault(None, []).append(f)
+ )
+ return f
@setupmethod
- def app_url_defaults(self, f: T_url_defaults) ->T_url_defaults:
+ def app_url_defaults(self, f: T_url_defaults) -> T_url_defaults:
"""Like :meth:`url_defaults`, but for every request, not only those handled by
the blueprint. Equivalent to :meth:`.Flask.url_defaults`.
"""
- pass
+ self.record_once(
+ lambda s: s.app.url_default_functions.setdefault(None, []).append(f)
+ )
+ return f
diff --git a/src/flask/sansio/scaffold.py b/src/flask/sansio/scaffold.py
index e35f461d..69e33a09 100644
--- a/src/flask/sansio/scaffold.py
+++ b/src/flask/sansio/scaffold.py
@@ -1,4 +1,5 @@
from __future__ import annotations
+
import importlib.util
import os
import pathlib
@@ -6,30 +7,46 @@ import sys
import typing as t
from collections import defaultdict
from functools import update_wrapper
+
from jinja2 import BaseLoader
from jinja2 import FileSystemLoader
from werkzeug.exceptions import default_exceptions
from werkzeug.exceptions import HTTPException
from werkzeug.utils import cached_property
+
from .. import typing as ft
from ..helpers import get_root_path
from ..templating import _default_template_ctx_processor
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from click import Group
+
+# a singleton sentinel value for parameter defaults
_sentinel = object()
-F = t.TypeVar('F', bound=t.Callable[..., t.Any])
-T_after_request = t.TypeVar('T_after_request', bound=ft.
- AfterRequestCallable[t.Any])
-T_before_request = t.TypeVar('T_before_request', bound=ft.BeforeRequestCallable
- )
-T_error_handler = t.TypeVar('T_error_handler', bound=ft.ErrorHandlerCallable)
-T_teardown = t.TypeVar('T_teardown', bound=ft.TeardownCallable)
-T_template_context_processor = t.TypeVar('T_template_context_processor',
- bound=ft.TemplateContextProcessorCallable)
-T_url_defaults = t.TypeVar('T_url_defaults', bound=ft.URLDefaultCallable)
-T_url_value_preprocessor = t.TypeVar('T_url_value_preprocessor', bound=ft.
- URLValuePreprocessorCallable)
-T_route = t.TypeVar('T_route', bound=ft.RouteCallable)
+
+F = t.TypeVar("F", bound=t.Callable[..., t.Any])
+T_after_request = t.TypeVar("T_after_request", bound=ft.AfterRequestCallable[t.Any])
+T_before_request = t.TypeVar("T_before_request", bound=ft.BeforeRequestCallable)
+T_error_handler = t.TypeVar("T_error_handler", bound=ft.ErrorHandlerCallable)
+T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable)
+T_template_context_processor = t.TypeVar(
+ "T_template_context_processor", bound=ft.TemplateContextProcessorCallable
+)
+T_url_defaults = t.TypeVar("T_url_defaults", bound=ft.URLDefaultCallable)
+T_url_value_preprocessor = t.TypeVar(
+ "T_url_value_preprocessor", bound=ft.URLValuePreprocessorCallable
+)
+T_route = t.TypeVar("T_route", bound=ft.RouteCallable)
+
+
+def setupmethod(f: F) -> F:
+ f_name = f.__name__
+
+ def wrapper_func(self: Scaffold, *args: t.Any, **kwargs: t.Any) -> t.Any:
+ self._check_setup_finished(f_name)
+ return f(self, *args, **kwargs)
+
+ return t.cast(F, update_wrapper(wrapper_func, f))
class Scaffold:
@@ -49,125 +66,274 @@ class Scaffold:
.. versionadded:: 2.0
"""
+
cli: Group
name: str
_static_folder: str | None = None
_static_url_path: str | None = None
- def __init__(self, import_name: str, static_folder: (str | os.PathLike[
- str] | None)=None, static_url_path: (str | None)=None,
- template_folder: (str | os.PathLike[str] | None)=None, root_path: (
- str | None)=None):
+ def __init__(
+ self,
+ import_name: str,
+ static_folder: str | os.PathLike[str] | None = None,
+ static_url_path: str | None = None,
+ template_folder: str | os.PathLike[str] | None = None,
+ root_path: str | None = None,
+ ):
+ #: The name of the package or module that this object belongs
+ #: to. Do not change this once it is set by the constructor.
self.import_name = import_name
- self.static_folder = static_folder
+
+ self.static_folder = static_folder # type: ignore
self.static_url_path = static_url_path
+
+ #: The path to the templates folder, relative to
+ #: :attr:`root_path`, to add to the template loader. ``None`` if
+ #: templates should not be added.
self.template_folder = template_folder
+
if root_path is None:
root_path = get_root_path(self.import_name)
+
+ #: Absolute path to the package on the filesystem. Used to look
+ #: up resources contained in the package.
self.root_path = root_path
+
+ #: A dictionary mapping endpoint names to view functions.
+ #:
+ #: To register a view function, use the :meth:`route` decorator.
+ #:
+ #: This data structure is internal. It should not be modified
+ #: directly and its format may change at any time.
self.view_functions: dict[str, ft.RouteCallable] = {}
- self.error_handler_spec: dict[ft.AppOrBlueprintKey, dict[int | None,
- dict[type[Exception], ft.ErrorHandlerCallable]]] = defaultdict(
- lambda : defaultdict(dict))
- self.before_request_funcs: dict[ft.AppOrBlueprintKey, list[ft.
- BeforeRequestCallable]] = defaultdict(list)
- self.after_request_funcs: dict[ft.AppOrBlueprintKey, list[ft.
- AfterRequestCallable[t.Any]]] = defaultdict(list)
- self.teardown_request_funcs: dict[ft.AppOrBlueprintKey, list[ft.
- TeardownCallable]] = defaultdict(list)
- self.template_context_processors: dict[ft.AppOrBlueprintKey, list[
- ft.TemplateContextProcessorCallable]] = defaultdict(list, {None:
- [_default_template_ctx_processor]})
- self.url_value_preprocessors: dict[ft.AppOrBlueprintKey, list[ft.
- URLValuePreprocessorCallable]] = defaultdict(list)
- self.url_default_functions: dict[ft.AppOrBlueprintKey, list[ft.
- URLDefaultCallable]] = defaultdict(list)
-
- def __repr__(self) ->str:
- return f'<{type(self).__name__} {self.name!r}>'
+
+ #: A data structure of registered error handlers, in the format
+ #: ``{scope: {code: {class: handler}}}``. The ``scope`` key is
+ #: the name of a blueprint the handlers are active for, or
+ #: ``None`` for all requests. The ``code`` key is the HTTP
+ #: status code for ``HTTPException``, or ``None`` for
+ #: other exceptions. The innermost dictionary maps exception
+ #: classes to handler functions.
+ #:
+ #: To register an error handler, use the :meth:`errorhandler`
+ #: decorator.
+ #:
+ #: This data structure is internal. It should not be modified
+ #: directly and its format may change at any time.
+ self.error_handler_spec: dict[
+ ft.AppOrBlueprintKey,
+ dict[int | None, dict[type[Exception], ft.ErrorHandlerCallable]],
+ ] = defaultdict(lambda: defaultdict(dict))
+
+ #: A data structure of functions to call at the beginning of
+ #: each request, in the format ``{scope: [functions]}``. The
+ #: ``scope`` key is the name of a blueprint the functions are
+ #: active for, or ``None`` for all requests.
+ #:
+ #: To register a function, use the :meth:`before_request`
+ #: decorator.
+ #:
+ #: This data structure is internal. It should not be modified
+ #: directly and its format may change at any time.
+ self.before_request_funcs: dict[
+ ft.AppOrBlueprintKey, list[ft.BeforeRequestCallable]
+ ] = defaultdict(list)
+
+ #: A data structure of functions to call at the end of each
+ #: request, in the format ``{scope: [functions]}``. The
+ #: ``scope`` key is the name of a blueprint the functions are
+ #: active for, or ``None`` for all requests.
+ #:
+ #: To register a function, use the :meth:`after_request`
+ #: decorator.
+ #:
+ #: This data structure is internal. It should not be modified
+ #: directly and its format may change at any time.
+ self.after_request_funcs: dict[
+ ft.AppOrBlueprintKey, list[ft.AfterRequestCallable[t.Any]]
+ ] = defaultdict(list)
+
+ #: A data structure of functions to call at the end of each
+ #: request even if an exception is raised, in the format
+ #: ``{scope: [functions]}``. The ``scope`` key is the name of a
+ #: blueprint the functions are active for, or ``None`` for all
+ #: requests.
+ #:
+ #: To register a function, use the :meth:`teardown_request`
+ #: decorator.
+ #:
+ #: This data structure is internal. It should not be modified
+ #: directly and its format may change at any time.
+ self.teardown_request_funcs: dict[
+ ft.AppOrBlueprintKey, list[ft.TeardownCallable]
+ ] = defaultdict(list)
+
+ #: A data structure of functions to call to pass extra context
+ #: values when rendering templates, in the format
+ #: ``{scope: [functions]}``. The ``scope`` key is the name of a
+ #: blueprint the functions are active for, or ``None`` for all
+ #: requests.
+ #:
+ #: To register a function, use the :meth:`context_processor`
+ #: decorator.
+ #:
+ #: This data structure is internal. It should not be modified
+ #: directly and its format may change at any time.
+ self.template_context_processors: dict[
+ ft.AppOrBlueprintKey, list[ft.TemplateContextProcessorCallable]
+ ] = defaultdict(list, {None: [_default_template_ctx_processor]})
+
+ #: A data structure of functions to call to modify the keyword
+ #: arguments passed to the view function, in the format
+ #: ``{scope: [functions]}``. The ``scope`` key is the name of a
+ #: blueprint the functions are active for, or ``None`` for all
+ #: requests.
+ #:
+ #: To register a function, use the
+ #: :meth:`url_value_preprocessor` decorator.
+ #:
+ #: This data structure is internal. It should not be modified
+ #: directly and its format may change at any time.
+ self.url_value_preprocessors: dict[
+ ft.AppOrBlueprintKey,
+ list[ft.URLValuePreprocessorCallable],
+ ] = defaultdict(list)
+
+ #: A data structure of functions to call to modify the keyword
+ #: arguments when generating URLs, in the format
+ #: ``{scope: [functions]}``. The ``scope`` key is the name of a
+ #: blueprint the functions are active for, or ``None`` for all
+ #: requests.
+ #:
+ #: To register a function, use the :meth:`url_defaults`
+ #: decorator.
+ #:
+ #: This data structure is internal. It should not be modified
+ #: directly and its format may change at any time.
+ self.url_default_functions: dict[
+ ft.AppOrBlueprintKey, list[ft.URLDefaultCallable]
+ ] = defaultdict(list)
+
+ def __repr__(self) -> str:
+ return f"<{type(self).__name__} {self.name!r}>"
+
+ def _check_setup_finished(self, f_name: str) -> None:
+ raise NotImplementedError
@property
- def static_folder(self) ->(str | None):
+ def static_folder(self) -> str | None:
"""The absolute path to the configured static folder. ``None``
if no static folder is set.
"""
- pass
+ if self._static_folder is not None:
+ return os.path.join(self.root_path, self._static_folder)
+ else:
+ return None
+
+ @static_folder.setter
+ def static_folder(self, value: str | os.PathLike[str] | None) -> None:
+ if value is not None:
+ value = os.fspath(value).rstrip(r"\/")
+
+ self._static_folder = value
@property
- def has_static_folder(self) ->bool:
+ def has_static_folder(self) -> bool:
"""``True`` if :attr:`static_folder` is set.
.. versionadded:: 0.5
"""
- pass
+ return self.static_folder is not None
@property
- def static_url_path(self) ->(str | None):
+ def static_url_path(self) -> str | None:
"""The URL prefix that the static route will be accessible from.
If it was not configured during init, it is derived from
:attr:`static_folder`.
"""
- pass
+ if self._static_url_path is not None:
+ return self._static_url_path
+
+ if self.static_folder is not None:
+ basename = os.path.basename(self.static_folder)
+ return f"/{basename}".rstrip("/")
+
+ return None
+
+ @static_url_path.setter
+ def static_url_path(self, value: str | None) -> None:
+ if value is not None:
+ value = value.rstrip("/")
+
+ self._static_url_path = value
@cached_property
- def jinja_loader(self) ->(BaseLoader | None):
+ def jinja_loader(self) -> BaseLoader | None:
"""The Jinja loader for this object's templates. By default this
is a class :class:`jinja2.loaders.FileSystemLoader` to
:attr:`template_folder` if it is set.
.. versionadded:: 0.5
"""
- pass
+ if self.template_folder is not None:
+ return FileSystemLoader(os.path.join(self.root_path, self.template_folder))
+ else:
+ return None
+
+ def _method_route(
+ self,
+ method: str,
+ rule: str,
+ options: dict[str, t.Any],
+ ) -> t.Callable[[T_route], T_route]:
+ if "methods" in options:
+ raise TypeError("Use the 'route' decorator to use the 'methods' argument.")
+
+ return self.route(rule, methods=[method], **options)
@setupmethod
- def get(self, rule: str, **options: t.Any) ->t.Callable[[T_route], T_route
- ]:
+ def get(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Shortcut for :meth:`route` with ``methods=["GET"]``.
.. versionadded:: 2.0
"""
- pass
+ return self._method_route("GET", rule, options)
@setupmethod
- def post(self, rule: str, **options: t.Any) ->t.Callable[[T_route], T_route
- ]:
+ def post(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Shortcut for :meth:`route` with ``methods=["POST"]``.
.. versionadded:: 2.0
"""
- pass
+ return self._method_route("POST", rule, options)
@setupmethod
- def put(self, rule: str, **options: t.Any) ->t.Callable[[T_route], T_route
- ]:
+ def put(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Shortcut for :meth:`route` with ``methods=["PUT"]``.
.. versionadded:: 2.0
"""
- pass
+ return self._method_route("PUT", rule, options)
@setupmethod
- def delete(self, rule: str, **options: t.Any) ->t.Callable[[T_route],
- T_route]:
+ def delete(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Shortcut for :meth:`route` with ``methods=["DELETE"]``.
.. versionadded:: 2.0
"""
- pass
+ return self._method_route("DELETE", rule, options)
@setupmethod
- def patch(self, rule: str, **options: t.Any) ->t.Callable[[T_route],
- T_route]:
+ def patch(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Shortcut for :meth:`route` with ``methods=["PATCH"]``.
.. versionadded:: 2.0
"""
- pass
+ return self._method_route("PATCH", rule, options)
@setupmethod
- def route(self, rule: str, **options: t.Any) ->t.Callable[[T_route],
- T_route]:
+ def route(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Decorate a view function to register it with the given URL
rule and options. Calls :meth:`add_url_rule`, which has more
details about the implementation.
@@ -190,13 +356,23 @@ class Scaffold:
:param options: Extra options passed to the
:class:`~werkzeug.routing.Rule` object.
"""
- pass
+
+ def decorator(f: T_route) -> T_route:
+ endpoint = options.pop("endpoint", None)
+ self.add_url_rule(rule, endpoint, f, **options)
+ return f
+
+ return decorator
@setupmethod
- def add_url_rule(self, rule: str, endpoint: (str | None)=None,
- view_func: (ft.RouteCallable | None)=None,
- provide_automatic_options: (bool | None)=None, **options: t.Any
- ) ->None:
+ def add_url_rule(
+ self,
+ rule: str,
+ endpoint: str | None = None,
+ view_func: ft.RouteCallable | None = None,
+ provide_automatic_options: bool | None = None,
+ **options: t.Any,
+ ) -> None:
"""Register a rule for routing incoming requests and building
URLs. The :meth:`route` decorator is a shortcut to call this
with the ``view_func`` argument. These are equivalent:
@@ -254,10 +430,10 @@ class Scaffold:
:param options: Extra options passed to the
:class:`~werkzeug.routing.Rule` object.
"""
- pass
+ raise NotImplementedError
@setupmethod
- def endpoint(self, endpoint: str) ->t.Callable[[F], F]:
+ def endpoint(self, endpoint: str) -> t.Callable[[F], F]:
"""Decorate a view function to register it for the given
endpoint. Used if a rule is added without a ``view_func`` with
:meth:`add_url_rule`.
@@ -273,10 +449,15 @@ class Scaffold:
:param endpoint: The endpoint name to associate with the view
function.
"""
- pass
+
+ def decorator(f: F) -> F:
+ self.view_functions[endpoint] = f
+ return f
+
+ return decorator
@setupmethod
- def before_request(self, f: T_before_request) ->T_before_request:
+ def before_request(self, f: T_before_request) -> T_before_request:
"""Register a function to run before each request.
For example, this can be used to open a database connection, or
@@ -299,10 +480,11 @@ class Scaffold:
every request that the blueprint handles. To register with a blueprint and
execute before every request, use :meth:`.Blueprint.before_app_request`.
"""
- pass
+ self.before_request_funcs.setdefault(None, []).append(f)
+ return f
@setupmethod
- def after_request(self, f: T_after_request) ->T_after_request:
+ def after_request(self, f: T_after_request) -> T_after_request:
"""Register a function to run after each request to this object.
The function is called with the response object, and must return
@@ -319,10 +501,11 @@ class Scaffold:
every request that the blueprint handles. To register with a blueprint and
execute after every request, use :meth:`.Blueprint.after_app_request`.
"""
- pass
+ self.after_request_funcs.setdefault(None, []).append(f)
+ return f
@setupmethod
- def teardown_request(self, f: T_teardown) ->T_teardown:
+ 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.
@@ -352,11 +535,14 @@ class Scaffold:
every request that the blueprint handles. To register with a blueprint and
execute after every request, use :meth:`.Blueprint.teardown_app_request`.
"""
- pass
+ self.teardown_request_funcs.setdefault(None, []).append(f)
+ return f
@setupmethod
- def context_processor(self, f: T_template_context_processor
- ) ->T_template_context_processor:
+ def context_processor(
+ self,
+ f: T_template_context_processor,
+ ) -> T_template_context_processor:
"""Registers a template context processor function. These functions run before
rendering a template. The keys of the returned dict are added as variables
available in the template.
@@ -366,11 +552,14 @@ class Scaffold:
for templates rendered from the blueprint's views. To register with a blueprint
and affect every template, use :meth:`.Blueprint.app_context_processor`.
"""
- pass
+ self.template_context_processors[None].append(f)
+ return f
@setupmethod
- def url_value_preprocessor(self, f: T_url_value_preprocessor
- ) ->T_url_value_preprocessor:
+ def url_value_preprocessor(
+ self,
+ f: T_url_value_preprocessor,
+ ) -> T_url_value_preprocessor:
"""Register a URL value preprocessor function for all view
functions in the application. These functions will be called before the
:meth:`before_request` functions.
@@ -388,10 +577,11 @@ class Scaffold:
requests that the blueprint handles. To register with a blueprint and affect
every request, use :meth:`.Blueprint.app_url_value_preprocessor`.
"""
- pass
+ self.url_value_preprocessors[None].append(f)
+ return f
@setupmethod
- def url_defaults(self, f: T_url_defaults) ->T_url_defaults:
+ def url_defaults(self, f: T_url_defaults) -> T_url_defaults:
"""Callback function for URL defaults for all view functions of the
application. It's called with the endpoint and values and should
update the values passed in place.
@@ -401,11 +591,13 @@ class Scaffold:
requests that the blueprint handles. To register with a blueprint and affect
every request, use :meth:`.Blueprint.app_url_defaults`.
"""
- pass
+ self.url_default_functions[None].append(f)
+ return f
@setupmethod
- def errorhandler(self, code_or_exception: (type[Exception] | int)
- ) ->t.Callable[[T_error_handler], T_error_handler]:
+ def errorhandler(
+ self, code_or_exception: type[Exception] | int
+ ) -> t.Callable[[T_error_handler], T_error_handler]:
"""Register a function to handle errors by code or exception class.
A decorator that is used to register a function given an
@@ -439,22 +631,32 @@ class Scaffold:
:param code_or_exception: the code as integer for the handler, or
an arbitrary exception
"""
- pass
+
+ def decorator(f: T_error_handler) -> T_error_handler:
+ self.register_error_handler(code_or_exception, f)
+ return f
+
+ return decorator
@setupmethod
- def register_error_handler(self, code_or_exception: (type[Exception] |
- int), f: ft.ErrorHandlerCallable) ->None:
+ def register_error_handler(
+ self,
+ code_or_exception: type[Exception] | int,
+ f: ft.ErrorHandlerCallable,
+ ) -> None:
"""Alternative error attach function to the :meth:`errorhandler`
decorator that is more straightforward to use for non decorator
usage.
.. versionadded:: 0.7
"""
- pass
+ exc_class, code = self._get_exc_class_and_code(code_or_exception)
+ self.error_handler_spec[None][code][exc_class] = f
@staticmethod
- def _get_exc_class_and_code(exc_class_or_code: (type[Exception] | int)
- ) ->tuple[type[Exception], int | None]:
+ def _get_exc_class_and_code(
+ exc_class_or_code: type[Exception] | int,
+ ) -> tuple[type[Exception], int | None]:
"""Get the exception class being handled. For HTTP status codes
or ``HTTPException`` subclasses, return both the exception and
status code.
@@ -462,22 +664,103 @@ class Scaffold:
:param exc_class_or_code: Any exception class, or an HTTP status
code as an integer.
"""
- pass
-
-
-def _endpoint_from_view_func(view_func: ft.RouteCallable) ->str:
+ exc_class: type[Exception]
+
+ if isinstance(exc_class_or_code, int):
+ try:
+ exc_class = default_exceptions[exc_class_or_code]
+ except KeyError:
+ raise ValueError(
+ f"'{exc_class_or_code}' is not a recognized HTTP"
+ " error code. Use a subclass of HTTPException with"
+ " that code instead."
+ ) from None
+ else:
+ exc_class = exc_class_or_code
+
+ if isinstance(exc_class, Exception):
+ raise TypeError(
+ f"{exc_class!r} is an instance, not a class. Handlers"
+ " can only be registered for Exception classes or HTTP"
+ " error codes."
+ )
+
+ if not issubclass(exc_class, Exception):
+ raise ValueError(
+ f"'{exc_class.__name__}' is not a subclass of Exception."
+ " Handlers can only be registered for Exception classes"
+ " or HTTP error codes."
+ )
+
+ if issubclass(exc_class, HTTPException):
+ return exc_class, exc_class.code
+ else:
+ return exc_class, None
+
+
+def _endpoint_from_view_func(view_func: ft.RouteCallable) -> str:
"""Internal helper that returns the default endpoint for a given
function. This always is the function name.
"""
- pass
+ assert view_func is not None, "expected view func if endpoint is not provided."
+ return view_func.__name__
-def _find_package_path(import_name: str) ->str:
- """Find the path that contains the package or module."""
- pass
+def _path_is_relative_to(path: pathlib.PurePath, base: str) -> bool:
+ # Path.is_relative_to doesn't exist until Python 3.9
+ try:
+ path.relative_to(base)
+ return True
+ except ValueError:
+ return False
-def find_package(import_name: str) ->tuple[str | None, str]:
+def _find_package_path(import_name: str) -> str:
+ """Find the path that contains the package or module."""
+ root_mod_name, _, _ = import_name.partition(".")
+
+ try:
+ root_spec = importlib.util.find_spec(root_mod_name)
+
+ if root_spec is None:
+ raise ValueError("not found")
+ except (ImportError, ValueError):
+ # ImportError: the machinery told us it does not exist
+ # ValueError:
+ # - the module name was invalid
+ # - the module name is __main__
+ # - we raised `ValueError` due to `root_spec` being `None`
+ return os.getcwd()
+
+ if root_spec.submodule_search_locations:
+ if root_spec.origin is None or root_spec.origin == "namespace":
+ # namespace package
+ package_spec = importlib.util.find_spec(import_name)
+
+ if package_spec is not None and package_spec.submodule_search_locations:
+ # Pick the path in the namespace that contains the submodule.
+ package_path = pathlib.Path(
+ os.path.commonpath(package_spec.submodule_search_locations)
+ )
+ search_location = next(
+ location
+ for location in root_spec.submodule_search_locations
+ if _path_is_relative_to(package_path, location)
+ )
+ else:
+ # Pick the first path.
+ search_location = root_spec.submodule_search_locations[0]
+
+ return os.path.dirname(search_location)
+ else:
+ # package with __init__.py
+ return os.path.dirname(os.path.dirname(root_spec.origin))
+ else:
+ # module
+ return os.path.dirname(root_spec.origin) # type: ignore[type-var, return-value]
+
+
+def find_package(import_name: str) -> tuple[str | None, str]:
"""Find the prefix that a package is installed under, and the path
that it would be imported from.
@@ -490,4 +773,29 @@ def find_package(import_name: str) ->tuple[str | None, str]:
for import. If the package is not installed, it's assumed that the
package was imported from the current working directory.
"""
- pass
+ package_path = _find_package_path(import_name)
+ py_prefix = os.path.abspath(sys.prefix)
+
+ # installed to the system
+ if _path_is_relative_to(pathlib.PurePath(package_path), py_prefix):
+ return py_prefix, package_path
+
+ site_parent, site_folder = os.path.split(package_path)
+
+ # installed to a virtualenv
+ if site_folder.lower() == "site-packages":
+ parent, folder = os.path.split(site_parent)
+
+ # Windows (prefix/lib/site-packages)
+ if folder.lower() == "lib":
+ return parent, package_path
+
+ # Unix (prefix/lib/pythonX.Y/site-packages)
+ if os.path.basename(parent).lower() == "lib":
+ return os.path.dirname(parent), package_path
+
+ # something else (prefix/site-packages)
+ return site_parent, package_path
+
+ # not installed
+ return None, package_path
diff --git a/src/flask/sessions.py b/src/flask/sessions.py
index 5ed091cb..ee19ad63 100644
--- a/src/flask/sessions.py
+++ b/src/flask/sessions.py
@@ -1,33 +1,56 @@
from __future__ import annotations
+
import hashlib
import typing as t
from collections.abc import MutableMapping
from datetime import datetime
from datetime import timezone
+
from itsdangerous import BadSignature
from itsdangerous import URLSafeTimedSerializer
from werkzeug.datastructures import CallbackDict
+
from .json.tag import TaggedJSONSerializer
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
import typing_extensions as te
+
from .app import Flask
from .wrappers import Request
from .wrappers import Response
-class SessionMixin(MutableMapping):
+# TODO generic when Python > 3.8
+class SessionMixin(MutableMapping): # type: ignore[type-arg]
"""Expands a basic dictionary with session attributes."""
@property
- def permanent(self) ->bool:
+ def permanent(self) -> bool:
"""This reflects the ``'_permanent'`` key in the dict."""
- pass
+ return self.get("_permanent", False)
+
+ @permanent.setter
+ def permanent(self, value: bool) -> None:
+ self["_permanent"] = bool(value)
+
+ #: Some implementations can detect whether a session is newly
+ #: created, but that is not guaranteed. Use with caution. The mixin
+ # default is hard-coded ``False``.
new = False
+
+ #: Some implementations can detect changes to the session and set
+ #: this when that happens. The mixin default is hard coded to
+ #: ``True``.
modified = True
+
+ #: Some implementations can detect when session data is read or
+ #: written and set this when that happens. The mixin default is hard
+ #: coded to ``True``.
accessed = True
-class SecureCookieSession(CallbackDict, SessionMixin):
+# TODO generic when Python > 3.8
+class SecureCookieSession(CallbackDict, SessionMixin): # type: ignore[type-arg]
"""Base class for sessions based on signed cookies.
This session backend will set the :attr:`modified` and
@@ -35,28 +58,54 @@ class SecureCookieSession(CallbackDict, SessionMixin):
session is new (vs. empty), so :attr:`new` remains hard coded to
``False``.
"""
+
+ #: When data is changed, this is set to ``True``. Only the session
+ #: dictionary itself is tracked; if the session contains mutable
+ #: data (for example a nested dict) then this must be set to
+ #: ``True`` manually when modifying that data. The session cookie
+ #: will only be written to the response if this is ``True``.
modified = False
- accessed = False
- def __init__(self, initial: t.Any=None) ->None:
+ #: When data is read or written, this is set to ``True``. Used by
+ # :class:`.SecureCookieSessionInterface` to add a ``Vary: Cookie``
+ #: header, which allows caching proxies to cache different pages for
+ #: different users.
+ accessed = False
- def on_update(self: te.Self) ->None:
+ def __init__(self, initial: t.Any = None) -> None:
+ def on_update(self: te.Self) -> None:
self.modified = True
self.accessed = True
+
super().__init__(initial, on_update)
- def __getitem__(self, key: str) ->t.Any:
+ def __getitem__(self, key: str) -> t.Any:
self.accessed = True
return super().__getitem__(key)
+ def get(self, key: str, default: t.Any = None) -> t.Any:
+ self.accessed = True
+ return super().get(key, default)
+
+ def setdefault(self, key: str, default: t.Any = None) -> t.Any:
+ self.accessed = True
+ return super().setdefault(key, default)
+
class NullSession(SecureCookieSession):
"""Class used to generate nicer error messages if sessions are not
available. Will still allow read-only access to the empty session
but fail on setting.
"""
- (__setitem__) = (__delitem__) = (clear) = (pop) = (popitem) = (update) = (
- setdefault) = _fail
+
+ def _fail(self, *args: t.Any, **kwargs: t.Any) -> t.NoReturn:
+ raise RuntimeError(
+ "The session is unavailable because no secret "
+ "key was set. Set the secret_key on the "
+ "application to something unique and secret."
+ )
+
+ __setitem__ = __delitem__ = clear = pop = popitem = update = setdefault = _fail # type: ignore # noqa: B950
del _fail
@@ -96,10 +145,21 @@ class SessionInterface:
.. versionadded:: 0.8
"""
+
+ #: :meth:`make_null_session` will look here for the class that should
+ #: be created when a null session is requested. Likewise the
+ #: :meth:`is_null_session` method will perform a typecheck against
+ #: this type.
null_session_class = NullSession
+
+ #: A flag that indicates if the session interface is pickle based.
+ #: This can be used by Flask extensions to make a decision in regards
+ #: to how to deal with the session object.
+ #:
+ #: .. versionadded:: 0.10
pickle_based = False
- def make_null_session(self, app: Flask) ->NullSession:
+ def make_null_session(self, app: Flask) -> NullSession:
"""Creates a null session which acts as a replacement object if the
real session support could not be loaded due to a configuration
error. This mainly aids the user experience because the job of the
@@ -109,22 +169,22 @@ class SessionInterface:
This creates an instance of :attr:`null_session_class` by default.
"""
- pass
+ return self.null_session_class()
- def is_null_session(self, obj: object) ->bool:
+ def is_null_session(self, obj: object) -> bool:
"""Checks if a given object is a null session. Null sessions are
not asked to be saved.
This checks if the object is an instance of :attr:`null_session_class`
by default.
"""
- pass
+ return isinstance(obj, self.null_session_class)
- def get_cookie_name(self, app: Flask) ->str:
+ def get_cookie_name(self, app: Flask) -> str:
"""The name of the session cookie. Uses``app.config["SESSION_COOKIE_NAME"]``."""
- pass
+ return app.config["SESSION_COOKIE_NAME"] # type: ignore[no-any-return]
- def get_cookie_domain(self, app: Flask) ->(str | None):
+ def get_cookie_domain(self, app: Flask) -> str | None:
"""The value of the ``Domain`` parameter on the session cookie. If not set,
browsers will only send the cookie to the exact domain it was set from.
Otherwise, they will send it to any subdomain of the given value as well.
@@ -134,46 +194,47 @@ class SessionInterface:
.. versionchanged:: 2.3
Not set by default, does not fall back to ``SERVER_NAME``.
"""
- pass
+ return app.config["SESSION_COOKIE_DOMAIN"] # type: ignore[no-any-return]
- def get_cookie_path(self, app: Flask) ->str:
+ def get_cookie_path(self, app: Flask) -> str:
"""Returns the path for which the cookie should be valid. The
default implementation uses the value from the ``SESSION_COOKIE_PATH``
config var if it's set, and falls back to ``APPLICATION_ROOT`` or
uses ``/`` if it's ``None``.
"""
- pass
+ return app.config["SESSION_COOKIE_PATH"] or app.config["APPLICATION_ROOT"] # type: ignore[no-any-return]
- def get_cookie_httponly(self, app: Flask) ->bool:
+ def get_cookie_httponly(self, app: Flask) -> bool:
"""Returns True if the session cookie should be httponly. This
currently just returns the value of the ``SESSION_COOKIE_HTTPONLY``
config var.
"""
- pass
+ return app.config["SESSION_COOKIE_HTTPONLY"] # type: ignore[no-any-return]
- def get_cookie_secure(self, app: Flask) ->bool:
+ def get_cookie_secure(self, app: Flask) -> bool:
"""Returns True if the cookie should be secure. This currently
just returns the value of the ``SESSION_COOKIE_SECURE`` setting.
"""
- pass
+ return app.config["SESSION_COOKIE_SECURE"] # type: ignore[no-any-return]
- def get_cookie_samesite(self, app: Flask) ->(str | None):
+ def get_cookie_samesite(self, app: Flask) -> str | None:
"""Return ``'Strict'`` or ``'Lax'`` if the cookie should use the
``SameSite`` attribute. This currently just returns the value of
the :data:`SESSION_COOKIE_SAMESITE` setting.
"""
- pass
+ return app.config["SESSION_COOKIE_SAMESITE"] # type: ignore[no-any-return]
- def get_expiration_time(self, app: Flask, session: SessionMixin) ->(
- datetime | None):
+ def get_expiration_time(self, app: Flask, session: SessionMixin) -> datetime | None:
"""A helper method that returns an expiration date for the session
or ``None`` if the session is linked to the browser session. The
default implementation returns now + the permanent session
lifetime configured on the application.
"""
- pass
+ if session.permanent:
+ return datetime.now(timezone.utc) + app.permanent_session_lifetime
+ return None
- def should_set_cookie(self, app: Flask, session: SessionMixin) ->bool:
+ def should_set_cookie(self, app: Flask, session: SessionMixin) -> bool:
"""Used by session backends to determine if a ``Set-Cookie`` header
should be set for this session cookie for this response. If the session
has been modified, the cookie is set. If the session is permanent and
@@ -184,10 +245,12 @@ class SessionInterface:
.. versionadded:: 0.11
"""
- pass
- def open_session(self, app: Flask, request: Request) ->(SessionMixin | None
- ):
+ return session.modified or (
+ session.permanent and app.config["SESSION_REFRESH_EACH_REQUEST"]
+ )
+
+ def open_session(self, app: Flask, request: Request) -> SessionMixin | None:
"""This is called at the beginning of each request, after
pushing the request context, before matching the URL.
@@ -199,34 +262,118 @@ class SessionInterface:
context will fall back to using :meth:`make_null_session`
in this case.
"""
- pass
+ raise NotImplementedError()
- def save_session(self, app: Flask, session: SessionMixin, response:
- Response) ->None:
+ def save_session(
+ self, app: Flask, session: SessionMixin, response: Response
+ ) -> None:
"""This is called at the end of each request, after generating
a response, before removing the request context. It is skipped
if :meth:`is_null_session` returns ``True``.
"""
- pass
+ raise NotImplementedError()
session_json_serializer = TaggedJSONSerializer()
-def _lazy_sha1(string: bytes=b'') ->t.Any:
+def _lazy_sha1(string: bytes = b"") -> t.Any:
"""Don't access ``hashlib.sha1`` until runtime. FIPS builds may not include
SHA-1, in which case the import and use as a default would fail before the
developer can configure something else.
"""
- pass
+ return hashlib.sha1(string)
class SecureCookieSessionInterface(SessionInterface):
"""The default session interface that stores sessions in signed cookies
through the :mod:`itsdangerous` module.
"""
- salt = 'cookie-session'
+
+ #: the salt that should be applied on top of the secret key for the
+ #: signing of cookie based sessions.
+ salt = "cookie-session"
+ #: the hash function to use for the signature. The default is sha1
digest_method = staticmethod(_lazy_sha1)
- key_derivation = 'hmac'
+ #: the name of the itsdangerous supported key derivation. The default
+ #: is hmac.
+ key_derivation = "hmac"
+ #: A python serializer for the payload. The default is a compact
+ #: JSON derived serializer with support for some extra Python types
+ #: such as datetime objects or tuples.
serializer = session_json_serializer
session_class = SecureCookieSession
+
+ def get_signing_serializer(self, app: Flask) -> URLSafeTimedSerializer | None:
+ if not app.secret_key:
+ return None
+ signer_kwargs = dict(
+ key_derivation=self.key_derivation, digest_method=self.digest_method
+ )
+ return URLSafeTimedSerializer(
+ app.secret_key,
+ salt=self.salt,
+ serializer=self.serializer,
+ signer_kwargs=signer_kwargs,
+ )
+
+ def open_session(self, app: Flask, request: Request) -> SecureCookieSession | None:
+ s = self.get_signing_serializer(app)
+ if s is None:
+ return None
+ val = request.cookies.get(self.get_cookie_name(app))
+ if not val:
+ return self.session_class()
+ max_age = int(app.permanent_session_lifetime.total_seconds())
+ try:
+ data = s.loads(val, max_age=max_age)
+ return self.session_class(data)
+ except BadSignature:
+ return self.session_class()
+
+ def save_session(
+ self, app: Flask, session: SessionMixin, response: Response
+ ) -> None:
+ name = self.get_cookie_name(app)
+ domain = self.get_cookie_domain(app)
+ path = self.get_cookie_path(app)
+ secure = self.get_cookie_secure(app)
+ samesite = self.get_cookie_samesite(app)
+ httponly = self.get_cookie_httponly(app)
+
+ # Add a "Vary: Cookie" header if the session was accessed at all.
+ if session.accessed:
+ response.vary.add("Cookie")
+
+ # If the session is modified to be empty, remove the cookie.
+ # If the session is empty, return without setting the cookie.
+ if not session:
+ if session.modified:
+ response.delete_cookie(
+ name,
+ domain=domain,
+ path=path,
+ secure=secure,
+ samesite=samesite,
+ httponly=httponly,
+ )
+ response.vary.add("Cookie")
+
+ return
+
+ if not self.should_set_cookie(app, session):
+ return
+
+ expires = self.get_expiration_time(app, session)
+ val = self.get_signing_serializer(app).dumps(dict(session)) # type: ignore
+ response.set_cookie(
+ name,
+ val, # type: ignore
+ expires=expires,
+ httponly=httponly,
+ domain=domain,
+ path=path,
+ secure=secure,
+ samesite=samesite,
+ )
+ response.vary.add("Cookie")
diff --git a/src/flask/signals.py b/src/flask/signals.py
index 140a1137..444fda99 100644
--- a/src/flask/signals.py
+++ b/src/flask/signals.py
@@ -1,13 +1,17 @@
from __future__ import annotations
+
from blinker import Namespace
+
+# This namespace is only for signals provided by Flask itself.
_signals = Namespace()
-template_rendered = _signals.signal('template-rendered')
-before_render_template = _signals.signal('before-render-template')
-request_started = _signals.signal('request-started')
-request_finished = _signals.signal('request-finished')
-request_tearing_down = _signals.signal('request-tearing-down')
-got_request_exception = _signals.signal('got-request-exception')
-appcontext_tearing_down = _signals.signal('appcontext-tearing-down')
-appcontext_pushed = _signals.signal('appcontext-pushed')
-appcontext_popped = _signals.signal('appcontext-popped')
-message_flashed = _signals.signal('message-flashed')
+
+template_rendered = _signals.signal("template-rendered")
+before_render_template = _signals.signal("before-render-template")
+request_started = _signals.signal("request-started")
+request_finished = _signals.signal("request-finished")
+request_tearing_down = _signals.signal("request-tearing-down")
+got_request_exception = _signals.signal("got-request-exception")
+appcontext_tearing_down = _signals.signal("appcontext-tearing-down")
+appcontext_pushed = _signals.signal("appcontext-pushed")
+appcontext_popped = _signals.signal("appcontext-popped")
+message_flashed = _signals.signal("message-flashed")
diff --git a/src/flask/templating.py b/src/flask/templating.py
index 861e328e..618a3b35 100644
--- a/src/flask/templating.py
+++ b/src/flask/templating.py
@@ -1,9 +1,12 @@
from __future__ import annotations
+
import typing as t
+
from jinja2 import BaseLoader
from jinja2 import Environment as BaseEnvironment
from jinja2 import Template
from jinja2 import TemplateNotFound
+
from .globals import _cv_app
from .globals import _cv_request
from .globals import current_app
@@ -11,17 +14,26 @@ from .globals import request
from .helpers import stream_with_context
from .signals import before_render_template
from .signals import template_rendered
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from .app import Flask
from .sansio.app import App
from .sansio.scaffold import Scaffold
-def _default_template_ctx_processor() ->dict[str, t.Any]:
+def _default_template_ctx_processor() -> dict[str, t.Any]:
"""Default template context processor. Injects `request`,
`session` and `g`.
"""
- pass
+ appctx = _cv_app.get(None)
+ reqctx = _cv_request.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
+ return rv
class Environment(BaseEnvironment):
@@ -30,9 +42,9 @@ class Environment(BaseEnvironment):
name of the blueprint to referenced templates if necessary.
"""
- def __init__(self, app: App, **options: t.Any) ->None:
- if 'loader' not in options:
- options['loader'] = app.create_global_jinja_loader()
+ def __init__(self, app: App, **options: t.Any) -> None:
+ if "loader" not in options:
+ options["loader"] = app.create_global_jinja_loader()
BaseEnvironment.__init__(self, **options)
self.app = app
@@ -42,33 +54,141 @@ class DispatchingJinjaLoader(BaseLoader):
the blueprint folders.
"""
- def __init__(self, app: App) ->None:
+ def __init__(self, app: App) -> None:
self.app = app
-
-def render_template(template_name_or_list: (str | Template | list[str |
- Template]), **context: t.Any) ->str:
+ def get_source(
+ self, environment: BaseEnvironment, template: str
+ ) -> tuple[str, str | None, t.Callable[[], bool] | None]:
+ if self.app.config["EXPLAIN_TEMPLATE_LOADING"]:
+ return self._get_source_explained(environment, template)
+ return self._get_source_fast(environment, template)
+
+ def _get_source_explained(
+ self, environment: BaseEnvironment, template: str
+ ) -> tuple[str, str | None, t.Callable[[], bool] | None]:
+ attempts = []
+ rv: tuple[str, str | None, t.Callable[[], bool] | None] | None
+ trv: None | (tuple[str, str | None, t.Callable[[], bool] | None]) = None
+
+ for srcobj, loader in self._iter_loaders(template):
+ try:
+ rv = loader.get_source(environment, template)
+ if trv is None:
+ trv = rv
+ except TemplateNotFound:
+ rv = None
+ attempts.append((loader, srcobj, rv))
+
+ from .debughelpers import explain_template_loading_attempts
+
+ explain_template_loading_attempts(self.app, template, attempts)
+
+ if trv is not None:
+ return trv
+ raise TemplateNotFound(template)
+
+ def _get_source_fast(
+ self, environment: BaseEnvironment, template: str
+ ) -> tuple[str, str | None, t.Callable[[], bool] | None]:
+ for _srcobj, loader in self._iter_loaders(template):
+ try:
+ return loader.get_source(environment, template)
+ except TemplateNotFound:
+ continue
+ raise TemplateNotFound(template)
+
+ def _iter_loaders(self, template: str) -> t.Iterator[tuple[Scaffold, BaseLoader]]:
+ loader = self.app.jinja_loader
+ if loader is not None:
+ yield self.app, loader
+
+ for blueprint in self.app.iter_blueprints():
+ loader = blueprint.jinja_loader
+ if loader is not None:
+ yield blueprint, loader
+
+ def list_templates(self) -> list[str]:
+ result = set()
+ loader = self.app.jinja_loader
+ if loader is not None:
+ result.update(loader.list_templates())
+
+ for blueprint in self.app.iter_blueprints():
+ loader = blueprint.jinja_loader
+ if loader is not None:
+ for template in loader.list_templates():
+ result.add(template)
+
+ return list(result)
+
+
+def _render(app: Flask, template: Template, context: dict[str, t.Any]) -> str:
+ app.update_template_context(context)
+ before_render_template.send(
+ app, _async_wrapper=app.ensure_sync, template=template, context=context
+ )
+ rv = template.render(context)
+ template_rendered.send(
+ app, _async_wrapper=app.ensure_sync, template=template, context=context
+ )
+ return rv
+
+
+def render_template(
+ template_name_or_list: str | Template | list[str | Template],
+ **context: t.Any,
+) -> str:
"""Render a template by name with the given context.
:param template_name_or_list: The name of the template to render. If
a list is given, the first name to exist will be rendered.
:param context: The variables to make available in the template.
"""
- pass
+ app = current_app._get_current_object() # type: ignore[attr-defined]
+ template = app.jinja_env.get_or_select_template(template_name_or_list)
+ return _render(app, template, context)
-def render_template_string(source: str, **context: t.Any) ->str:
+def render_template_string(source: str, **context: t.Any) -> str:
"""Render a template from the given source string with the given
context.
:param source: The source code of the template to render.
:param context: The variables to make available in the template.
"""
- pass
+ app = current_app._get_current_object() # type: ignore[attr-defined]
+ template = app.jinja_env.from_string(source)
+ return _render(app, template, context)
+
+
+def _stream(
+ app: Flask, template: Template, context: dict[str, t.Any]
+) -> t.Iterator[str]:
+ app.update_template_context(context)
+ before_render_template.send(
+ app, _async_wrapper=app.ensure_sync, template=template, context=context
+ )
+
+ def generate() -> t.Iterator[str]:
+ yield from template.generate(context)
+ template_rendered.send(
+ 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
-def stream_template(template_name_or_list: (str | Template | list[str |
- Template]), **context: t.Any) ->t.Iterator[str]:
+def stream_template(
+ template_name_or_list: str | Template | list[str | Template],
+ **context: t.Any,
+) -> t.Iterator[str]:
"""Render a template by name with the given context as a stream.
This returns an iterator of strings, which can be used as a
streaming response from a view.
@@ -79,10 +199,12 @@ def stream_template(template_name_or_list: (str | Template | list[str |
.. versionadded:: 2.2
"""
- pass
+ app = current_app._get_current_object() # type: ignore[attr-defined]
+ template = app.jinja_env.get_or_select_template(template_name_or_list)
+ return _stream(app, template, context)
-def stream_template_string(source: str, **context: t.Any) ->t.Iterator[str]:
+def stream_template_string(source: str, **context: t.Any) -> t.Iterator[str]:
"""Render a template from the given source string with the given
context as a stream. This returns an iterator of strings, which can
be used as a streaming response from a view.
@@ -92,4 +214,6 @@ def stream_template_string(source: str, **context: t.Any) ->t.Iterator[str]:
.. versionadded:: 2.2
"""
- pass
+ app = current_app._get_current_object() # type: ignore[attr-defined]
+ template = app.jinja_env.from_string(source)
+ return _stream(app, template, context)
diff --git a/src/flask/testing.py b/src/flask/testing.py
index 7c533f33..a27b7c8f 100644
--- a/src/flask/testing.py
+++ b/src/flask/testing.py
@@ -1,4 +1,5 @@
from __future__ import annotations
+
import importlib.metadata
import typing as t
from contextlib import contextmanager
@@ -6,15 +7,19 @@ from contextlib import ExitStack
from copy import copy
from types import TracebackType
from urllib.parse import urlsplit
+
import werkzeug.test
from click.testing import CliRunner
from werkzeug.test import Client
from werkzeug.wrappers import Request as BaseRequest
+
from .cli import ScriptInfo
from .sessions import SessionMixin
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from _typeshed.wsgi import WSGIEnvironment
from werkzeug.test import TestResponse
+
from .app import Flask
@@ -40,40 +45,65 @@ class EnvironBuilder(werkzeug.test.EnvironBuilder):
:class:`~werkzeug.test.EnvironBuilder`.
"""
- def __init__(self, app: Flask, path: str='/', base_url: (str | None)=
- None, subdomain: (str | None)=None, url_scheme: (str | None)=None,
- *args: t.Any, **kwargs: t.Any) ->None:
- assert not (base_url or subdomain or url_scheme) or (base_url is not
- None) != bool(subdomain or url_scheme
- ), 'Cannot pass "subdomain" or "url_scheme" with "base_url".'
+ def __init__(
+ self,
+ app: Flask,
+ path: str = "/",
+ base_url: str | None = None,
+ subdomain: str | None = None,
+ url_scheme: str | None = None,
+ *args: t.Any,
+ **kwargs: t.Any,
+ ) -> None:
+ assert not (base_url or subdomain or url_scheme) or (
+ base_url is not None
+ ) != bool(
+ subdomain or url_scheme
+ ), 'Cannot pass "subdomain" or "url_scheme" with "base_url".'
+
if base_url is None:
- http_host = app.config.get('SERVER_NAME') or 'localhost'
- app_root = app.config['APPLICATION_ROOT']
+ http_host = app.config.get("SERVER_NAME") or "localhost"
+ app_root = app.config["APPLICATION_ROOT"]
+
if subdomain:
- http_host = f'{subdomain}.{http_host}'
+ http_host = f"{subdomain}.{http_host}"
+
if url_scheme is None:
- url_scheme = app.config['PREFERRED_URL_SCHEME']
+ url_scheme = app.config["PREFERRED_URL_SCHEME"]
+
url = urlsplit(path)
base_url = (
- f"{url.scheme or url_scheme}://{url.netloc or http_host}/{app_root.lstrip('/')}"
- )
+ f"{url.scheme or url_scheme}://{url.netloc or http_host}"
+ f"/{app_root.lstrip('/')}"
+ )
path = url.path
+
if url.query:
- sep = b'?' if isinstance(url.query, bytes) else '?'
+ sep = b"?" if isinstance(url.query, bytes) else "?"
path += sep + url.query
+
self.app = app
super().__init__(path, base_url, *args, **kwargs)
- def json_dumps(self, obj: t.Any, **kwargs: t.Any) ->str:
+ def json_dumps(self, obj: t.Any, **kwargs: t.Any) -> str: # type: ignore
"""Serialize ``obj`` to a JSON-formatted string.
The serialization will be configured according to the config associated
with this EnvironBuilder's ``app``.
"""
- pass
+ return self.app.json.dumps(obj, **kwargs)
+
+
+_werkzeug_version = ""
+
+
+def _get_werkzeug_version() -> str:
+ global _werkzeug_version
+ if not _werkzeug_version:
+ _werkzeug_version = importlib.metadata.version("werkzeug")
-_werkzeug_version = ''
+ return _werkzeug_version
class FlaskClient(Client):
@@ -89,19 +119,23 @@ class FlaskClient(Client):
Basic usage is outlined in the :doc:`/testing` chapter.
"""
+
application: Flask
- def __init__(self, *args: t.Any, **kwargs: t.Any) ->None:
+ def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
super().__init__(*args, **kwargs)
self.preserve_context = False
self._new_contexts: list[t.ContextManager[t.Any]] = []
self._context_stack = ExitStack()
- self.environ_base = {'REMOTE_ADDR': '127.0.0.1', 'HTTP_USER_AGENT':
- f'Werkzeug/{_get_werkzeug_version()}'}
+ self.environ_base = {
+ "REMOTE_ADDR": "127.0.0.1",
+ "HTTP_USER_AGENT": f"Werkzeug/{_get_werkzeug_version()}",
+ }
@contextmanager
- def session_transaction(self, *args: t.Any, **kwargs: t.Any) ->t.Iterator[
- SessionMixin]:
+ def session_transaction(
+ self, *args: t.Any, **kwargs: t.Any
+ ) -> t.Iterator[SessionMixin]:
"""When used in combination with a ``with`` statement this opens a
session transaction. This can be used to modify the session that
the test client uses. Once the ``with`` block is left the session is
@@ -118,16 +152,112 @@ class FlaskClient(Client):
:meth:`~flask.Flask.test_request_context` which are directly
passed through.
"""
- pass
+ if self._cookies is None:
+ raise TypeError(
+ "Cookies are disabled. Create a client with 'use_cookies=True'."
+ )
+
+ app = self.application
+ ctx = app.test_request_context(*args, **kwargs)
+ self._add_cookies_to_wsgi(ctx.request.environ)
+
+ with ctx:
+ sess = app.session_interface.open_session(app, ctx.request)
+
+ if sess is None:
+ raise RuntimeError("Session backend did not open a session.")
- def __enter__(self) ->FlaskClient:
+ yield sess
+ resp = app.response_class()
+
+ if app.session_interface.is_null_session(sess):
+ return
+
+ with ctx:
+ app.session_interface.save_session(app, sess, resp)
+
+ self._update_cookies_from_response(
+ ctx.request.host.partition(":")[0],
+ ctx.request.path,
+ resp.headers.getlist("Set-Cookie"),
+ )
+
+ def _copy_environ(self, other: WSGIEnvironment) -> WSGIEnvironment:
+ out = {**self.environ_base, **other}
+
+ if self.preserve_context:
+ out["werkzeug.debug.preserve_context"] = self._new_contexts.append
+
+ return out
+
+ def _request_from_builder_args(
+ self, args: tuple[t.Any, ...], kwargs: dict[str, t.Any]
+ ) -> BaseRequest:
+ kwargs["environ_base"] = self._copy_environ(kwargs.get("environ_base", {}))
+ builder = EnvironBuilder(self.application, *args, **kwargs)
+
+ try:
+ return builder.get_request()
+ finally:
+ builder.close()
+
+ def open(
+ self,
+ *args: t.Any,
+ buffered: bool = False,
+ follow_redirects: bool = False,
+ **kwargs: t.Any,
+ ) -> TestResponse:
+ if args and isinstance(
+ args[0], (werkzeug.test.EnvironBuilder, dict, BaseRequest)
+ ):
+ if isinstance(args[0], werkzeug.test.EnvironBuilder):
+ builder = copy(args[0])
+ builder.environ_base = self._copy_environ(builder.environ_base or {}) # type: ignore[arg-type]
+ request = builder.get_request()
+ elif isinstance(args[0], dict):
+ request = EnvironBuilder.from_environ(
+ args[0], app=self.application, environ_base=self._copy_environ({})
+ ).get_request()
+ else:
+ # isinstance(args[0], BaseRequest)
+ request = copy(args[0])
+ request.environ = self._copy_environ(request.environ)
+ else:
+ # request is None
+ request = self._request_from_builder_args(args, kwargs)
+
+ # Pop any previously preserved contexts. This prevents contexts
+ # from being preserved across redirects or multiple requests
+ # within a single block.
+ self._context_stack.close()
+
+ response = super().open(
+ request,
+ buffered=buffered,
+ follow_redirects=follow_redirects,
+ )
+ response.json_module = self.application.json # type: ignore[assignment]
+
+ # Re-push contexts that were preserved during the request.
+ while self._new_contexts:
+ cm = self._new_contexts.pop()
+ self._context_stack.enter_context(cm)
+
+ return response
+
+ def __enter__(self) -> FlaskClient:
if self.preserve_context:
- raise RuntimeError('Cannot nest client invocations')
+ raise RuntimeError("Cannot nest client invocations")
self.preserve_context = True
return self
- def __exit__(self, exc_type: (type | None), exc_value: (BaseException |
- None), tb: (TracebackType | None)) ->None:
+ def __exit__(
+ self,
+ exc_type: type | None,
+ exc_value: BaseException | None,
+ tb: TracebackType | None,
+ ) -> None:
self.preserve_context = False
self._context_stack.close()
@@ -138,12 +268,13 @@ class FlaskCliRunner(CliRunner):
:meth:`~flask.Flask.test_cli_runner`. See :ref:`testing-cli`.
"""
- def __init__(self, app: Flask, **kwargs: t.Any) ->None:
+ def __init__(self, app: Flask, **kwargs: t.Any) -> None:
self.app = app
super().__init__(**kwargs)
- def invoke(self, cli: t.Any=None, args: t.Any=None, **kwargs: t.Any
- ) ->t.Any:
+ def invoke( # type: ignore
+ self, cli: t.Any = None, args: t.Any = None, **kwargs: t.Any
+ ) -> t.Any:
"""Invokes a CLI command in an isolated environment. See
:meth:`CliRunner.invoke <click.testing.CliRunner.invoke>` for
full method documentation. See :ref:`testing-cli` for examples.
@@ -158,4 +289,10 @@ class FlaskCliRunner(CliRunner):
:return: a :class:`~click.testing.Result` object.
"""
- pass
+ if cli is None:
+ cli = self.app.cli
+
+ if "obj" not in kwargs:
+ kwargs["obj"] = ScriptInfo(create_app=lambda: self.app)
+
+ return super().invoke(cli, args, **kwargs)
diff --git a/src/flask/typing.py b/src/flask/typing.py
index cc21de38..cf6d4ae6 100644
--- a/src/flask/typing.py
+++ b/src/flask/typing.py
@@ -1,38 +1,90 @@
from __future__ import annotations
+
import typing as t
-if t.TYPE_CHECKING:
- from _typeshed.wsgi import WSGIApplication
- from werkzeug.datastructures import Headers
- from werkzeug.sansio.response import Response
-ResponseValue = t.Union['Response', str, bytes, t.List[t.Any], t.Mapping[
- str, t.Any], t.Iterator[str], t.Iterator[bytes]]
+
+if t.TYPE_CHECKING: # pragma: no cover
+ from _typeshed.wsgi import WSGIApplication # noqa: F401
+ from werkzeug.datastructures import Headers # noqa: F401
+ from werkzeug.sansio.response import Response # noqa: F401
+
+# The possible types that are directly convertible or are a Response object.
+ResponseValue = t.Union[
+ "Response",
+ str,
+ bytes,
+ t.List[t.Any],
+ # Only dict is actually accepted, but Mapping allows for TypedDict.
+ t.Mapping[str, t.Any],
+ t.Iterator[str],
+ t.Iterator[bytes],
+]
+
+# the possible types for an individual HTTP header
+# This should be a Union, but mypy doesn't pass unless it's a TypeVar.
HeaderValue = t.Union[str, t.List[str], t.Tuple[str, ...]]
-HeadersValue = t.Union['Headers', t.Mapping[str, HeaderValue], t.Sequence[t
- .Tuple[str, HeaderValue]]]
-ResponseReturnValue = t.Union[ResponseValue, t.Tuple[ResponseValue,
- HeadersValue], t.Tuple[ResponseValue, int], t.Tuple[ResponseValue, int,
- HeadersValue], 'WSGIApplication']
-ResponseClass = t.TypeVar('ResponseClass', bound='Response')
-AppOrBlueprintKey = t.Optional[str]
-AfterRequestCallable = t.Union[t.Callable[[ResponseClass], ResponseClass],
- t.Callable[[ResponseClass], t.Awaitable[ResponseClass]]]
-BeforeFirstRequestCallable = t.Union[t.Callable[[], None], t.Callable[[], t
- .Awaitable[None]]]
-BeforeRequestCallable = t.Union[t.Callable[[], t.Optional[
- ResponseReturnValue]], t.Callable[[], t.Awaitable[t.Optional[
- ResponseReturnValue]]]]
+
+# the possible types for HTTP headers
+HeadersValue = t.Union[
+ "Headers",
+ t.Mapping[str, HeaderValue],
+ t.Sequence[t.Tuple[str, HeaderValue]],
+]
+
+# The possible types returned by a route function.
+ResponseReturnValue = t.Union[
+ ResponseValue,
+ t.Tuple[ResponseValue, HeadersValue],
+ t.Tuple[ResponseValue, int],
+ t.Tuple[ResponseValue, int, HeadersValue],
+ "WSGIApplication",
+]
+
+# Allow any subclass of werkzeug.Response, such as the one from Flask,
+# as a callback argument. Using werkzeug.Response directly makes a
+# callback annotated with flask.Response fail type checking.
+ResponseClass = t.TypeVar("ResponseClass", bound="Response")
+
+AppOrBlueprintKey = t.Optional[str] # The App key is None, whereas blueprints are named
+AfterRequestCallable = t.Union[
+ t.Callable[[ResponseClass], ResponseClass],
+ t.Callable[[ResponseClass], t.Awaitable[ResponseClass]],
+]
+BeforeFirstRequestCallable = t.Union[
+ t.Callable[[], None], t.Callable[[], t.Awaitable[None]]
+]
+BeforeRequestCallable = t.Union[
+ t.Callable[[], t.Optional[ResponseReturnValue]],
+ t.Callable[[], t.Awaitable[t.Optional[ResponseReturnValue]]],
+]
ShellContextProcessorCallable = t.Callable[[], t.Dict[str, t.Any]]
-TeardownCallable = t.Union[t.Callable[[t.Optional[BaseException]], None], t
- .Callable[[t.Optional[BaseException]], t.Awaitable[None]]]
-TemplateContextProcessorCallable = t.Union[t.Callable[[], t.Dict[str, t.Any
- ]], t.Callable[[], t.Awaitable[t.Dict[str, t.Any]]]]
+TeardownCallable = t.Union[
+ t.Callable[[t.Optional[BaseException]], None],
+ t.Callable[[t.Optional[BaseException]], t.Awaitable[None]],
+]
+TemplateContextProcessorCallable = t.Union[
+ t.Callable[[], t.Dict[str, t.Any]],
+ t.Callable[[], t.Awaitable[t.Dict[str, t.Any]]],
+]
TemplateFilterCallable = t.Callable[..., t.Any]
TemplateGlobalCallable = t.Callable[..., t.Any]
TemplateTestCallable = t.Callable[..., bool]
URLDefaultCallable = t.Callable[[str, t.Dict[str, t.Any]], None]
-URLValuePreprocessorCallable = t.Callable[[t.Optional[str], t.Optional[t.
- Dict[str, t.Any]]], None]
-ErrorHandlerCallable = t.Union[t.Callable[[t.Any], ResponseReturnValue], t.
- Callable[[t.Any], t.Awaitable[ResponseReturnValue]]]
-RouteCallable = t.Union[t.Callable[..., ResponseReturnValue], t.Callable[
- ..., t.Awaitable[ResponseReturnValue]]]
+URLValuePreprocessorCallable = t.Callable[
+ [t.Optional[str], t.Optional[t.Dict[str, t.Any]]], None
+]
+
+# This should take Exception, but that either breaks typing the argument
+# with a specific exception, or decorating multiple times with different
+# exceptions (and using a union type on the argument).
+# https://github.com/pallets/flask/issues/4095
+# https://github.com/pallets/flask/issues/4295
+# https://github.com/pallets/flask/issues/4297
+ErrorHandlerCallable = t.Union[
+ t.Callable[[t.Any], ResponseReturnValue],
+ t.Callable[[t.Any], t.Awaitable[ResponseReturnValue]],
+]
+
+RouteCallable = t.Union[
+ t.Callable[..., ResponseReturnValue],
+ t.Callable[..., t.Awaitable[ResponseReturnValue]],
+]
diff --git a/src/flask/views.py b/src/flask/views.py
index 25272f32..794fdc06 100644
--- a/src/flask/views.py
+++ b/src/flask/views.py
@@ -1,11 +1,16 @@
from __future__ import annotations
+
import typing as t
+
from . import typing as ft
from .globals import current_app
from .globals import request
-F = t.TypeVar('F', bound=t.Callable[..., t.Any])
-http_method_funcs = frozenset(['get', 'post', 'head', 'options', 'delete',
- 'put', 'trace', 'patch'])
+
+F = t.TypeVar("F", bound=t.Callable[..., t.Any])
+
+http_method_funcs = frozenset(
+ ["get", "post", "head", "options", "delete", "put", "trace", "patch"]
+)
class View:
@@ -39,21 +44,48 @@ class View:
Set :attr:`init_every_request` to ``False`` for efficiency, unless
you need to store request-global data on ``self``.
"""
+
+ #: The methods this view is registered for. Uses the same default
+ #: (``["GET", "HEAD", "OPTIONS"]``) as ``route`` and
+ #: ``add_url_rule`` by default.
methods: t.ClassVar[t.Collection[str] | None] = None
+
+ #: Control whether the ``OPTIONS`` method is handled automatically.
+ #: Uses the same default (``True``) as ``route`` and
+ #: ``add_url_rule`` by default.
provide_automatic_options: t.ClassVar[bool | None] = None
+
+ #: A list of decorators to apply, in order, to the generated view
+ #: function. Remember that ``@decorator`` syntax is applied bottom
+ #: to top, so the first decorator in the list would be the bottom
+ #: decorator.
+ #:
+ #: .. versionadded:: 0.8
decorators: t.ClassVar[list[t.Callable[[F], F]]] = []
+
+ #: Create a new instance of this view class for every request by
+ #: default. If a view subclass sets this to ``False``, the same
+ #: instance is used for every request.
+ #:
+ #: A single instance is more efficient, especially if complex setup
+ #: is done during init. However, storing data on ``self`` is no
+ #: longer safe across requests, and :data:`~flask.g` should be used
+ #: instead.
+ #:
+ #: .. versionadded:: 2.2
init_every_request: t.ClassVar[bool] = True
- def dispatch_request(self) ->ft.ResponseReturnValue:
+ def dispatch_request(self) -> ft.ResponseReturnValue:
"""The actual view function behavior. Subclasses must override
this and return a valid response. Any variables from the URL
rule are passed as keyword arguments.
"""
- pass
+ raise NotImplementedError()
@classmethod
- def as_view(cls, name: str, *class_args: t.Any, **class_kwargs: t.Any
- ) ->ft.RouteCallable:
+ def as_view(
+ cls, name: str, *class_args: t.Any, **class_kwargs: t.Any
+ ) -> ft.RouteCallable:
"""Convert the class into a view function that can be registered
for a route.
@@ -69,7 +101,38 @@ class View:
.. versionchanged:: 2.2
Added the ``init_every_request`` class attribute.
"""
- pass
+ if cls.init_every_request:
+
+ def view(**kwargs: t.Any) -> ft.ResponseReturnValue:
+ self = view.view_class( # type: ignore[attr-defined]
+ *class_args, **class_kwargs
+ )
+ return current_app.ensure_sync(self.dispatch_request)(**kwargs) # type: ignore[no-any-return]
+
+ else:
+ self = cls(*class_args, **class_kwargs)
+
+ def view(**kwargs: t.Any) -> ft.ResponseReturnValue:
+ return current_app.ensure_sync(self.dispatch_request)(**kwargs) # type: ignore[no-any-return]
+
+ if cls.decorators:
+ view.__name__ = name
+ view.__module__ = cls.__module__
+ for decorator in cls.decorators:
+ view = decorator(view)
+
+ # We attach the view class to the view function for two reasons:
+ # first of all it allows us to easily figure out what class-based
+ # view this thing came from, secondly it's also used for instantiating
+ # the view class so you can actually replace it with something else
+ # for testing purposes and debugging.
+ view.view_class = cls # type: ignore
+ view.__name__ = name
+ view.__doc__ = cls.__doc__
+ view.__module__ = cls.__module__
+ view.methods = cls.methods # type: ignore
+ view.provide_automatic_options = cls.provide_automatic_options # type: ignore
+ return view
class MethodView(View):
@@ -99,15 +162,30 @@ class MethodView(View):
)
"""
- def __init_subclass__(cls, **kwargs: t.Any) ->None:
+ def __init_subclass__(cls, **kwargs: t.Any) -> None:
super().__init_subclass__(**kwargs)
- if 'methods' not in cls.__dict__:
+
+ if "methods" not in cls.__dict__:
methods = set()
+
for base in cls.__bases__:
- if getattr(base, 'methods', None):
- methods.update(base.methods)
+ if getattr(base, "methods", None):
+ methods.update(base.methods) # type: ignore[attr-defined]
+
for key in http_method_funcs:
if hasattr(cls, key):
methods.add(key.upper())
+
if methods:
cls.methods = methods
+
+ def dispatch_request(self, **kwargs: t.Any) -> ft.ResponseReturnValue:
+ meth = getattr(self, request.method.lower(), None)
+
+ # If the request method is HEAD and we don't have a handler for it
+ # retry with GET.
+ if meth is None and request.method == "HEAD":
+ meth = getattr(self, "get", None)
+
+ assert meth is not None, f"Unimplemented method {request.method!r}"
+ return current_app.ensure_sync(meth)(**kwargs) # type: ignore[no-any-return]
diff --git a/src/flask/wrappers.py b/src/flask/wrappers.py
index e086f271..c1eca807 100644
--- a/src/flask/wrappers.py
+++ b/src/flask/wrappers.py
@@ -1,13 +1,17 @@
from __future__ import annotations
+
import typing as t
+
from werkzeug.exceptions import BadRequest
from werkzeug.exceptions import HTTPException
from werkzeug.wrappers import Request as RequestBase
from werkzeug.wrappers import Response as ResponseBase
+
from . import json
from .globals import current_app
from .helpers import _split_blueprint_path
-if t.TYPE_CHECKING:
+
+if t.TYPE_CHECKING: # pragma: no cover
from werkzeug.routing import Rule
@@ -23,18 +27,41 @@ class Request(RequestBase):
provides all of the attributes Werkzeug defines plus a few Flask
specific ones.
"""
+
json_module: t.Any = json
+
+ #: The internal URL rule that matched the request. This can be
+ #: useful to inspect which methods are allowed for the URL from
+ #: a before/after handler (``request.url_rule.methods``) etc.
+ #: Though if the request's method was invalid for the URL rule,
+ #: the valid list is available in ``routing_exception.valid_methods``
+ #: instead (an attribute of the Werkzeug exception
+ #: :exc:`~werkzeug.exceptions.MethodNotAllowed`)
+ #: because the request was never internally bound.
+ #:
+ #: .. versionadded:: 0.6
url_rule: Rule | None = None
+
+ #: A dict of view arguments that matched the request. If an exception
+ #: happened when matching, this will be ``None``.
view_args: dict[str, t.Any] | None = None
+
+ #: If matching the URL failed, this is the exception that will be
+ #: raised / was raised as part of the request handling. This is
+ #: usually a :exc:`~werkzeug.exceptions.NotFound` exception or
+ #: something similar.
routing_exception: HTTPException | None = None
@property
- def max_content_length(self) ->(int | None):
+ def max_content_length(self) -> int | None: # type: ignore[override]
"""Read-only view of the ``MAX_CONTENT_LENGTH`` config key."""
- pass
+ if current_app:
+ return current_app.config["MAX_CONTENT_LENGTH"] # type: ignore[no-any-return]
+ else:
+ return None
@property
- def endpoint(self) ->(str | None):
+ def endpoint(self) -> str | None:
"""The endpoint that matched the request URL.
This will be ``None`` if matching failed or has not been
@@ -43,10 +70,13 @@ class Request(RequestBase):
This in combination with :attr:`view_args` can be used to
reconstruct the same URL or a modified URL.
"""
- pass
+ if self.url_rule is not None:
+ return self.url_rule.endpoint
+
+ return None
@property
- def blueprint(self) ->(str | None):
+ def blueprint(self) -> str | None:
"""The registered name of the current blueprint.
This will be ``None`` if the endpoint is not part of a
@@ -57,10 +87,15 @@ class Request(RequestBase):
created with. It may have been nested, or registered with a
different name.
"""
- pass
+ endpoint = self.endpoint
+
+ if endpoint is not None and "." in endpoint:
+ return endpoint.rpartition(".")[0]
+
+ return None
@property
- def blueprints(self) ->list[str]:
+ def blueprints(self) -> list[str]:
"""The registered names of the current blueprint upwards through
parent blueprints.
@@ -69,7 +104,36 @@ class Request(RequestBase):
.. versionadded:: 2.0.1
"""
- pass
+ name = self.blueprint
+
+ if name is None:
+ return []
+
+ return _split_blueprint_path(name)
+
+ def _load_form_data(self) -> None:
+ super()._load_form_data()
+
+ # In debug mode we're replacing the files multidict with an ad-hoc
+ # subclass that raises a different error for key errors.
+ if (
+ current_app
+ and current_app.debug
+ and self.mimetype != "multipart/form-data"
+ and not self.files
+ ):
+ from .debughelpers import attach_enctype_error_multidict
+
+ attach_enctype_error_multidict(self)
+
+ def on_json_loading_failed(self, e: ValueError | None) -> t.Any:
+ try:
+ return super().on_json_loading_failed(e)
+ except BadRequest as e:
+ if current_app and current_app.debug:
+ raise
+
+ raise BadRequest() from e
class Response(ResponseBase):
@@ -89,15 +153,22 @@ class Response(ResponseBase):
Added :attr:`max_cookie_size`.
"""
- default_mimetype: str | None = 'text/html'
+
+ default_mimetype: str | None = "text/html"
+
json_module = json
+
autocorrect_location_header = False
@property
- def max_cookie_size(self) ->int:
+ def max_cookie_size(self) -> int: # type: ignore
"""Read-only view of the :data:`MAX_COOKIE_SIZE` config key.
See :attr:`~werkzeug.wrappers.Response.max_cookie_size` in
Werkzeug's docs.
"""
- pass
+ if current_app:
+ return current_app.config["MAX_COOKIE_SIZE"] # type: ignore[no-any-return]
+
+ # return Werkzeug's default when not in an app context
+ return super().max_cookie_size