This commit is contained in:
2025-05-22 21:22:15 +02:00
parent 3d57f842f9
commit 97cb9c8703
156 changed files with 1205 additions and 6603 deletions

View File

@@ -1,7 +1,9 @@
from __future__ import annotations
import typing as t
from . import json as json
from .app import Flask as Flask
from .app import Request as Request
from .app import Response as Response
from .blueprints import Blueprint as Blueprint
from .config import Config as Config
from .ctx import after_this_request as after_this_request
@@ -37,66 +39,22 @@ from .templating import render_template as render_template
from .templating import render_template_string as render_template_string
from .templating import stream_template as stream_template
from .templating import stream_template_string as stream_template_string
__version__ = "2.3.3"
from .wrappers import Request as Request
from .wrappers import Response as Response
def __getattr__(name):
if name == "_app_ctx_stack":
import warnings
from .globals import __app_ctx_stack
warnings.warn(
"'_app_ctx_stack' is deprecated and will be removed in Flask 2.4.",
DeprecationWarning,
stacklevel=2,
)
return __app_ctx_stack
if name == "_request_ctx_stack":
import warnings
from .globals import __request_ctx_stack
warnings.warn(
"'_request_ctx_stack' is deprecated and will be removed in Flask 2.4.",
DeprecationWarning,
stacklevel=2,
)
return __request_ctx_stack
if name == "escape":
import warnings
from markupsafe import escape
warnings.warn(
"'flask.escape' is deprecated and will be removed in Flask 2.4. Import"
" 'markupsafe.escape' instead.",
DeprecationWarning,
stacklevel=2,
)
return escape
if name == "Markup":
import warnings
from markupsafe import Markup
warnings.warn(
"'flask.Markup' is deprecated and will be removed in Flask 2.4. Import"
" 'markupsafe.Markup' instead.",
DeprecationWarning,
stacklevel=2,
)
return Markup
if name == "signals_available":
def __getattr__(name: str) -> t.Any:
if name == "__version__":
import importlib.metadata
import warnings
warnings.warn(
"'signals_available' is deprecated and will be removed in Flask 2.4."
" Signals are always available",
"The '__version__' attribute is deprecated and will be removed in"
" Flask 3.1. Use feature detection or"
" 'importlib.metadata.version(\"flask\")' instead.",
DeprecationWarning,
stacklevel=2,
)
return True
return importlib.metadata.version("flask")
raise AttributeError(name)

File diff suppressed because it is too large Load Diff

View File

@@ -2,625 +2,90 @@ from __future__ import annotations
import os
import typing as t
from collections import defaultdict
from functools import update_wrapper
from datetime import timedelta
from . import typing as ft
from .scaffold import _endpoint_from_view_func
from .scaffold import _sentinel
from .scaffold import Scaffold
from .scaffold import setupmethod
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 # noqa
if t.TYPE_CHECKING: # pragma: no cover
from .app import Flask
DeferredSetupFunction = t.Callable[["BlueprintSetupState"], t.Callable]
T_after_request = t.TypeVar("T_after_request", bound=ft.AfterRequestCallable)
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
)
from .wrappers import Response
class BlueprintSetupState:
"""Temporary holder object for registering a blueprint with the
application. An instance of this class is created by the
:meth:`~flask.Blueprint.make_setup_state` method and later passed
to all register callback functions.
"""
class Blueprint(SansioBlueprint):
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.
def __init__(
self,
blueprint: Blueprint,
app: Flask,
options: t.Any,
first_registration: bool,
) -> None:
#: a reference to the current application
self.app = app
By default, this returns :data:`SEND_FILE_MAX_AGE_DEFAULT` from
the configuration of :data:`~flask.current_app`. This defaults
to ``None``, which tells the browser to use conditional requests
instead of a timed cache, which is usually preferable.
#: a reference to the blueprint that created this setup state.
self.blueprint = blueprint
Note this is a duplicate of the same method in the Flask
class.
#: a dictionary with all options that were passed to the
#: :meth:`~flask.Flask.register_blueprint` method.
self.options = options
.. versionchanged:: 2.0
The default configuration is ``None`` instead of 12 hours.
#: 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")
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")
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", "")
#: 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: t.Callable | 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.
.. versionadded:: 0.9
"""
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"))
value = current_app.config["SEND_FILE_MAX_AGE_DEFAULT"]
self.app.add_url_rule(
rule,
f"{self.name_prefix}.{self.name}.{endpoint}".lstrip("."),
view_func,
defaults=defaults,
**options,
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
set.
Note this is a duplicate of the same method in the Flask
class.
.. versionadded:: 0.5
"""
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]:
"""Open a resource file relative to :attr:`root_path` for
reading.
class Blueprint(Scaffold):
"""Represents a blueprint, a collection of routes and other
app-related functions that can be registered on a real application
later.
For example, if the file ``schema.sql`` is next to the file
``app.py`` where the ``Flask`` app is defined, it can be opened
with:
A blueprint is an object that allows defining application functions
without requiring an application object ahead of time. It uses the
same decorators as :class:`~flask.Flask`, but defers the need for an
application by recording them for later registration.
.. code-block:: python
Decorating a function with a blueprint creates a deferred function
that is called with :class:`~flask.blueprints.BlueprintSetupState`
when the blueprint is registered on an application.
with app.open_resource("schema.sql") as f:
conn.executescript(f.read())
See :doc:`/blueprints` for more information.
:param resource: Path to the resource relative to
:attr:`root_path`.
:param mode: Open the file in this mode. Only reading is
supported, valid values are "r" (or "rt") and "rb".
:param name: The name of the blueprint. Will be prepended to each
endpoint name.
:param import_name: The name of the blueprint package, usually
``__name__``. This helps locate the ``root_path`` for the
blueprint.
:param static_folder: A folder with static files that should be
served by the blueprint's static route. The path is relative to
the blueprint's root path. Blueprint static files are disabled
by default.
:param static_url_path: The url to serve static files from.
Defaults to ``static_folder``. If the blueprint does not have
a ``url_prefix``, the app's static route will take precedence,
and the blueprint's static files won't be accessible.
:param template_folder: A folder with templates that should be added
to the app's template search path. The path is relative to the
blueprint's root path. Blueprint templates are disabled by
default. Blueprint templates have a lower precedence than those
in the app's templates folder.
:param url_prefix: A path to prepend to all of the blueprint's URLs,
to make them distinct from the rest of the app's routes.
:param subdomain: A subdomain that blueprint routes will match on by
default.
:param url_defaults: A dict of default values that blueprint routes
will receive by default.
:param root_path: By default, the blueprint will automatically set
this based on ``import_name``. In certain situations this
automatic detection can fail, so the path can be specified
manually instead.
Note this is a duplicate of the same method in the Flask
class.
.. versionchanged:: 1.1.0
Blueprints have a ``cli`` group to register nested CLI commands.
The ``cli_group`` parameter controls the name of the group under
the ``flask`` command.
.. versionadded:: 0.7
"""
_got_registered_once = False
def __init__(
self,
name: str,
import_name: str,
static_folder: str | os.PathLike | None = None,
static_url_path: str | None = None,
template_folder: str | os.PathLike | None = None,
url_prefix: str | None = None,
subdomain: str | None = None,
url_defaults: dict | None = None,
root_path: str | None = None,
cli_group: str | None = _sentinel, # type: ignore
):
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:
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]] = []
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: t.Callable) -> 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.
"""
self.deferred_functions.append(func)
if mode not in {"r", "rt", "rb"}:
raise ValueError("Resources can only be opened for reading.")
@setupmethod
def record_once(self, func: t.Callable) -> 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.
"""
def wrapper(state: BlueprintSetupState) -> None:
if state.first_registration:
func(state)
self.record(update_wrapper(wrapper, func))
def make_setup_state(
self, app: Flask, options: dict, 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.
"""
return BlueprintSetupState(self, app, options, first_registration)
@setupmethod
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.
.. versionchanged:: 2.0.1
The ``name`` option can be used to change the (pre-dotted)
name the blueprint is registered with. This allows the same
blueprint to be registered multiple times with unique names
for ``url_for``.
.. versionadded:: 2.0
"""
if blueprint is self:
raise ValueError("Cannot register a blueprint on itself")
self._blueprints.append((blueprint, options))
def register(self, app: Flask, options: dict) -> 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
each :meth:`record` callback with it.
:param app: The application this blueprint is being registered
with.
:param options: Keyword arguments forwarded from
:meth:`~Flask.register_blueprint`.
.. versionchanged:: 2.3
Nested blueprints now correctly apply subdomains.
.. versionchanged:: 2.1
Registering the same blueprint with the same name multiple
times is an error.
.. versionchanged:: 2.0.1
Nested blueprints are registered with their dotted name.
This allows different blueprints with the same name to be
nested at different locations.
.. versionchanged:: 2.0.1
The ``name`` option can be used to change the (pre-dotted)
name the blueprint is registered with. This allows the same
blueprint to be registered multiple times with unique names
for ``url_for``.
"""
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,
endpoint="static",
)
# Merge blueprint data into parent.
if first_bp_registration or first_name_registration:
def extend(bp_dict, parent_dict):
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)
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)
@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:
"""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.
"""
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]:
"""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.
"""
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:
"""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`.
:param name: the optional name of the filter, otherwise the
function name will be used.
"""
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]:
"""Register a template test, available in any template rendered by the
application. Equivalent to :meth:`.Flask.template_test`.
.. versionadded:: 0.10
:param name: the optional name of the test, otherwise the
function name will be used.
"""
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:
"""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`.
.. versionadded:: 0.10
:param name: the optional name of the test, otherwise the
function name will be used.
"""
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]:
"""Register a template global, available in any template rendered by the
application. Equivalent to :meth:`.Flask.template_global`.
.. versionadded:: 0.10
:param name: the optional name of the global, otherwise the
function name will be used.
"""
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:
"""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`.
.. versionadded:: 0.10
:param name: the optional name of the global, otherwise the
function name will be used.
"""
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:
"""Like :meth:`before_request`, but before every request, not only those handled
by the blueprint. Equivalent to :meth:`.Flask.before_request`.
"""
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:
"""Like :meth:`after_request`, but after every request, not only those handled
by the blueprint. Equivalent to :meth:`.Flask.after_request`.
"""
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:
"""Like :meth:`teardown_request`, but after every request, not only those
handled by the blueprint. Equivalent to :meth:`.Flask.teardown_request`.
"""
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:
"""Like :meth:`context_processor`, but for templates rendered by every view, not
only by the blueprint. Equivalent to :meth:`.Flask.context_processor`.
"""
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]:
"""Like :meth:`errorhandler`, but for every request, not only those handled by
the blueprint. Equivalent to :meth:`.Flask.errorhandler`.
"""
def decorator(f: T_error_handler) -> T_error_handler:
self.record_once(lambda s: s.app.errorhandler(code)(f))
return f
return decorator
@setupmethod
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`.
"""
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:
"""Like :meth:`url_defaults`, but for every request, not only those handled by
the blueprint. Equivalent to :meth:`.Flask.url_defaults`.
"""
self.record_once(
lambda s: s.app.url_default_functions.setdefault(None, []).append(f)
)
return f
return open(os.path.join(self.root_path, resource), mode)

View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import ast
import collections.abc as cabc
import importlib.metadata
import inspect
import os
@@ -11,6 +12,7 @@ import traceback
import typing as t
from functools import update_wrapper
from operator import itemgetter
from types import ModuleType
import click
from click.core import ParameterSource
@@ -23,6 +25,12 @@ 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
@@ -30,7 +38,7 @@ class NoAppException(click.UsageError):
"""Raised if an application cannot be found or loaded."""
def find_best_app(module):
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.
"""
@@ -83,7 +91,7 @@ def find_best_app(module):
)
def _called_with_wrong_args(f):
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.
@@ -109,7 +117,7 @@ def _called_with_wrong_args(f):
del tb
def find_app_by_string(module, app_name):
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.
"""
@@ -140,7 +148,11 @@ def find_app_by_string(module, app_name):
# 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}
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.
@@ -185,7 +197,7 @@ def find_app_by_string(module, app_name):
)
def prepare_import(path):
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.
"""
@@ -214,13 +226,29 @@ def prepare_import(path):
return ".".join(module_name[::-1])
def locate_app(module_name, app_name, raise_if_not_found=True):
@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:
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()}"
@@ -228,7 +256,7 @@ def locate_app(module_name, app_name, raise_if_not_found=True):
elif raise_if_not_found:
raise NoAppException(f"Could not import {module_name!r}.") from None
else:
return
return None
module = sys.modules[module_name]
@@ -238,7 +266,7 @@ def locate_app(module_name, app_name, raise_if_not_found=True):
return find_app_by_string(module, app_name)
def get_version(ctx, param, value):
def get_version(ctx: click.Context, param: click.Parameter, value: t.Any) -> None:
if not value or ctx.resilient_parsing:
return
@@ -299,7 +327,7 @@ class ScriptInfo:
return self._loaded_app
if self.create_app is not None:
app = self.create_app()
app: Flask | None = self.create_app()
else:
if self.app_import_path:
path, name = (
@@ -312,10 +340,10 @@ class ScriptInfo:
import_name = prepare_import(path)
app = locate_app(import_name, None, raise_if_not_found=False)
if app:
if app is not None:
break
if not app:
if app is None:
raise NoAppException(
"Could not locate a Flask application. Use the"
" 'flask --app' option, 'FLASK_APP' environment"
@@ -334,8 +362,10 @@ class ScriptInfo:
pass_script_info = click.make_pass_decorator(ScriptInfo, ensure=True)
F = t.TypeVar("F", bound=t.Callable[..., t.Any])
def with_appcontext(f):
def with_appcontext(f: F) -> F:
"""Wraps a callback so that it's guaranteed to be executed with the
script's application context.
@@ -350,14 +380,14 @@ def with_appcontext(f):
"""
@click.pass_context
def decorator(__ctx, *args, **kwargs):
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())
app = ctx.ensure_object(ScriptInfo).load_app()
ctx.with_resource(app.app_context())
return __ctx.invoke(f, *args, **kwargs)
return ctx.invoke(f, *args, **kwargs)
return update_wrapper(decorator, f)
return update_wrapper(decorator, f) # type: ignore[return-value]
class AppGroup(click.Group):
@@ -368,27 +398,31 @@ class AppGroup(click.Group):
Not to be confused with :class:`FlaskGroup`.
"""
def command(self, *args, **kwargs):
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``.
"""
wrap_for_ctx = kwargs.pop("with_appcontext", True)
def decorator(f):
def decorator(f: t.Callable[..., t.Any]) -> click.Command:
if wrap_for_ctx:
f = with_appcontext(f)
return click.Group.command(self, *args, **kwargs)(f)
return super(AppGroup, self).command(*args, **kwargs)(f) # type: ignore[no-any-return]
return decorator
def group(self, *args, **kwargs):
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`.
"""
kwargs.setdefault("cls", AppGroup)
return click.Group.group(self, *args, **kwargs)
return super().group(*args, **kwargs) # type: ignore[no-any-return]
def _set_app(ctx: click.Context, param: click.Option, value: str | None) -> str | None:
@@ -545,7 +579,7 @@ class FlaskGroup(AppGroup):
self._loaded_plugin_commands = False
def _load_plugin_commands(self):
def _load_plugin_commands(self) -> None:
if self._loaded_plugin_commands:
return
@@ -562,7 +596,7 @@ class FlaskGroup(AppGroup):
self._loaded_plugin_commands = True
def get_command(self, ctx, name):
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.
@@ -584,12 +618,12 @@ class FlaskGroup(AppGroup):
# Push an app context for the loaded app unless it is already
# active somehow. This makes the context available to parameter
# and command callbacks without needing @with_appcontext.
if not current_app or current_app._get_current_object() is not app:
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):
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))
@@ -645,14 +679,14 @@ class FlaskGroup(AppGroup):
return super().parse_args(ctx, args)
def _path_is_ancestor(path, other):
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``."""
return os.path.join(path, other[len(path) :].lstrip(os.sep)) == other
def load_dotenv(path: str | os.PathLike | 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
@@ -713,7 +747,7 @@ def load_dotenv(path: str | os.PathLike | None = None) -> bool:
return loaded # True if at least one file was located and loaded.
def show_server_banner(debug, app_import_path):
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.
"""
@@ -735,10 +769,12 @@ class CertParamType(click.ParamType):
name = "path"
def __init__(self):
def __init__(self) -> None:
self.path_type = click.Path(exists=True, dir_okay=False, resolve_path=True)
def convert(self, value, param, ctx):
def convert(
self, value: t.Any, param: click.Parameter | None, ctx: click.Context | None
) -> t.Any:
try:
import ssl
except ImportError:
@@ -773,7 +809,7 @@ class CertParamType(click.ParamType):
raise
def _validate_key(ctx, param, value):
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.
"""
@@ -795,7 +831,9 @@ def _validate_key(ctx, param, value):
if is_context:
raise click.BadParameter(
'When "--cert" is an SSLContext object, "--key is not used.', ctx, param
'When "--cert" is an SSLContext object, "--key" is not used.',
ctx,
param,
)
if not cert:
@@ -816,8 +854,11 @@ class SeparatedPathType(click.Path):
validated as a :class:`click.Path` type.
"""
def convert(self, value, param, ctx):
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]
@@ -876,16 +917,16 @@ class SeparatedPathType(click.Path):
)
@pass_script_info
def run_command(
info,
host,
port,
reload,
debugger,
with_threads,
cert,
extra_files,
exclude_patterns,
):
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
@@ -895,7 +936,7 @@ def run_command(
option.
"""
try:
app = info.load_app()
app: WSGIApplication = info.load_app()
except Exception as e:
if is_running_from_reloader():
# When reloading, print out the error immediately, but raise
@@ -903,7 +944,9 @@ def run_command(
traceback.print_exc()
err = e
def app(environ, start_response):
def app(
environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable[bytes]:
raise err from None
else:
@@ -954,7 +997,7 @@ def shell_command() -> None:
f"App: {current_app.import_name}\n"
f"Instance: {current_app.instance_path}"
)
ctx: dict = {}
ctx: dict[str, t.Any] = {}
# Support the regular Python interpreter startup script if someone
# is using it.

View File

@@ -8,27 +8,48 @@ import typing as t
from werkzeug.utils import import_string
if t.TYPE_CHECKING:
import typing_extensions as te
class ConfigAttribute:
from .sansio.app import App
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 | 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
def __get__(self, obj: t.Any, owner: t.Any = None) -> t.Any:
@t.overload
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 | 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: t.Any, 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.
@@ -73,7 +94,9 @@ class Config(dict):
"""
def __init__(
self, root_path: str | os.PathLike, defaults: dict | None = None
self,
root_path: str | os.PathLike[str],
defaults: dict[str, t.Any] | None = None,
) -> None:
super().__init__(defaults or {})
self.root_path = root_path
@@ -166,7 +189,9 @@ class Config(dict):
return True
def from_pyfile(self, filename: str | os.PathLike, silent: bool = False) -> bool:
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.
@@ -235,8 +260,8 @@ class Config(dict):
def from_file(
self,
filename: str | os.PathLike,
load: t.Callable[[t.IO[t.Any]], t.Mapping],
filename: str | os.PathLike[str],
load: t.Callable[[t.IO[t.Any]], t.Mapping[str, t.Any]],
silent: bool = False,
text: bool = True,
) -> bool:

View File

@@ -15,6 +15,8 @@ from .signals import appcontext_popped
from .signals import appcontext_pushed
if t.TYPE_CHECKING: # pragma: no cover
from _typeshed.wsgi import WSGIEnvironment
from .app import Flask
from .sessions import SessionMixin
from .wrappers import Request
@@ -112,7 +114,9 @@ class _AppCtxGlobals:
return object.__repr__(self)
def after_this_request(f: ft.AfterRequestCallable) -> ft.AfterRequestCallable:
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.
@@ -145,7 +149,10 @@ def after_this_request(f: ft.AfterRequestCallable) -> ft.AfterRequestCallable:
return f
def copy_current_request_context(f: t.Callable) -> t.Callable:
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
@@ -179,11 +186,11 @@ def copy_current_request_context(f: t.Callable) -> t.Callable:
ctx = ctx.copy()
def wrapper(*args, **kwargs):
with ctx:
return ctx.app.ensure_sync(f)(*args, **kwargs)
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]
return update_wrapper(wrapper, f)
return update_wrapper(wrapper, f) # type: ignore[return-value]
def has_request_context() -> bool:
@@ -239,7 +246,7 @@ class AppContext:
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] = []
self._cv_tokens: list[contextvars.Token[AppContext]] = []
def push(self) -> None:
"""Binds the app context to the current context."""
@@ -302,7 +309,7 @@ class RequestContext:
def __init__(
self,
app: Flask,
environ: dict,
environ: WSGIEnvironment,
request: Request | None = None,
session: SessionMixin | None = None,
) -> None:
@@ -321,9 +328,11 @@ class RequestContext:
# 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] = []
self._after_request_functions: list[ft.AfterRequestCallable[t.Any]] = []
self._cv_tokens: list[tuple[contextvars.Token, AppContext | None]] = []
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.

View File

@@ -2,9 +2,16 @@ from __future__ import annotations
import typing as t
from .app import Flask
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
class UnexpectedUnicodeError(AssertionError, UnicodeError):
@@ -18,7 +25,7 @@ class DebugFilesKeyError(KeyError, AssertionError):
provide a better error message than just a generic KeyError/BadRequest.
"""
def __init__(self, request, key):
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"
@@ -36,7 +43,7 @@ class DebugFilesKeyError(KeyError, AssertionError):
)
self.msg = "".join(buf)
def __str__(self):
def __str__(self) -> str:
return self.msg
@@ -47,8 +54,9 @@ class FormDataRoutingRedirect(AssertionError):
307 or 308.
"""
def __init__(self, request):
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"
f" a redirect to the canonical URL '{exc.new_url}'."
@@ -70,7 +78,7 @@ class FormDataRoutingRedirect(AssertionError):
super().__init__("".join(buf))
def attach_enctype_error_multidict(request):
def attach_enctype_error_multidict(request: Request) -> None:
"""Patch ``request.files.__getitem__`` to raise a descriptive error
about ``enctype=multipart/form-data``.
@@ -79,8 +87,8 @@ def attach_enctype_error_multidict(request):
"""
oldcls = request.files.__class__
class newcls(oldcls):
def __getitem__(self, key):
class newcls(oldcls): # type: ignore[valid-type, misc]
def __getitem__(self, key: str) -> t.Any:
try:
return super().__getitem__(key)
except KeyError as e:
@@ -96,7 +104,7 @@ def attach_enctype_error_multidict(request):
request.files.__class__ = newcls
def _dump_loader_info(loader) -> t.Generator:
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("_"):
@@ -113,7 +121,17 @@ def _dump_loader_info(loader) -> t.Generator:
yield f"{key}: {value!r}"
def explain_template_loading_attempts(app: Flask, template, attempts) -> None:
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"""
info = [f"Locating template {template!r}:"]
total_found = 0
@@ -122,7 +140,7 @@ def explain_template_loading_attempts(app: Flask, template, attempts) -> None:
blueprint = request_ctx.request.blueprint
for idx, (loader, srcobj, triple) in enumerate(attempts):
if isinstance(srcobj, Flask):
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})"

View File

@@ -14,25 +14,6 @@ if t.TYPE_CHECKING: # pragma: no cover
from .wrappers import Request
class _FakeStack:
def __init__(self, name: str, cv: ContextVar[t.Any]) -> None:
self.name = name
self.cv = cv
@property
def top(self) -> t.Any | None:
import warnings
warnings.warn(
f"'_{self.name}_ctx_stack' is deprecated and will be removed in Flask 2.4."
f" Use 'g' to store data, or '{self.name}_ctx' to access the current"
" context.",
DeprecationWarning,
stacklevel=2,
)
return self.cv.get(None)
_no_app_msg = """\
Working outside of application context.
@@ -41,7 +22,6 @@ 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_stack = _FakeStack("app", _cv_app)
app_ctx: AppContext = LocalProxy( # type: ignore[assignment]
_cv_app, unbound_message=_no_app_msg
)
@@ -60,7 +40,6 @@ 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_stack = _FakeStack("request", _cv_request)
request_ctx: RequestContext = LocalProxy( # type: ignore[assignment]
_cv_request, unbound_message=_no_req_msg
)
@@ -70,27 +49,3 @@ request: Request = LocalProxy( # type: ignore[assignment]
session: SessionMixin = LocalProxy( # type: ignore[assignment]
_cv_request, "session", unbound_message=_no_req_msg
)
def __getattr__(name: str) -> t.Any:
if name == "_app_ctx_stack":
import warnings
warnings.warn(
"'_app_ctx_stack' is deprecated and will be removed in Flask 2.4.",
DeprecationWarning,
stacklevel=2,
)
return __app_ctx_stack
if name == "_request_ctx_stack":
import warnings
warnings.warn(
"'_request_ctx_stack' is deprecated and will be removed in Flask 2.4.",
DeprecationWarning,
stacklevel=2,
)
return __request_ctx_stack
raise AttributeError(name)

View File

@@ -2,18 +2,16 @@ from __future__ import annotations
import importlib.util
import os
import socket
import sys
import typing as t
import warnings
from datetime import datetime
from functools import lru_cache
from functools import update_wrapper
from threading import RLock
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
@@ -23,7 +21,6 @@ from .globals import session
from .signals import message_flashed
if t.TYPE_CHECKING: # pragma: no cover
from werkzeug.wrappers import Response as BaseResponse
from .wrappers import Response
@@ -51,9 +48,7 @@ def get_load_dotenv(default: bool = True) -> bool:
def stream_with_context(
generator_or_function: (
t.Iterator[t.AnyStr] | t.Callable[..., t.Iterator[t.AnyStr]]
)
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
@@ -89,16 +84,16 @@ def stream_with_context(
.. versionadded:: 0.9
"""
try:
gen = iter(generator_or_function) # type: ignore
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
gen = generator_or_function(*args, **kwargs) # type: ignore[operator]
return stream_with_context(gen)
return update_wrapper(decorator, generator_or_function) # type: ignore
return update_wrapper(decorator, generator_or_function) # type: ignore[arg-type]
def generator() -> t.Generator:
def generator() -> t.Iterator[t.AnyStr | None]:
ctx = _cv_request.get(None)
if ctx is None:
raise RuntimeError(
@@ -126,7 +121,7 @@ def stream_with_context(
# real generator is executed.
wrapped_g = generator()
next(wrapped_g)
return wrapped_g
return wrapped_g # type: ignore[return-value]
def make_response(*args: t.Any) -> Response:
@@ -175,7 +170,7 @@ def make_response(*args: t.Any) -> Response:
return current_app.response_class()
if len(args) == 1:
args = args[0]
return current_app.make_response(args) # type: ignore
return current_app.make_response(args)
def url_for(
@@ -391,7 +386,7 @@ def _prepare_send_file_kwargs(**kwargs: t.Any) -> dict[str, t.Any]:
def send_file(
path_or_file: os.PathLike | str | t.BinaryIO,
path_or_file: os.PathLike[t.AnyStr] | str | t.BinaryIO,
mimetype: str | None = None,
as_attachment: bool = False,
download_name: str | None = None,
@@ -492,7 +487,7 @@ def send_file(
.. versionchanged:: 0.7
MIME guessing and etag support for file-like objects was
deprecated because it was unreliable. Pass a filename if you are
removed because it was unreliable. Pass a filename if you are
able to, otherwise attach an etag yourself.
.. versionchanged:: 0.5
@@ -517,8 +512,8 @@ def send_file(
def send_from_directory(
directory: os.PathLike | str,
path: os.PathLike | str,
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`.
@@ -613,82 +608,7 @@ def get_root_path(import_name: str) -> str:
)
# filepath is import_name.py for a module, or __init__.py for a package.
return os.path.dirname(os.path.abspath(filepath))
class locked_cached_property(werkzeug.utils.cached_property):
"""A :func:`property` that is only evaluated once. Like
:class:`werkzeug.utils.cached_property` except access uses a lock
for thread safety.
.. deprecated:: 2.3
Will be removed in Flask 2.4. Use a lock inside the decorated function if
locking is needed.
.. versionchanged:: 2.0
Inherits from Werkzeug's ``cached_property`` (and ``property``).
"""
def __init__(
self,
fget: t.Callable[[t.Any], t.Any],
name: str | None = None,
doc: str | None = None,
) -> None:
import warnings
warnings.warn(
"'locked_cached_property' is deprecated and will be removed in Flask 2.4."
" Use a lock inside the decorated function if locking is needed.",
DeprecationWarning,
stacklevel=2,
)
super().__init__(fget, name=name, doc=doc)
self.lock = RLock()
def __get__(self, obj: object, type: type = None) -> t.Any: # type: ignore
if obj is None:
return self
with self.lock:
return super().__get__(obj, type=type)
def __set__(self, obj: object, value: t.Any) -> None:
with self.lock:
super().__set__(obj, value)
def __delete__(self, obj: object) -> None:
with self.lock:
super().__delete__(obj)
def is_ip(value: str) -> bool:
"""Determine if the given string is an IP address.
:param value: value to check
:type value: str
:return: True if string is an IP address
:rtype: bool
.. deprecated:: 2.3
Will be removed in Flask 2.4.
"""
warnings.warn(
"The 'is_ip' function is deprecated and will be removed in Flask 2.4.",
DeprecationWarning,
stacklevel=2,
)
for family in (socket.AF_INET, socket.AF_INET6):
try:
socket.inet_pton(family, value)
except OSError:
pass
else:
return True
return False
return os.path.dirname(os.path.abspath(filepath)) # type: ignore[no-any-return]
@lru_cache(maxsize=None)

View File

@@ -167,4 +167,4 @@ def jsonify(*args: t.Any, **kwargs: t.Any) -> Response:
.. versionadded:: 0.2
"""
return current_app.json.response(*args, **kwargs)
return current_app.json.response(*args, **kwargs) # type: ignore[return-value]

View File

@@ -11,8 +11,9 @@ from datetime import date
from werkzeug.http import http_date
if t.TYPE_CHECKING: # pragma: no cover
from ..app import Flask
from ..wrappers import Response
from werkzeug.sansio.response import Response
from ..sansio.app import App
class JSONProvider:
@@ -34,8 +35,8 @@ class JSONProvider:
.. versionadded:: 2.2
"""
def __init__(self, app: Flask) -> None:
self._app = weakref.proxy(app)
def __init__(self, app: App) -> None:
self._app: App = weakref.proxy(app)
def dumps(self, obj: t.Any, **kwargs: t.Any) -> str:
"""Serialize data as JSON.
@@ -134,9 +135,7 @@ class DefaultJSONProvider(JSONProvider):
method) will call the ``__html__`` method to get a string.
"""
default: t.Callable[[t.Any], t.Any] = staticmethod(
_default
) # type: ignore[assignment]
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``.

View File

@@ -61,9 +61,9 @@ class JSONTag:
__slots__ = ("serializer",)
#: The tag to mark the serialized object with. If ``None``, this tag is
#: The tag to mark the serialized object with. If empty, this tag is
#: only used as an intermediate step during tagging.
key: str | None = None
key: str = ""
def __init__(self, serializer: TaggedJSONSerializer) -> None:
"""Create a tagger for the given serializer."""
@@ -83,7 +83,7 @@ class JSONTag:
will already be removed."""
raise NotImplementedError
def tag(self, value: t.Any) -> 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."""
return {self.key: self.to_json(value)}
@@ -274,7 +274,7 @@ class TaggedJSONSerializer:
tag = tag_class(self)
key = tag.key
if key is not None:
if key:
if not force and key in self.tags:
raise KeyError(f"Tag '{key}' is already registered.")
@@ -285,7 +285,7 @@ class TaggedJSONSerializer:
else:
self.order.insert(index, tag)
def tag(self, value: t.Any) -> dict[str, t.Any]:
def tag(self, value: t.Any) -> t.Any:
"""Convert a value to a tagged representation if necessary."""
for tag in self.order:
if tag.check(value):
@@ -305,10 +305,22 @@ class TaggedJSONSerializer:
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:
"""Tag the value and dump it to a compact JSON string."""
return dumps(self.tag(value), separators=(",", ":"))
def loads(self, value: str) -> t.Any:
"""Load data from a JSON string and deserialized any tagged objects."""
return loads(value, object_hook=self.untag)
return self._untag_scan(loads(value))

View File

@@ -9,7 +9,7 @@ from werkzeug.local import LocalProxy
from .globals import request
if t.TYPE_CHECKING: # pragma: no cover
from .app import Flask
from .sansio.app import App
@LocalProxy
@@ -22,7 +22,10 @@ def wsgi_errors_stream() -> t.TextIO:
can't import this directly, you can refer to it as
``ext://flask.logging.wsgi_errors_stream``.
"""
return request.environ["wsgi.errors"] if request else sys.stderr
if request:
return request.environ["wsgi.errors"] # type: ignore[no-any-return]
return sys.stderr
def has_level_handler(logger: logging.Logger) -> bool:
@@ -52,7 +55,7 @@ default_handler.setFormatter(
)
def create_logger(app: Flask) -> logging.Logger:
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

View File

@@ -1,873 +0,0 @@
from __future__ import annotations
import importlib.util
import os
import pathlib
import sys
import typing as t
from collections import defaultdict
from datetime import timedelta
from functools import update_wrapper
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 .cli import AppGroup
from .globals import current_app
from .helpers import get_root_path
from .helpers import send_from_directory
from .templating import _default_template_ctx_processor
if t.TYPE_CHECKING: # pragma: no cover
from .wrappers import Response
# 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_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, *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:
"""Common behavior shared between :class:`~flask.Flask` and
:class:`~flask.blueprints.Blueprint`.
:param import_name: The import name of the module where this object
is defined. Usually :attr:`__name__` should be used.
:param static_folder: Path to a folder of static files to serve.
If this is set, a static route will be added.
:param static_url_path: URL prefix for the static route.
:param template_folder: Path to a folder containing template files.
for rendering. If this is set, a Jinja loader will be added.
:param root_path: The path that static, template, and resource files
are relative to. Typically not set, it is discovered based on
the ``import_name``.
.. versionadded:: 2.0
"""
name: str
_static_folder: str | None = None
_static_url_path: str | None = None
def __init__(
self,
import_name: str,
static_folder: str | os.PathLike | None = None,
static_url_path: str | None = None,
template_folder: str | os.PathLike | 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 # 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
#: 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()
#: 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, t.Callable] = {}
#: 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]
] = 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:
"""The absolute path to the configured static folder. ``None``
if no static folder is set.
"""
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 | None) -> None:
if value is not None:
value = os.fspath(value).rstrip(r"\/")
self._static_folder = value
@property
def has_static_folder(self) -> bool:
"""``True`` if :attr:`static_folder` is set.
.. versionadded:: 0.5
"""
return self.static_folder is not None
@property
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`.
"""
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
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.
By default, this returns :data:`SEND_FILE_MAX_AGE_DEFAULT` from
the configuration of :data:`~flask.current_app`. This defaults
to ``None``, which tells the browser to use conditional requests
instead of a timed cache, which is usually preferable.
.. versionchanged:: 2.0
The default configuration is ``None`` instead of 12 hours.
.. versionadded:: 0.9
"""
value = current_app.config["SEND_FILE_MAX_AGE_DEFAULT"]
if value is None:
return None
if isinstance(value, timedelta):
return int(value.total_seconds())
return value
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
set.
.. versionadded:: 0.5
"""
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
)
@cached_property
def jinja_loader(self) -> FileSystemLoader | 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
"""
if self.template_folder is not None:
return FileSystemLoader(os.path.join(self.root_path, self.template_folder))
else:
return None
def open_resource(self, resource: str, mode: str = "rb") -> t.IO[t.AnyStr]:
"""Open a resource file relative to :attr:`root_path` for
reading.
For example, if the file ``schema.sql`` is next to the file
``app.py`` where the ``Flask`` app is defined, it can be opened
with:
.. code-block:: python
with app.open_resource("schema.sql") as f:
conn.executescript(f.read())
:param resource: Path to the resource relative to
:attr:`root_path`.
:param mode: Open the file in this mode. Only reading is
supported, valid values are "r" (or "rt") and "rb".
"""
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)
def _method_route(
self,
method: str,
rule: str,
options: dict,
) -> 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]:
"""Shortcut for :meth:`route` with ``methods=["GET"]``.
.. versionadded:: 2.0
"""
return self._method_route("GET", rule, options)
@setupmethod
def post(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Shortcut for :meth:`route` with ``methods=["POST"]``.
.. versionadded:: 2.0
"""
return self._method_route("POST", rule, options)
@setupmethod
def put(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Shortcut for :meth:`route` with ``methods=["PUT"]``.
.. versionadded:: 2.0
"""
return self._method_route("PUT", rule, options)
@setupmethod
def delete(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Shortcut for :meth:`route` with ``methods=["DELETE"]``.
.. versionadded:: 2.0
"""
return self._method_route("DELETE", rule, options)
@setupmethod
def patch(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Shortcut for :meth:`route` with ``methods=["PATCH"]``.
.. versionadded:: 2.0
"""
return self._method_route("PATCH", rule, options)
@setupmethod
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.
.. code-block:: python
@app.route("/")
def index():
return "Hello, World!"
See :ref:`url-route-registrations`.
The endpoint name for the route defaults to the name of the view
function if the ``endpoint`` parameter isn't passed.
The ``methods`` parameter defaults to ``["GET"]``. ``HEAD`` and
``OPTIONS`` are added automatically.
:param rule: The URL rule string.
:param options: Extra options passed to the
:class:`~werkzeug.routing.Rule` object.
"""
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:
"""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:
.. code-block:: python
@app.route("/")
def index():
...
.. code-block:: python
def index():
...
app.add_url_rule("/", view_func=index)
See :ref:`url-route-registrations`.
The endpoint name for the route defaults to the name of the view
function if the ``endpoint`` parameter isn't passed. An error
will be raised if a function has already been registered for the
endpoint.
The ``methods`` parameter defaults to ``["GET"]``. ``HEAD`` is
always added automatically, and ``OPTIONS`` is added
automatically by default.
``view_func`` does not necessarily need to be passed, but if the
rule should participate in routing an endpoint name must be
associated with a view function at some point with the
:meth:`endpoint` decorator.
.. code-block:: python
app.add_url_rule("/", endpoint="index")
@app.endpoint("index")
def index():
...
If ``view_func`` has a ``required_methods`` attribute, those
methods are added to the passed and automatic methods. If it
has a ``provide_automatic_methods`` attribute, it is used as the
default if the parameter is not passed.
:param rule: The URL rule string.
:param endpoint: The endpoint name to associate with the rule
and view function. Used when routing and building URLs.
Defaults to ``view_func.__name__``.
:param view_func: The view function to associate with the
endpoint name.
:param provide_automatic_options: Add the ``OPTIONS`` method and
respond to ``OPTIONS`` requests automatically.
:param options: Extra options passed to the
:class:`~werkzeug.routing.Rule` object.
"""
raise NotImplementedError
@setupmethod
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`.
.. code-block:: python
app.add_url_rule("/ex", endpoint="example")
@app.endpoint("example")
def example():
...
:param endpoint: The endpoint name to associate with the view
function.
"""
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:
"""Register a function to run before each request.
For example, this can be used to open a database connection, or
to load the logged in user from the session.
.. code-block:: python
@app.before_request
def load_user():
if "user_id" in session:
g.user = db.session.get(session["user_id"])
The function will be called without any arguments. If it returns
a non-``None`` value, the value is handled as if it was the
return value from the view, and further request handling is
stopped.
This is available on both app and blueprint objects. When used on an app, this
executes before every request. When used on a blueprint, this executes before
every request that the blueprint handles. To register with a blueprint and
execute before every request, use :meth:`.Blueprint.before_app_request`.
"""
self.before_request_funcs.setdefault(None, []).append(f)
return f
@setupmethod
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
a response object. This allows the functions to modify or
replace the response before it is sent.
If a function raises an exception, any remaining
``after_request`` functions will not be called. Therefore, this
should not be used for actions that must execute, such as to
close resources. Use :meth:`teardown_request` for that.
This is available on both app and blueprint objects. When used on an app, this
executes after every request. When used on a blueprint, this executes after
every request that the blueprint handles. To register with a blueprint and
execute after every request, use :meth:`.Blueprint.after_app_request`.
"""
self.after_request_funcs.setdefault(None, []).append(f)
return f
@setupmethod
def teardown_request(self, f: T_teardown) -> T_teardown:
"""Register a function to be called when the request context is
popped. Typically this happens at the end of each request, but
contexts may be pushed manually as well during testing.
.. code-block:: python
with app.test_request_context():
...
When the ``with`` block exits (or ``ctx.pop()`` is called), the
teardown functions are called just before the request context is
made inactive.
When a teardown function was called because of an unhandled
exception it will be passed an error object. If an
:meth:`errorhandler` is registered, it will handle the exception
and the teardown will not receive it.
Teardown functions must avoid raising exceptions. If they
execute code that might fail they must surround that code with a
``try``/``except`` block and log any errors.
The return values of teardown functions are ignored.
This is available on both app and blueprint objects. When used on an app, this
executes after every request. When used on a blueprint, this executes after
every request that the blueprint handles. To register with a blueprint and
execute after every request, use :meth:`.Blueprint.teardown_app_request`.
"""
self.teardown_request_funcs.setdefault(None, []).append(f)
return f
@setupmethod
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.
This is available on both app and blueprint objects. When used on an app, this
is called for every rendered template. When used on a blueprint, this is called
for templates rendered from the blueprint's views. To register with a blueprint
and affect every template, use :meth:`.Blueprint.app_context_processor`.
"""
self.template_context_processors[None].append(f)
return f
@setupmethod
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.
The function can modify the values captured from the matched url before
they are passed to the view. For example, this can be used to pop a
common language code value and place it in ``g`` rather than pass it to
every view.
The function is passed the endpoint name and values dict. The return
value is ignored.
This is available on both app and blueprint objects. When used on an app, this
is called for every request. When used on a blueprint, this is called for
requests that the blueprint handles. To register with a blueprint and affect
every request, use :meth:`.Blueprint.app_url_value_preprocessor`.
"""
self.url_value_preprocessors[None].append(f)
return f
@setupmethod
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.
This is available on both app and blueprint objects. When used on an app, this
is called for every request. When used on a blueprint, this is called for
requests that the blueprint handles. To register with a blueprint and affect
every request, use :meth:`.Blueprint.app_url_defaults`.
"""
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]:
"""Register a function to handle errors by code or exception class.
A decorator that is used to register a function given an
error code. Example::
@app.errorhandler(404)
def page_not_found(error):
return 'This page does not exist', 404
You can also register handlers for arbitrary exceptions::
@app.errorhandler(DatabaseError)
def special_exception_handler(error):
return 'Database connection failed', 500
This is available on both app and blueprint objects. When used on an app, this
can handle errors from every request. When used on a blueprint, this can handle
errors from requests that the blueprint handles. To register with a blueprint
and affect every request, use :meth:`.Blueprint.app_errorhandler`.
.. versionadded:: 0.7
Use :meth:`register_error_handler` instead of modifying
:attr:`error_handler_spec` directly, for application wide error
handlers.
.. versionadded:: 0.7
One can now additionally also register custom exception types
that do not necessarily have to be a subclass of the
:class:`~werkzeug.exceptions.HTTPException` class.
:param code_or_exception: the code as integer for the handler, or
an arbitrary exception
"""
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:
"""Alternative error attach function to the :meth:`errorhandler`
decorator that is more straightforward to use for non decorator
usage.
.. versionadded:: 0.7
"""
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]:
"""Get the exception class being handled. For HTTP status codes
or ``HTTPException`` subclasses, return both the exception and
status code.
:param exc_class_or_code: Any exception class, or an HTTP status
code as an integer.
"""
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: t.Callable) -> str:
"""Internal helper that returns the default endpoint for a given
function. This always is the function name.
"""
assert view_func is not None, "expected view func if endpoint is not provided."
return view_func.__name__
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_path(import_name):
"""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.origin in {"namespace", None}:
# 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)
elif root_spec.submodule_search_locations:
# package with __init__.py
return os.path.dirname(os.path.dirname(root_spec.origin))
else:
# module
return os.path.dirname(root_spec.origin)
def find_package(import_name: str):
"""Find the prefix that a package is installed under, and the path
that it would be imported from.
The prefix is the directory containing the standard directory
hierarchy (lib, bin, etc.). If the package is not installed to the
system (:attr:`sys.prefix`) or a virtualenv (``site-packages``),
``None`` is returned.
The path is the entry in :attr:`sys.path` that contains the package
for import. If the package is not installed, it's assumed that the
package was imported from the current working directory.
"""
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

View File

@@ -13,11 +13,15 @@ from werkzeug.datastructures import CallbackDict
from .json.tag import TaggedJSONSerializer
if t.TYPE_CHECKING: # pragma: no cover
import typing_extensions as te
from .app import Flask
from .wrappers import Request, Response
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
@@ -45,7 +49,8 @@ class SessionMixin(MutableMapping):
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
@@ -68,7 +73,7 @@ class SecureCookieSession(CallbackDict, SessionMixin):
accessed = False
def __init__(self, initial: t.Any = None) -> None:
def on_update(self) -> None:
def on_update(self: te.Self) -> None:
self.modified = True
self.accessed = True
@@ -177,7 +182,7 @@ class SessionInterface:
def get_cookie_name(self, app: Flask) -> str:
"""The name of the session cookie. Uses``app.config["SESSION_COOKIE_NAME"]``."""
return app.config["SESSION_COOKIE_NAME"]
return app.config["SESSION_COOKIE_NAME"] # type: ignore[no-any-return]
def get_cookie_domain(self, app: Flask) -> str | None:
"""The value of the ``Domain`` parameter on the session cookie. If not set,
@@ -189,8 +194,7 @@ class SessionInterface:
.. versionchanged:: 2.3
Not set by default, does not fall back to ``SERVER_NAME``.
"""
rv = app.config["SESSION_COOKIE_DOMAIN"]
return rv if rv else None
return app.config["SESSION_COOKIE_DOMAIN"] # type: ignore[no-any-return]
def get_cookie_path(self, app: Flask) -> str:
"""Returns the path for which the cookie should be valid. The
@@ -198,27 +202,27 @@ class SessionInterface:
config var if it's set, and falls back to ``APPLICATION_ROOT`` or
uses ``/`` if it's ``None``.
"""
return app.config["SESSION_COOKIE_PATH"] or app.config["APPLICATION_ROOT"]
return app.config["SESSION_COOKIE_PATH"] or app.config["APPLICATION_ROOT"] # type: ignore[no-any-return]
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.
"""
return app.config["SESSION_COOKIE_HTTPONLY"]
return app.config["SESSION_COOKIE_HTTPONLY"] # type: ignore[no-any-return]
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.
"""
return app.config["SESSION_COOKIE_SECURE"]
return app.config["SESSION_COOKIE_SECURE"] # type: ignore[no-any-return]
def get_cookie_samesite(self, app: Flask) -> str:
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.
"""
return app.config["SESSION_COOKIE_SAMESITE"]
return app.config["SESSION_COOKIE_SAMESITE"] # type: ignore[no-any-return]
def get_expiration_time(self, app: Flask, session: SessionMixin) -> datetime | None:
"""A helper method that returns an expiration date for the session

View File

@@ -1,8 +1,5 @@
from __future__ import annotations
import typing as t
import warnings
from blinker import Namespace
# This namespace is only for signals provided by Flask itself.
@@ -18,16 +15,3 @@ 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")
def __getattr__(name: str) -> t.Any:
if name == "signals_available":
warnings.warn(
"The 'signals_available' attribute is deprecated and will be removed in"
" Flask 2.4. Signals are always available.",
DeprecationWarning,
stacklevel=2,
)
return True
raise AttributeError(name)

View File

@@ -17,7 +17,8 @@ from .signals import template_rendered
if t.TYPE_CHECKING: # pragma: no cover
from .app import Flask
from .scaffold import Scaffold
from .sansio.app import App
from .sansio.scaffold import Scaffold
def _default_template_ctx_processor() -> dict[str, t.Any]:
@@ -41,7 +42,7 @@ class Environment(BaseEnvironment):
name of the blueprint to referenced templates if necessary.
"""
def __init__(self, app: Flask, **options: t.Any) -> None:
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)
@@ -53,19 +54,19 @@ class DispatchingJinjaLoader(BaseLoader):
the blueprint folders.
"""
def __init__(self, app: Flask) -> None:
def __init__(self, app: App) -> None:
self.app = app
def get_source( # type: ignore
self, environment: Environment, template: str
) -> tuple[str, str | None, t.Callable | None]:
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: Environment, template: str
) -> tuple[str, str | None, t.Callable | None]:
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
@@ -88,8 +89,8 @@ class DispatchingJinjaLoader(BaseLoader):
raise TemplateNotFound(template)
def _get_source_fast(
self, environment: Environment, template: str
) -> tuple[str, str | None, t.Callable | None]:
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)
@@ -97,9 +98,7 @@ class DispatchingJinjaLoader(BaseLoader):
continue
raise TemplateNotFound(template)
def _iter_loaders(
self, template: str
) -> t.Generator[tuple[Scaffold, BaseLoader], None, None]:
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

View File

@@ -17,6 +17,7 @@ from .cli import ScriptInfo
from .sessions import SessionMixin
if t.TYPE_CHECKING: # pragma: no cover
from _typeshed.wsgi import WSGIEnvironment
from werkzeug.test import TestResponse
from .app import Flask
@@ -134,7 +135,7 @@ class FlaskClient(Client):
@contextmanager
def session_transaction(
self, *args: t.Any, **kwargs: t.Any
) -> t.Generator[SessionMixin, None, None]:
) -> 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
@@ -181,7 +182,7 @@ class FlaskClient(Client):
resp.headers.getlist("Set-Cookie"),
)
def _copy_environ(self, other):
def _copy_environ(self, other: WSGIEnvironment) -> WSGIEnvironment:
out = {**self.environ_base, **other}
if self.preserve_context:
@@ -189,7 +190,9 @@ class FlaskClient(Client):
return out
def _request_from_builder_args(self, args, kwargs):
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)
@@ -210,7 +213,7 @@ class FlaskClient(Client):
):
if isinstance(args[0], werkzeug.test.EnvironBuilder):
builder = copy(args[0])
builder.environ_base = self._copy_environ(builder.environ_base or {})
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(
@@ -287,7 +290,7 @@ class FlaskCliRunner(CliRunner):
:return: a :class:`~click.testing.Result` object.
"""
if cli is None:
cli = self.app.cli # type: ignore
cli = self.app.cli
if "obj" not in kwargs:
kwargs["obj"] = ScriptInfo(create_app=lambda: self.app)

View File

@@ -5,7 +5,7 @@ import typing as t
if t.TYPE_CHECKING: # pragma: no cover
from _typeshed.wsgi import WSGIApplication # noqa: F401
from werkzeug.datastructures import Headers # noqa: F401
from werkzeug.wrappers import Response # 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[
@@ -61,12 +61,17 @@ TeardownCallable = t.Union[
t.Callable[[t.Optional[BaseException]], None],
t.Callable[[t.Optional[BaseException]], t.Awaitable[None]],
]
TemplateContextProcessorCallable = t.Callable[[], t.Dict[str, t.Any]]
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, dict], None]
URLValuePreprocessorCallable = t.Callable[[t.Optional[str], t.Optional[dict]], None]
URLDefaultCallable = t.Callable[[str, t.Dict[str, t.Any]], None]
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
@@ -74,7 +79,10 @@ URLValuePreprocessorCallable = t.Callable[[t.Optional[str], t.Optional[dict]], N
# https://github.com/pallets/flask/issues/4095
# https://github.com/pallets/flask/issues/4295
# https://github.com/pallets/flask/issues/4297
ErrorHandlerCallable = t.Callable[[t.Any], ResponseReturnValue]
ErrorHandlerCallable = t.Union[
t.Callable[[t.Any], ResponseReturnValue],
t.Callable[[t.Any], t.Awaitable[ResponseReturnValue]],
]
RouteCallable = t.Union[
t.Callable[..., ResponseReturnValue],

View File

@@ -6,6 +6,7 @@ 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"]
@@ -60,7 +61,7 @@ class View:
#: decorator.
#:
#: .. versionadded:: 0.8
decorators: t.ClassVar[list[t.Callable]] = []
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
@@ -106,13 +107,13 @@ class View:
self = view.view_class( # type: ignore[attr-defined]
*class_args, **class_kwargs
)
return current_app.ensure_sync(self.dispatch_request)(**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)
return current_app.ensure_sync(self.dispatch_request)(**kwargs) # type: ignore[no-any-return]
if cls.decorators:
view.__name__ = name
@@ -187,4 +188,4 @@ class MethodView(View):
meth = getattr(self, "get", None)
assert meth is not None, f"Unimplemented method {request.method!r}"
return current_app.ensure_sync(meth)(**kwargs)
return current_app.ensure_sync(meth)(**kwargs) # type: ignore[no-any-return]

View File

@@ -3,6 +3,7 @@ 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
@@ -49,13 +50,13 @@ class Request(RequestBase):
#: raised / was raised as part of the request handling. This is
#: usually a :exc:`~werkzeug.exceptions.NotFound` exception or
#: something similar.
routing_exception: Exception | None = None
routing_exception: HTTPException | None = None
@property
def max_content_length(self) -> int | None: # type: ignore
def max_content_length(self) -> int | None: # type: ignore[override]
"""Read-only view of the ``MAX_CONTENT_LENGTH`` config key."""
if current_app:
return current_app.config["MAX_CONTENT_LENGTH"]
return current_app.config["MAX_CONTENT_LENGTH"] # type: ignore[no-any-return]
else:
return None
@@ -167,7 +168,7 @@ class Response(ResponseBase):
Werkzeug's docs.
"""
if current_app:
return current_app.config["MAX_COOKIE_SIZE"]
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