add __future__ annotations

This commit is contained in:
David Lord 2023-04-20 08:30:45 -07:00
parent cfa863c357
commit 44ffe6c6d6
No known key found for this signature in database
GPG key ID: 7A1C87E3F5BC42A8
19 changed files with 277 additions and 262 deletions

View file

@ -46,6 +46,7 @@ Unreleased
:issue:`5051`
- The ``routes`` command shows each rule's ``subdomain`` or ``host`` when domain
matching is in use. :issue:`5004`
- Use postponed evaluation of annotations. :pr:`5071`
Version 2.2.4

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import functools
import inspect
import logging
@ -95,7 +97,7 @@ else:
return inspect.iscoroutinefunction(func)
def _make_timedelta(value: t.Union[timedelta, int, None]) -> t.Optional[timedelta]:
def _make_timedelta(value: timedelta | int | None) -> timedelta | None:
if value is None or isinstance(value, timedelta):
return value
@ -280,7 +282,7 @@ class Flask(Scaffold):
"PERMANENT_SESSION_LIFETIME", get_converter=_make_timedelta
)
json_provider_class: t.Type[JSONProvider] = DefaultJSONProvider
json_provider_class: type[JSONProvider] = DefaultJSONProvider
"""A subclass of :class:`~flask.json.provider.JSONProvider`. An
instance is created and assigned to :attr:`app.json` when creating
the app.
@ -348,7 +350,7 @@ class Flask(Scaffold):
#: client class. Defaults to :class:`~flask.testing.FlaskClient`.
#:
#: .. versionadded:: 0.7
test_client_class: t.Optional[t.Type["FlaskClient"]] = None
test_client_class: type[FlaskClient] | None = None
#: The :class:`~click.testing.CliRunner` subclass, by default
#: :class:`~flask.testing.FlaskCliRunner` that is used by
@ -356,7 +358,7 @@ class Flask(Scaffold):
#: Flask app object as the first argument.
#:
#: .. versionadded:: 1.0
test_cli_runner_class: t.Optional[t.Type["FlaskCliRunner"]] = None
test_cli_runner_class: type[FlaskCliRunner] | None = None
#: the session interface to use. By default an instance of
#: :class:`~flask.sessions.SecureCookieSessionInterface` is used here.
@ -367,15 +369,15 @@ class Flask(Scaffold):
def __init__(
self,
import_name: str,
static_url_path: t.Optional[str] = None,
static_folder: t.Optional[t.Union[str, os.PathLike]] = "static",
static_host: t.Optional[str] = None,
static_url_path: str | None = None,
static_folder: str | os.PathLike | None = "static",
static_host: str | None = None,
host_matching: bool = False,
subdomain_matching: bool = False,
template_folder: t.Optional[t.Union[str, os.PathLike]] = "templates",
instance_path: t.Optional[str] = None,
template_folder: str | os.PathLike | None = "templates",
instance_path: str | None = None,
instance_relative_config: bool = False,
root_path: t.Optional[str] = None,
root_path: str | None = None,
):
super().__init__(
import_name=import_name,
@ -435,8 +437,8 @@ class Flask(Scaffold):
#: Otherwise, its return value is returned by ``url_for``.
#:
#: .. versionadded:: 0.9
self.url_build_error_handlers: t.List[
t.Callable[[Exception, str, t.Dict[str, t.Any]], str]
self.url_build_error_handlers: list[
t.Callable[[Exception, str, dict[str, t.Any]], str]
] = []
#: A list of functions that are called when the application context
@ -445,13 +447,13 @@ class Flask(Scaffold):
#: from databases.
#:
#: .. versionadded:: 0.9
self.teardown_appcontext_funcs: t.List[ft.TeardownCallable] = []
self.teardown_appcontext_funcs: list[ft.TeardownCallable] = []
#: A list of shell context processor functions that should be run
#: when a shell context is created.
#:
#: .. versionadded:: 0.11
self.shell_context_processors: t.List[ft.ShellContextProcessorCallable] = []
self.shell_context_processors: list[ft.ShellContextProcessorCallable] = []
#: Maps registered blueprint names to blueprint objects. The
#: dict retains the order the blueprints were registered in.
@ -459,7 +461,7 @@ class Flask(Scaffold):
#: not track how often they were attached.
#:
#: .. versionadded:: 0.7
self.blueprints: t.Dict[str, "Blueprint"] = {}
self.blueprints: dict[str, Blueprint] = {}
#: a place where extensions can store application specific state. For
#: example this is where an extension could store database engines and
@ -734,7 +736,7 @@ class Flask(Scaffold):
:param context: the context as a dictionary that is updated in place
to add extra variables.
"""
names: t.Iterable[t.Optional[str]] = (None,)
names: t.Iterable[str | None] = (None,)
# A template may be rendered outside a request context.
if request:
@ -785,9 +787,9 @@ class Flask(Scaffold):
def run(
self,
host: t.Optional[str] = None,
port: t.Optional[int] = None,
debug: t.Optional[bool] = None,
host: str | None = None,
port: int | None = None,
debug: bool | None = None,
load_dotenv: bool = True,
**options: t.Any,
) -> None:
@ -906,7 +908,7 @@ class Flask(Scaffold):
# without reloader and that stuff from an interactive shell.
self._got_first_request = False
def test_client(self, use_cookies: bool = True, **kwargs: t.Any) -> "FlaskClient":
def test_client(self, use_cookies: bool = True, **kwargs: t.Any) -> FlaskClient:
"""Creates a test client for this application. For information
about unit testing head over to :doc:`/testing`.
@ -964,7 +966,7 @@ class Flask(Scaffold):
self, self.response_class, use_cookies=use_cookies, **kwargs
)
def test_cli_runner(self, **kwargs: t.Any) -> "FlaskCliRunner":
def test_cli_runner(self, **kwargs: t.Any) -> FlaskCliRunner:
"""Create a CLI runner for testing CLI commands.
See :ref:`testing-cli`.
@ -982,7 +984,7 @@ class Flask(Scaffold):
return cls(self, **kwargs) # type: ignore
@setupmethod
def register_blueprint(self, blueprint: "Blueprint", **options: t.Any) -> None:
def register_blueprint(self, blueprint: Blueprint, **options: t.Any) -> None:
"""Register a :class:`~flask.Blueprint` on the application. Keyword
arguments passed to this method will override the defaults set on the
blueprint.
@ -1009,7 +1011,7 @@ class Flask(Scaffold):
"""
blueprint.register(self, options)
def iter_blueprints(self) -> t.ValuesView["Blueprint"]:
def iter_blueprints(self) -> t.ValuesView[Blueprint]:
"""Iterates over all blueprints by the order they were registered.
.. versionadded:: 0.11
@ -1020,9 +1022,9 @@ class Flask(Scaffold):
def add_url_rule(
self,
rule: str,
endpoint: t.Optional[str] = None,
view_func: t.Optional[ft.RouteCallable] = None,
provide_automatic_options: t.Optional[bool] = None,
endpoint: str | None = None,
view_func: ft.RouteCallable | None = None,
provide_automatic_options: bool | None = None,
**options: t.Any,
) -> None:
if endpoint is None:
@ -1077,7 +1079,7 @@ class Flask(Scaffold):
@setupmethod
def template_filter(
self, name: t.Optional[str] = None
self, name: str | None = None
) -> t.Callable[[T_template_filter], T_template_filter]:
"""A decorator that is used to register custom template filter.
You can specify a name for the filter, otherwise the function
@ -1099,7 +1101,7 @@ class Flask(Scaffold):
@setupmethod
def add_template_filter(
self, f: ft.TemplateFilterCallable, name: t.Optional[str] = None
self, f: ft.TemplateFilterCallable, name: str | None = None
) -> None:
"""Register a custom template filter. Works exactly like the
:meth:`template_filter` decorator.
@ -1111,7 +1113,7 @@ class Flask(Scaffold):
@setupmethod
def template_test(
self, name: t.Optional[str] = None
self, name: str | None = None
) -> t.Callable[[T_template_test], T_template_test]:
"""A decorator that is used to register custom template test.
You can specify a name for the test, otherwise the function
@ -1140,7 +1142,7 @@ class Flask(Scaffold):
@setupmethod
def add_template_test(
self, f: ft.TemplateTestCallable, name: t.Optional[str] = None
self, f: ft.TemplateTestCallable, name: str | None = None
) -> None:
"""Register a custom template test. Works exactly like the
:meth:`template_test` decorator.
@ -1154,7 +1156,7 @@ class Flask(Scaffold):
@setupmethod
def template_global(
self, name: t.Optional[str] = None
self, name: str | None = None
) -> t.Callable[[T_template_global], T_template_global]:
"""A decorator that is used to register a custom template global function.
You can specify a name for the global function, otherwise the function
@ -1178,7 +1180,7 @@ class Flask(Scaffold):
@setupmethod
def add_template_global(
self, f: ft.TemplateGlobalCallable, name: t.Optional[str] = None
self, f: ft.TemplateGlobalCallable, name: str | None = None
) -> None:
"""Register a custom template global function. Works exactly like the
:meth:`template_global` decorator.
@ -1235,7 +1237,7 @@ class Flask(Scaffold):
self.shell_context_processors.append(f)
return f
def _find_error_handler(self, e: Exception) -> t.Optional[ft.ErrorHandlerCallable]:
def _find_error_handler(self, e: Exception) -> ft.ErrorHandlerCallable | None:
"""Return a registered error handler for an exception in this order:
blueprint handler for a specific code, app handler for a specific code,
blueprint handler for an exception class, app handler for an exception
@ -1260,7 +1262,7 @@ class Flask(Scaffold):
def handle_http_exception(
self, e: HTTPException
) -> t.Union[HTTPException, ft.ResponseReturnValue]:
) -> HTTPException | ft.ResponseReturnValue:
"""Handles an HTTP exception. By default this will invoke the
registered error handlers and fall back to returning the
exception as response.
@ -1330,7 +1332,7 @@ class Flask(Scaffold):
def handle_user_exception(
self, e: Exception
) -> t.Union[HTTPException, ft.ResponseReturnValue]:
) -> HTTPException | ft.ResponseReturnValue:
"""This method is called whenever an exception occurs that
should be handled. A special case is :class:`~werkzeug
.exceptions.HTTPException` which is forwarded to the
@ -1404,7 +1406,7 @@ class Flask(Scaffold):
raise e
self.log_exception(exc_info)
server_error: t.Union[InternalServerError, ft.ResponseReturnValue]
server_error: InternalServerError | ft.ResponseReturnValue
server_error = InternalServerError(original_exception=e)
handler = self._find_error_handler(server_error)
@ -1415,9 +1417,7 @@ class Flask(Scaffold):
def log_exception(
self,
exc_info: t.Union[
t.Tuple[type, BaseException, TracebackType], t.Tuple[None, None, None]
],
exc_info: (tuple[type, BaseException, TracebackType] | tuple[None, None, None]),
) -> None:
"""Logs an exception. This is called by :meth:`handle_exception`
if debugging is disabled and right before the handler is called.
@ -1430,7 +1430,7 @@ class Flask(Scaffold):
f"Exception on {request.path} [{request.method}]", exc_info=exc_info
)
def raise_routing_exception(self, request: Request) -> "te.NoReturn":
def raise_routing_exception(self, request: Request) -> te.NoReturn:
"""Intercept routing exceptions and possibly do something else.
In debug mode, intercept a routing redirect and replace it with
@ -1480,7 +1480,7 @@ class Flask(Scaffold):
):
return self.make_default_options_response()
# otherwise dispatch to the handler for that endpoint
view_args: t.Dict[str, t.Any] = req.view_args # type: ignore[assignment]
view_args: dict[str, t.Any] = req.view_args # type: ignore[assignment]
return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
def full_dispatch_request(self) -> Response:
@ -1503,7 +1503,7 @@ class Flask(Scaffold):
def finalize_request(
self,
rv: t.Union[ft.ResponseReturnValue, HTTPException],
rv: ft.ResponseReturnValue | HTTPException,
from_error_handler: bool = False,
) -> Response:
"""Given the return value from a view function this finalizes
@ -1545,7 +1545,7 @@ class Flask(Scaffold):
rv.allow.update(methods)
return rv
def should_ignore_error(self, error: t.Optional[BaseException]) -> bool:
def should_ignore_error(self, error: BaseException | None) -> bool:
"""This is called to figure out if an error should be ignored
or not as far as the teardown system is concerned. If this
function returns ``True`` then the teardown handlers will not be
@ -1596,10 +1596,10 @@ class Flask(Scaffold):
self,
endpoint: str,
*,
_anchor: t.Optional[str] = None,
_method: t.Optional[str] = None,
_scheme: t.Optional[str] = None,
_external: t.Optional[bool] = None,
_anchor: str | None = None,
_method: str | None = None,
_scheme: str | None = None,
_external: bool | None = None,
**values: t.Any,
) -> str:
"""Generate a URL to the given endpoint with the given values.
@ -1871,9 +1871,7 @@ class Flask(Scaffold):
return rv
def create_url_adapter(
self, request: t.Optional[Request]
) -> t.Optional[MapAdapter]:
def create_url_adapter(self, request: Request | None) -> MapAdapter | None:
"""Creates a URL adapter for the given request. The URL adapter
is created at a point where the request context is not yet set
up so the request is passed explicitly.
@ -1920,7 +1918,7 @@ class Flask(Scaffold):
.. versionadded:: 0.7
"""
names: t.Iterable[t.Optional[str]] = (None,)
names: t.Iterable[str | None] = (None,)
# url_for may be called outside a request context, parse the
# passed endpoint instead of using request.blueprints.
@ -1935,7 +1933,7 @@ class Flask(Scaffold):
func(endpoint, values)
def handle_url_build_error(
self, error: BuildError, endpoint: str, values: t.Dict[str, t.Any]
self, error: BuildError, endpoint: str, values: dict[str, t.Any]
) -> str:
"""Called by :meth:`.url_for` if a
:exc:`~werkzeug.routing.BuildError` was raised. If this returns
@ -1968,7 +1966,7 @@ class Flask(Scaffold):
raise error
def preprocess_request(self) -> t.Optional[ft.ResponseReturnValue]:
def preprocess_request(self) -> ft.ResponseReturnValue | None:
"""Called before the request is dispatched. Calls
:attr:`url_value_preprocessors` registered with the app and the
current blueprint (if any). Then calls :attr:`before_request_funcs`
@ -2024,7 +2022,7 @@ class Flask(Scaffold):
return response
def do_teardown_request(
self, exc: t.Optional[BaseException] = _sentinel # type: ignore
self, exc: BaseException | None = _sentinel # type: ignore
) -> None:
"""Called after the request is dispatched and the response is
returned, right before the request context is popped.
@ -2057,7 +2055,7 @@ class Flask(Scaffold):
request_tearing_down.send(self, _async_wrapper=self.ensure_sync, exc=exc)
def do_teardown_appcontext(
self, exc: t.Optional[BaseException] = _sentinel # type: ignore
self, exc: BaseException | None = _sentinel # type: ignore
) -> None:
"""Called right before the application context is popped.
@ -2200,7 +2198,7 @@ class Flask(Scaffold):
start the response.
"""
ctx = self.request_context(environ)
error: t.Optional[BaseException] = None
error: BaseException | None = None
try:
try:
ctx.push()

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import os
import typing as t
from collections import defaultdict
@ -38,8 +40,8 @@ class BlueprintSetupState:
def __init__(
self,
blueprint: "Blueprint",
app: "Flask",
blueprint: Blueprint,
app: Flask,
options: t.Any,
first_registration: bool,
) -> None:
@ -85,8 +87,8 @@ class BlueprintSetupState:
def add_url_rule(
self,
rule: str,
endpoint: t.Optional[str] = None,
view_func: t.Optional[t.Callable] = None,
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)
@ -173,14 +175,14 @@ class Blueprint(Scaffold):
self,
name: str,
import_name: str,
static_folder: t.Optional[t.Union[str, os.PathLike]] = None,
static_url_path: t.Optional[str] = None,
template_folder: t.Optional[t.Union[str, os.PathLike]] = None,
url_prefix: t.Optional[str] = None,
subdomain: t.Optional[str] = None,
url_defaults: t.Optional[dict] = None,
root_path: t.Optional[str] = None,
cli_group: t.Optional[str] = _sentinel, # type: ignore
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,
@ -199,14 +201,14 @@ class Blueprint(Scaffold):
self.name = name
self.url_prefix = url_prefix
self.subdomain = subdomain
self.deferred_functions: t.List[DeferredSetupFunction] = []
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: t.List[t.Tuple["Blueprint", dict]] = []
self._blueprints: list[tuple[Blueprint, dict]] = []
def _check_setup_finished(self, f_name: str) -> None:
if self._got_registered_once:
@ -242,7 +244,7 @@ class Blueprint(Scaffold):
self.record(update_wrapper(wrapper, func))
def make_setup_state(
self, app: "Flask", options: dict, first_registration: bool = False
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.
@ -251,7 +253,7 @@ class Blueprint(Scaffold):
return BlueprintSetupState(self, app, options, first_registration)
@setupmethod
def register_blueprint(self, blueprint: "Blueprint", **options: t.Any) -> None:
def register_blueprint(self, blueprint: Blueprint, **options: t.Any) -> None:
"""Register a :class:`~flask.Blueprint` on this blueprint. Keyword
arguments passed to this method will override the defaults set
on the blueprint.
@ -268,7 +270,7 @@ class Blueprint(Scaffold):
raise ValueError("Cannot register a blueprint on itself")
self._blueprints.append((blueprint, options))
def register(self, app: "Flask", options: dict) -> None:
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
@ -408,9 +410,9 @@ class Blueprint(Scaffold):
def add_url_rule(
self,
rule: str,
endpoint: t.Optional[str] = None,
view_func: t.Optional[ft.RouteCallable] = None,
provide_automatic_options: t.Optional[bool] = None,
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
@ -437,7 +439,7 @@ class Blueprint(Scaffold):
@setupmethod
def app_template_filter(
self, name: t.Optional[str] = None
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`.
@ -454,7 +456,7 @@ class Blueprint(Scaffold):
@setupmethod
def add_app_template_filter(
self, f: ft.TemplateFilterCallable, name: t.Optional[str] = None
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
@ -471,7 +473,7 @@ class Blueprint(Scaffold):
@setupmethod
def app_template_test(
self, name: t.Optional[str] = None
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`.
@ -490,7 +492,7 @@ class Blueprint(Scaffold):
@setupmethod
def add_app_template_test(
self, f: ft.TemplateTestCallable, name: t.Optional[str] = None
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
@ -509,7 +511,7 @@ class Blueprint(Scaffold):
@setupmethod
def app_template_global(
self, name: t.Optional[str] = None
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`.
@ -528,7 +530,7 @@ class Blueprint(Scaffold):
@setupmethod
def add_app_template_global(
self, f: ft.TemplateGlobalCallable, name: t.Optional[str] = None
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
@ -589,7 +591,7 @@ class Blueprint(Scaffold):
@setupmethod
def app_errorhandler(
self, code: t.Union[t.Type[Exception], int]
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`.

View file

@ -285,7 +285,7 @@ class ScriptInfo:
self.create_app = create_app
#: A dictionary with arbitrary data that can be associated with
#: this script info.
self.data: t.Dict[t.Any, t.Any] = {}
self.data: dict[t.Any, t.Any] = {}
self.set_debug_flag = set_debug_flag
self._loaded_app: Flask | None = None

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import errno
import json
import os
@ -10,7 +12,7 @@ from werkzeug.utils import import_string
class ConfigAttribute:
"""Makes an attribute forward to the config"""
def __init__(self, name: str, get_converter: t.Optional[t.Callable] = None) -> None:
def __init__(self, name: str, get_converter: t.Callable | None = None) -> None:
self.__name__ = name
self.get_converter = get_converter
@ -70,7 +72,7 @@ class Config(dict):
:param defaults: an optional dictionary of default values
"""
def __init__(self, root_path: str, defaults: t.Optional[dict] = None) -> None:
def __init__(self, root_path: str, defaults: dict | None = None) -> None:
super().__init__(defaults or {})
self.root_path = root_path
@ -191,7 +193,7 @@ class Config(dict):
self.from_object(d)
return True
def from_object(self, obj: t.Union[object, str]) -> None:
def from_object(self, obj: object | str) -> None:
"""Updates the values from the given object. An object can be of one
of the following two types:
@ -278,7 +280,7 @@ class Config(dict):
return self.from_mapping(obj)
def from_mapping(
self, mapping: t.Optional[t.Mapping[str, t.Any]] = None, **kwargs: t.Any
self, mapping: t.Mapping[str, t.Any] | None = None, **kwargs: t.Any
) -> bool:
"""Updates the config like :meth:`update` ignoring items with
non-upper keys.
@ -287,7 +289,7 @@ class Config(dict):
.. versionadded:: 0.11
"""
mappings: t.Dict[str, t.Any] = {}
mappings: dict[str, t.Any] = {}
if mapping is not None:
mappings.update(mapping)
mappings.update(kwargs)
@ -298,7 +300,7 @@ class Config(dict):
def get_namespace(
self, namespace: str, lowercase: bool = True, trim_namespace: bool = True
) -> t.Dict[str, t.Any]:
) -> dict[str, t.Any]:
"""Returns a dictionary containing a subset of configuration options
that match the specified namespace/prefix. Example usage::

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import contextvars
import sys
import typing as t
@ -60,7 +62,7 @@ class _AppCtxGlobals:
except KeyError:
raise AttributeError(name) from None
def get(self, name: str, default: t.Optional[t.Any] = None) -> t.Any:
def get(self, name: str, default: t.Any | None = None) -> t.Any:
"""Get an attribute by name, or a default value. Like
:meth:`dict.get`.
@ -233,18 +235,18 @@ class AppContext:
running CLI commands.
"""
def __init__(self, app: "Flask") -> None:
def __init__(self, app: Flask) -> None:
self.app = app
self.url_adapter = app.create_url_adapter(None)
self.g: _AppCtxGlobals = app.app_ctx_globals_class()
self._cv_tokens: t.List[contextvars.Token] = []
self._cv_tokens: list[contextvars.Token] = []
def push(self) -> None:
"""Binds the app context to the current context."""
self._cv_tokens.append(_cv_app.set(self))
appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync)
def pop(self, exc: t.Optional[BaseException] = _sentinel) -> None: # type: ignore
def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore
"""Pops the app context."""
try:
if len(self._cv_tokens) == 1:
@ -262,15 +264,15 @@ class AppContext:
appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync)
def __enter__(self) -> "AppContext":
def __enter__(self) -> AppContext:
self.push()
return self
def __exit__(
self,
exc_type: t.Optional[type],
exc_value: t.Optional[BaseException],
tb: t.Optional[TracebackType],
exc_type: type | None,
exc_value: BaseException | None,
tb: TracebackType | None,
) -> None:
self.pop(exc_value)
@ -299,10 +301,10 @@ class RequestContext:
def __init__(
self,
app: "Flask",
app: Flask,
environ: dict,
request: t.Optional["Request"] = None,
session: t.Optional["SessionMixin"] = None,
request: Request | None = None,
session: SessionMixin | None = None,
) -> None:
self.app = app
if request is None:
@ -314,16 +316,16 @@ class RequestContext:
self.url_adapter = app.create_url_adapter(self.request)
except HTTPException as e:
self.request.routing_exception = e
self.flashes: t.Optional[t.List[t.Tuple[str, str]]] = None
self.session: t.Optional["SessionMixin"] = session
self.flashes: list[tuple[str, str]] | None = None
self.session: SessionMixin | None = session
# Functions that should be executed after the request on the response
# object. These will be called before the regular "after_request"
# functions.
self._after_request_functions: t.List[ft.AfterRequestCallable] = []
self._after_request_functions: list[ft.AfterRequestCallable] = []
self._cv_tokens: t.List[t.Tuple[contextvars.Token, t.Optional[AppContext]]] = []
self._cv_tokens: list[tuple[contextvars.Token, AppContext | None]] = []
def copy(self) -> "RequestContext":
def copy(self) -> RequestContext:
"""Creates a copy of this request context with the same request object.
This can be used to move a request context to a different greenlet.
Because the actual request object is the same this cannot be used to
@ -382,7 +384,7 @@ class RequestContext:
if self.url_adapter is not None:
self.match_request()
def pop(self, exc: t.Optional[BaseException] = _sentinel) -> None: # type: ignore
def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore
"""Pops the request context and unbinds it by doing that. This will
also trigger the execution of functions registered by the
:meth:`~flask.Flask.teardown_request` decorator.
@ -419,15 +421,15 @@ class RequestContext:
f"Popped wrong request context. ({ctx!r} instead of {self!r})"
)
def __enter__(self) -> "RequestContext":
def __enter__(self) -> RequestContext:
self.push()
return self
def __exit__(
self,
exc_type: t.Optional[type],
exc_value: t.Optional[BaseException],
tb: t.Optional[TracebackType],
exc_type: type | None,
exc_value: BaseException | None,
tb: TracebackType | None,
) -> None:
self.pop(exc_value)

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import typing as t
from .app import Flask

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import typing as t
from contextvars import ContextVar
@ -18,7 +20,7 @@ class _FakeStack:
self.cv = cv
@property
def top(self) -> t.Optional[t.Any]:
def top(self) -> t.Any | None:
import warnings
warnings.warn(
@ -38,15 +40,15 @@ This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.\
"""
_cv_app: ContextVar["AppContext"] = ContextVar("flask.app_ctx")
_cv_app: ContextVar[AppContext] = ContextVar("flask.app_ctx")
__app_ctx_stack = _FakeStack("app", _cv_app)
app_ctx: "AppContext" = LocalProxy( # type: ignore[assignment]
app_ctx: AppContext = LocalProxy( # type: ignore[assignment]
_cv_app, unbound_message=_no_app_msg
)
current_app: "Flask" = LocalProxy( # type: ignore[assignment]
current_app: Flask = LocalProxy( # type: ignore[assignment]
_cv_app, "app", unbound_message=_no_app_msg
)
g: "_AppCtxGlobals" = LocalProxy( # type: ignore[assignment]
g: _AppCtxGlobals = LocalProxy( # type: ignore[assignment]
_cv_app, "g", unbound_message=_no_app_msg
)
@ -57,15 +59,15 @@ This typically means that you attempted to use functionality that needed
an active HTTP request. Consult the documentation on testing for
information about how to avoid this problem.\
"""
_cv_request: ContextVar["RequestContext"] = ContextVar("flask.request_ctx")
_cv_request: ContextVar[RequestContext] = ContextVar("flask.request_ctx")
__request_ctx_stack = _FakeStack("request", _cv_request)
request_ctx: "RequestContext" = LocalProxy( # type: ignore[assignment]
request_ctx: RequestContext = LocalProxy( # type: ignore[assignment]
_cv_request, unbound_message=_no_req_msg
)
request: "Request" = LocalProxy( # type: ignore[assignment]
request: Request = LocalProxy( # type: ignore[assignment]
_cv_request, "request", unbound_message=_no_req_msg
)
session: "SessionMixin" = LocalProxy( # type: ignore[assignment]
session: SessionMixin = LocalProxy( # type: ignore[assignment]
_cv_request, "session", unbound_message=_no_req_msg
)

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import os
import pkgutil
import socket
@ -50,9 +52,9 @@ def get_load_dotenv(default: bool = True) -> bool:
def stream_with_context(
generator_or_function: t.Union[
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
@ -128,7 +130,7 @@ def stream_with_context(
return wrapped_g
def make_response(*args: t.Any) -> "Response":
def make_response(*args: t.Any) -> Response:
"""Sometimes it is necessary to set additional headers in a view. Because
views do not have to return response objects but can return a value that
is converted into a response object by Flask itself, it becomes tricky to
@ -180,10 +182,10 @@ def make_response(*args: t.Any) -> "Response":
def url_for(
endpoint: str,
*,
_anchor: t.Optional[str] = None,
_method: t.Optional[str] = None,
_scheme: t.Optional[str] = None,
_external: t.Optional[bool] = None,
_anchor: str | None = None,
_method: str | None = None,
_scheme: str | None = None,
_external: bool | None = None,
**values: t.Any,
) -> str:
"""Generate a URL to the given endpoint with the given values.
@ -232,8 +234,8 @@ def url_for(
def redirect(
location: str, code: int = 302, Response: t.Optional[t.Type["BaseResponse"]] = None
) -> "BaseResponse":
location: str, code: int = 302, Response: type[BaseResponse] | None = None
) -> BaseResponse:
"""Create a redirect response object.
If :data:`~flask.current_app` is available, it will use its
@ -255,9 +257,7 @@ def redirect(
return _wz_redirect(location, code=code, Response=Response)
def abort(
code: t.Union[int, "BaseResponse"], *args: t.Any, **kwargs: t.Any
) -> "te.NoReturn":
def abort(code: int | BaseResponse, *args: t.Any, **kwargs: t.Any) -> te.NoReturn:
"""Raise an :exc:`~werkzeug.exceptions.HTTPException` for the given
status code.
@ -338,7 +338,7 @@ def flash(message: str, category: str = "message") -> None:
def get_flashed_messages(
with_categories: bool = False, category_filter: t.Iterable[str] = ()
) -> t.Union[t.List[str], t.List[t.Tuple[str, str]]]:
) -> list[str] | list[tuple[str, str]]:
"""Pulls all flashed messages from the session and returns them.
Further calls in the same request to the function will return
the same messages. By default just the messages are returned,
@ -378,7 +378,7 @@ def get_flashed_messages(
return flashes
def _prepare_send_file_kwargs(**kwargs: t.Any) -> t.Dict[str, t.Any]:
def _prepare_send_file_kwargs(**kwargs: t.Any) -> dict[str, t.Any]:
if kwargs.get("max_age") is None:
kwargs["max_age"] = current_app.get_send_file_max_age
@ -392,17 +392,15 @@ def _prepare_send_file_kwargs(**kwargs: t.Any) -> t.Dict[str, t.Any]:
def send_file(
path_or_file: t.Union[os.PathLike, str, t.BinaryIO],
mimetype: t.Optional[str] = None,
path_or_file: os.PathLike | str | t.BinaryIO,
mimetype: str | None = None,
as_attachment: bool = False,
download_name: t.Optional[str] = None,
download_name: str | None = None,
conditional: bool = True,
etag: t.Union[bool, str] = True,
last_modified: t.Optional[t.Union[datetime, int, float]] = None,
max_age: t.Optional[
t.Union[int, t.Callable[[t.Optional[str]], t.Optional[int]]]
] = None,
) -> "Response":
etag: bool | str = True,
last_modified: datetime | int | float | None = None,
max_age: None | (int | t.Callable[[str | None], int | None]) = None,
) -> Response:
"""Send the contents of a file to the client.
The first argument can be a file path or a file-like object. Paths
@ -520,10 +518,10 @@ def send_file(
def send_from_directory(
directory: t.Union[os.PathLike, str],
path: t.Union[os.PathLike, str],
directory: os.PathLike | str,
path: os.PathLike | str,
**kwargs: t.Any,
) -> "Response":
) -> Response:
"""Send a file from within a directory using :func:`send_file`.
.. code-block:: python
@ -627,8 +625,8 @@ class locked_cached_property(werkzeug.utils.cached_property):
def __init__(
self,
fget: t.Callable[[t.Any], t.Any],
name: t.Optional[str] = None,
doc: t.Optional[str] = None,
name: str | None = None,
doc: str | None = None,
) -> None:
import warnings
@ -687,8 +685,8 @@ def is_ip(value: str) -> bool:
@lru_cache(maxsize=None)
def _split_blueprint_path(name: str) -> t.List[str]:
out: t.List[str] = [name]
def _split_blueprint_path(name: str) -> list[str]:
out: list[str] = [name]
if "." in name:
out.extend(_split_blueprint_path(name.rpartition(".")[0]))

View file

@ -72,7 +72,7 @@ class JSONProvider:
return self.loads(fp.read(), **kwargs)
def _prepare_response_obj(
self, args: t.Tuple[t.Any, ...], kwargs: t.Dict[str, t.Any]
self, args: tuple[t.Any, ...], kwargs: dict[str, t.Any]
) -> t.Any:
if args and kwargs:
raise TypeError("app.json.response() takes either args or kwargs, not both")
@ -204,7 +204,7 @@ class DefaultJSONProvider(JSONProvider):
:param kwargs: Treat as a dict to serialize.
"""
obj = self._prepare_response_obj(args, kwargs)
dump_args: t.Dict[str, t.Any] = {}
dump_args: dict[str, t.Any] = {}
if (self.compact is None and self._app.debug) or self.compact is False:
dump_args.setdefault("indent", 2)

View file

@ -40,6 +40,8 @@ be processed before ``dict``.
app.session_interface.serializer.register(TagOrderedDict, index=0)
"""
from __future__ import annotations
import typing as t
from base64 import b64decode
from base64 import b64encode
@ -61,9 +63,9 @@ class JSONTag:
#: The tag to mark the serialized object with. If ``None``, this tag is
#: only used as an intermediate step during tagging.
key: t.Optional[str] = None
key: str | None = None
def __init__(self, serializer: "TaggedJSONSerializer") -> None:
def __init__(self, serializer: TaggedJSONSerializer) -> None:
"""Create a tagger for the given serializer."""
self.serializer = serializer
@ -244,17 +246,17 @@ class TaggedJSONSerializer:
]
def __init__(self) -> None:
self.tags: t.Dict[str, JSONTag] = {}
self.order: t.List[JSONTag] = []
self.tags: dict[str, JSONTag] = {}
self.order: list[JSONTag] = []
for cls in self.default_tags:
self.register(cls)
def register(
self,
tag_class: t.Type[JSONTag],
tag_class: type[JSONTag],
force: bool = False,
index: t.Optional[int] = None,
index: int | None = None,
) -> None:
"""Register a new tag with this serializer.
@ -283,7 +285,7 @@ class TaggedJSONSerializer:
else:
self.order.insert(index, tag)
def tag(self, value: t.Any) -> t.Dict[str, t.Any]:
def tag(self, value: t.Any) -> dict[str, t.Any]:
"""Convert a value to a tagged representation if necessary."""
for tag in self.order:
if tag.check(value):
@ -291,7 +293,7 @@ class TaggedJSONSerializer:
return value
def untag(self, value: t.Dict[str, t.Any]) -> t.Any:
def untag(self, value: dict[str, t.Any]) -> t.Any:
"""Convert a tagged representation back to the original type."""
if len(value) != 1:
return value

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import logging
import sys
import typing as t
@ -50,7 +52,7 @@ default_handler.setFormatter(
)
def create_logger(app: "Flask") -> logging.Logger:
def create_logger(app: Flask) -> logging.Logger:
"""Get the Flask app's logger and configure it if needed.
The logger name will be the same as

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import importlib.util
import os
import pathlib
@ -70,16 +72,16 @@ class Scaffold:
"""
name: str
_static_folder: t.Optional[str] = None
_static_url_path: t.Optional[str] = None
_static_folder: str | None = None
_static_url_path: str | None = None
def __init__(
self,
import_name: str,
static_folder: t.Optional[t.Union[str, os.PathLike]] = None,
static_url_path: t.Optional[str] = None,
template_folder: t.Optional[t.Union[str, os.PathLike]] = None,
root_path: t.Optional[str] = None,
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.
@ -112,7 +114,7 @@ class Scaffold:
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.view_functions: t.Dict[str, t.Callable] = {}
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
@ -127,9 +129,9 @@ class Scaffold:
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.error_handler_spec: t.Dict[
self.error_handler_spec: dict[
ft.AppOrBlueprintKey,
t.Dict[t.Optional[int], t.Dict[t.Type[Exception], ft.ErrorHandlerCallable]],
dict[int | None, dict[type[Exception], ft.ErrorHandlerCallable]],
] = defaultdict(lambda: defaultdict(dict))
#: A data structure of functions to call at the beginning of
@ -142,8 +144,8 @@ class Scaffold:
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.before_request_funcs: t.Dict[
ft.AppOrBlueprintKey, t.List[ft.BeforeRequestCallable]
self.before_request_funcs: dict[
ft.AppOrBlueprintKey, list[ft.BeforeRequestCallable]
] = defaultdict(list)
#: A data structure of functions to call at the end of each
@ -156,8 +158,8 @@ class Scaffold:
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.after_request_funcs: t.Dict[
ft.AppOrBlueprintKey, t.List[ft.AfterRequestCallable]
self.after_request_funcs: dict[
ft.AppOrBlueprintKey, list[ft.AfterRequestCallable]
] = defaultdict(list)
#: A data structure of functions to call at the end of each
@ -171,8 +173,8 @@ class Scaffold:
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.teardown_request_funcs: t.Dict[
ft.AppOrBlueprintKey, t.List[ft.TeardownCallable]
self.teardown_request_funcs: dict[
ft.AppOrBlueprintKey, list[ft.TeardownCallable]
] = defaultdict(list)
#: A data structure of functions to call to pass extra context
@ -186,8 +188,8 @@ class Scaffold:
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.template_context_processors: t.Dict[
ft.AppOrBlueprintKey, t.List[ft.TemplateContextProcessorCallable]
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
@ -201,9 +203,9 @@ class Scaffold:
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.url_value_preprocessors: t.Dict[
self.url_value_preprocessors: dict[
ft.AppOrBlueprintKey,
t.List[ft.URLValuePreprocessorCallable],
list[ft.URLValuePreprocessorCallable],
] = defaultdict(list)
#: A data structure of functions to call to modify the keyword
@ -217,8 +219,8 @@ class Scaffold:
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.url_default_functions: t.Dict[
ft.AppOrBlueprintKey, t.List[ft.URLDefaultCallable]
self.url_default_functions: dict[
ft.AppOrBlueprintKey, list[ft.URLDefaultCallable]
] = defaultdict(list)
def __repr__(self) -> str:
@ -228,7 +230,7 @@ class Scaffold:
raise NotImplementedError
@property
def static_folder(self) -> t.Optional[str]:
def static_folder(self) -> str | None:
"""The absolute path to the configured static folder. ``None``
if no static folder is set.
"""
@ -238,7 +240,7 @@ class Scaffold:
return None
@static_folder.setter
def static_folder(self, value: t.Optional[t.Union[str, os.PathLike]]) -> None:
def static_folder(self, value: str | os.PathLike | None) -> None:
if value is not None:
value = os.fspath(value).rstrip(r"\/")
@ -253,7 +255,7 @@ class Scaffold:
return self.static_folder is not None
@property
def static_url_path(self) -> t.Optional[str]:
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
@ -269,13 +271,13 @@ class Scaffold:
return None
@static_url_path.setter
def static_url_path(self, value: t.Optional[str]) -> None:
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: t.Optional[str]) -> t.Optional[int]:
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.
@ -299,7 +301,7 @@ class Scaffold:
return value
def send_static_file(self, filename: str) -> "Response":
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
@ -318,7 +320,7 @@ class Scaffold:
)
@cached_property
def jinja_loader(self) -> t.Optional[FileSystemLoader]:
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.
@ -440,9 +442,9 @@ class Scaffold:
def add_url_rule(
self,
rule: str,
endpoint: t.Optional[str] = None,
view_func: t.Optional[ft.RouteCallable] = None,
provide_automatic_options: t.Optional[bool] = None,
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
@ -668,7 +670,7 @@ class Scaffold:
@setupmethod
def errorhandler(
self, code_or_exception: t.Union[t.Type[Exception], int]
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.
@ -713,7 +715,7 @@ class Scaffold:
@setupmethod
def register_error_handler(
self,
code_or_exception: t.Union[t.Type[Exception], int],
code_or_exception: type[Exception] | int,
f: ft.ErrorHandlerCallable,
) -> None:
"""Alternative error attach function to the :meth:`errorhandler`
@ -727,8 +729,8 @@ class Scaffold:
@staticmethod
def _get_exc_class_and_code(
exc_class_or_code: t.Union[t.Type[Exception], int]
) -> t.Tuple[t.Type[Exception], t.Optional[int]]:
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.
@ -736,7 +738,7 @@ class Scaffold:
:param exc_class_or_code: Any exception class, or an HTTP status
code as an integer.
"""
exc_class: t.Type[Exception]
exc_class: type[Exception]
if isinstance(exc_class_or_code, int):
try:

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import hashlib
import typing as t
from collections.abc import MutableMapping
@ -92,7 +94,7 @@ class NullSession(SecureCookieSession):
but fail on setting.
"""
def _fail(self, *args: t.Any, **kwargs: t.Any) -> "te.NoReturn":
def _fail(self, *args: t.Any, **kwargs: t.Any) -> te.NoReturn:
raise RuntimeError(
"The session is unavailable because no secret "
"key was set. Set the secret_key on the "
@ -153,7 +155,7 @@ class SessionInterface:
#: .. versionadded:: 0.10
pickle_based = False
def make_null_session(self, app: "Flask") -> NullSession:
def make_null_session(self, app: Flask) -> NullSession:
"""Creates a null session which acts as a replacement object if the
real session support could not be loaded due to a configuration
error. This mainly aids the user experience because the job of the
@ -174,11 +176,11 @@ class SessionInterface:
"""
return isinstance(obj, self.null_session_class)
def get_cookie_name(self, app: "Flask") -> str:
def get_cookie_name(self, app: Flask) -> str:
"""The name of the session cookie. Uses``app.config["SESSION_COOKIE_NAME"]``."""
return app.config["SESSION_COOKIE_NAME"]
def get_cookie_domain(self, app: "Flask") -> t.Optional[str]:
def get_cookie_domain(self, app: Flask) -> str | None:
"""The value of the ``Domain`` parameter on the session cookie. If not set,
browsers will only send the cookie to the exact domain it was set from.
Otherwise, they will send it to any subdomain of the given value as well.
@ -191,7 +193,7 @@ class SessionInterface:
rv = app.config["SESSION_COOKIE_DOMAIN"]
return rv if rv else None
def get_cookie_path(self, app: "Flask") -> str:
def get_cookie_path(self, app: Flask) -> str:
"""Returns the path for which the cookie should be valid. The
default implementation uses the value from the ``SESSION_COOKIE_PATH``
config var if it's set, and falls back to ``APPLICATION_ROOT`` or
@ -199,29 +201,27 @@ class SessionInterface:
"""
return app.config["SESSION_COOKIE_PATH"] or app.config["APPLICATION_ROOT"]
def get_cookie_httponly(self, app: "Flask") -> bool:
def get_cookie_httponly(self, app: Flask) -> bool:
"""Returns True if the session cookie should be httponly. This
currently just returns the value of the ``SESSION_COOKIE_HTTPONLY``
config var.
"""
return app.config["SESSION_COOKIE_HTTPONLY"]
def get_cookie_secure(self, app: "Flask") -> bool:
def get_cookie_secure(self, app: Flask) -> bool:
"""Returns True if the cookie should be secure. This currently
just returns the value of the ``SESSION_COOKIE_SECURE`` setting.
"""
return app.config["SESSION_COOKIE_SECURE"]
def get_cookie_samesite(self, app: "Flask") -> str:
def get_cookie_samesite(self, app: Flask) -> str:
"""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"]
def get_expiration_time(
self, app: "Flask", session: SessionMixin
) -> t.Optional[datetime]:
def get_expiration_time(self, app: Flask, session: SessionMixin) -> datetime | None:
"""A helper method that returns an expiration date for the session
or ``None`` if the session is linked to the browser session. The
default implementation returns now + the permanent session
@ -231,7 +231,7 @@ class SessionInterface:
return datetime.now(timezone.utc) + app.permanent_session_lifetime
return None
def should_set_cookie(self, app: "Flask", session: SessionMixin) -> bool:
def should_set_cookie(self, app: Flask, session: SessionMixin) -> bool:
"""Used by session backends to determine if a ``Set-Cookie`` header
should be set for this session cookie for this response. If the session
has been modified, the cookie is set. If the session is permanent and
@ -247,9 +247,7 @@ class SessionInterface:
session.permanent and app.config["SESSION_REFRESH_EACH_REQUEST"]
)
def open_session(
self, app: "Flask", request: "Request"
) -> t.Optional[SessionMixin]:
def open_session(self, app: Flask, request: Request) -> SessionMixin | None:
"""This is called at the beginning of each request, after
pushing the request context, before matching the URL.
@ -264,7 +262,7 @@ class SessionInterface:
raise NotImplementedError()
def save_session(
self, app: "Flask", session: SessionMixin, response: "Response"
self, app: Flask, session: SessionMixin, response: Response
) -> None:
"""This is called at the end of each request, after generating
a response, before removing the request context. It is skipped
@ -295,9 +293,7 @@ class SecureCookieSessionInterface(SessionInterface):
serializer = session_json_serializer
session_class = SecureCookieSession
def get_signing_serializer(
self, app: "Flask"
) -> t.Optional[URLSafeTimedSerializer]:
def get_signing_serializer(self, app: Flask) -> URLSafeTimedSerializer | None:
if not app.secret_key:
return None
signer_kwargs = dict(
@ -310,9 +306,7 @@ class SecureCookieSessionInterface(SessionInterface):
signer_kwargs=signer_kwargs,
)
def open_session(
self, app: "Flask", request: "Request"
) -> t.Optional[SecureCookieSession]:
def open_session(self, app: Flask, request: Request) -> SecureCookieSession | None:
s = self.get_signing_serializer(app)
if s is None:
return None
@ -327,7 +321,7 @@ class SecureCookieSessionInterface(SessionInterface):
return self.session_class()
def save_session(
self, app: "Flask", session: SessionMixin, response: "Response"
self, app: Flask, session: SessionMixin, response: Response
) -> None:
name = self.get_cookie_name(app)
domain = self.get_cookie_domain(app)

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import typing as t
from jinja2 import BaseLoader
@ -18,13 +20,13 @@ if t.TYPE_CHECKING: # pragma: no cover
from .scaffold import Scaffold
def _default_template_ctx_processor() -> t.Dict[str, t.Any]:
def _default_template_ctx_processor() -> dict[str, t.Any]:
"""Default template context processor. Injects `request`,
`session` and `g`.
"""
appctx = _cv_app.get(None)
reqctx = _cv_request.get(None)
rv: t.Dict[str, t.Any] = {}
rv: dict[str, t.Any] = {}
if appctx is not None:
rv["g"] = appctx.g
if reqctx is not None:
@ -39,7 +41,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: Flask, **options: t.Any) -> None:
if "loader" not in options:
options["loader"] = app.create_global_jinja_loader()
BaseEnvironment.__init__(self, **options)
@ -51,24 +53,22 @@ class DispatchingJinjaLoader(BaseLoader):
the blueprint folders.
"""
def __init__(self, app: "Flask") -> None:
def __init__(self, app: Flask) -> None:
self.app = app
def get_source( # type: ignore
self, environment: Environment, template: str
) -> t.Tuple[str, t.Optional[str], t.Optional[t.Callable]]:
) -> tuple[str, str | None, t.Callable | 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
) -> t.Tuple[str, t.Optional[str], t.Optional[t.Callable]]:
) -> tuple[str, str | None, t.Callable | None]:
attempts = []
rv: t.Optional[t.Tuple[str, t.Optional[str], t.Optional[t.Callable[[], bool]]]]
trv: t.Optional[
t.Tuple[str, t.Optional[str], t.Optional[t.Callable[[], bool]]]
] = None
rv: tuple[str, str | None, t.Callable[[], bool] | None] | None
trv: None | (tuple[str, str | None, t.Callable[[], bool] | None]) = None
for srcobj, loader in self._iter_loaders(template):
try:
@ -89,7 +89,7 @@ class DispatchingJinjaLoader(BaseLoader):
def _get_source_fast(
self, environment: Environment, template: str
) -> t.Tuple[str, t.Optional[str], t.Optional[t.Callable]]:
) -> tuple[str, str | None, t.Callable | None]:
for _srcobj, loader in self._iter_loaders(template):
try:
return loader.get_source(environment, template)
@ -99,7 +99,7 @@ class DispatchingJinjaLoader(BaseLoader):
def _iter_loaders(
self, template: str
) -> t.Generator[t.Tuple["Scaffold", BaseLoader], None, None]:
) -> t.Generator[tuple[Scaffold, BaseLoader], None, None]:
loader = self.app.jinja_loader
if loader is not None:
yield self.app, loader
@ -109,7 +109,7 @@ class DispatchingJinjaLoader(BaseLoader):
if loader is not None:
yield blueprint, loader
def list_templates(self) -> t.List[str]:
def list_templates(self) -> list[str]:
result = set()
loader = self.app.jinja_loader
if loader is not None:
@ -124,7 +124,7 @@ class DispatchingJinjaLoader(BaseLoader):
return list(result)
def _render(app: "Flask", template: Template, context: t.Dict[str, t.Any]) -> str:
def _render(app: Flask, template: Template, context: dict[str, t.Any]) -> str:
app.update_template_context(context)
before_render_template.send(
app, _async_wrapper=app.ensure_sync, template=template, context=context
@ -137,7 +137,7 @@ def _render(app: "Flask", template: Template, context: t.Dict[str, t.Any]) -> st
def render_template(
template_name_or_list: t.Union[str, Template, t.List[t.Union[str, Template]]],
template_name_or_list: str | Template | list[str | Template],
**context: t.Any,
) -> str:
"""Render a template by name with the given context.
@ -164,7 +164,7 @@ def render_template_string(source: str, **context: t.Any) -> str:
def _stream(
app: "Flask", template: Template, context: t.Dict[str, t.Any]
app: Flask, template: Template, context: dict[str, t.Any]
) -> t.Iterator[str]:
app.update_template_context(context)
before_render_template.send(
@ -187,7 +187,7 @@ def _stream(
def stream_template(
template_name_or_list: t.Union[str, Template, t.List[t.Union[str, Template]]],
template_name_or_list: str | Template | list[str | Template],
**context: t.Any,
) -> t.Iterator[str]:
"""Render a template by name with the given context as a stream.

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import typing as t
from contextlib import contextmanager
from contextlib import ExitStack
@ -43,11 +45,11 @@ class EnvironBuilder(werkzeug.test.EnvironBuilder):
def __init__(
self,
app: "Flask",
app: Flask,
path: str = "/",
base_url: t.Optional[str] = None,
subdomain: t.Optional[str] = None,
url_scheme: t.Optional[str] = None,
base_url: str | None = None,
subdomain: str | None = None,
url_scheme: str | None = None,
*args: t.Any,
**kwargs: t.Any,
) -> None:
@ -104,12 +106,12 @@ class FlaskClient(Client):
Basic usage is outlined in the :doc:`/testing` chapter.
"""
application: "Flask"
application: Flask
def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
super().__init__(*args, **kwargs)
self.preserve_context = False
self._new_contexts: t.List[t.ContextManager[t.Any]] = []
self._new_contexts: list[t.ContextManager[t.Any]] = []
self._context_stack = ExitStack()
self.environ_base = {
"REMOTE_ADDR": "127.0.0.1",
@ -199,7 +201,7 @@ class FlaskClient(Client):
buffered: bool = False,
follow_redirects: bool = False,
**kwargs: t.Any,
) -> "TestResponse":
) -> TestResponse:
if args and isinstance(
args[0], (werkzeug.test.EnvironBuilder, dict, BaseRequest)
):
@ -238,7 +240,7 @@ class FlaskClient(Client):
return response
def __enter__(self) -> "FlaskClient":
def __enter__(self) -> FlaskClient:
if self.preserve_context:
raise RuntimeError("Cannot nest client invocations")
self.preserve_context = True
@ -246,9 +248,9 @@ class FlaskClient(Client):
def __exit__(
self,
exc_type: t.Optional[type],
exc_value: t.Optional[BaseException],
tb: t.Optional[TracebackType],
exc_type: type | None,
exc_value: BaseException | None,
tb: TracebackType | None,
) -> None:
self.preserve_context = False
self._context_stack.close()
@ -260,7 +262,7 @@ class FlaskCliRunner(CliRunner):
:meth:`~flask.Flask.test_cli_runner`. See :ref:`testing-cli`.
"""
def __init__(self, app: "Flask", **kwargs: t.Any) -> None:
def __init__(self, app: Flask, **kwargs: t.Any) -> None:
self.app = app
super().__init__(**kwargs)

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import typing as t
if t.TYPE_CHECKING: # pragma: no cover

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import typing as t
from . import typing as ft
@ -45,12 +47,12 @@ class View:
#: The methods this view is registered for. Uses the same default
#: (``["GET", "HEAD", "OPTIONS"]``) as ``route`` and
#: ``add_url_rule`` by default.
methods: t.ClassVar[t.Optional[t.Collection[str]]] = None
methods: t.ClassVar[t.Collection[str] | None] = None
#: Control whether the ``OPTIONS`` method is handled automatically.
#: Uses the same default (``True``) as ``route`` and
#: ``add_url_rule`` by default.
provide_automatic_options: t.ClassVar[t.Optional[bool]] = None
provide_automatic_options: t.ClassVar[bool | None] = None
#: A list of decorators to apply, in order, to the generated view
#: function. Remember that ``@decorator`` syntax is applied bottom
@ -58,7 +60,7 @@ class View:
#: decorator.
#:
#: .. versionadded:: 0.8
decorators: t.ClassVar[t.List[t.Callable]] = []
decorators: t.ClassVar[list[t.Callable]] = []
#: Create a new instance of this view class for every request by
#: default. If a view subclass sets this to ``False``, the same

View file

@ -1,3 +1,5 @@
from __future__ import annotations
import typing as t
from werkzeug.exceptions import BadRequest
@ -37,20 +39,20 @@ class Request(RequestBase):
#: because the request was never internally bound.
#:
#: .. versionadded:: 0.6
url_rule: t.Optional["Rule"] = None
url_rule: Rule | None = None
#: A dict of view arguments that matched the request. If an exception
#: happened when matching, this will be ``None``.
view_args: t.Optional[t.Dict[str, t.Any]] = None
view_args: dict[str, t.Any] | None = None
#: If matching the URL failed, this is the exception that will be
#: raised / was raised as part of the request handling. This is
#: usually a :exc:`~werkzeug.exceptions.NotFound` exception or
#: something similar.
routing_exception: t.Optional[Exception] = None
routing_exception: Exception | None = None
@property
def max_content_length(self) -> t.Optional[int]: # type: ignore
def max_content_length(self) -> int | None: # type: ignore
"""Read-only view of the ``MAX_CONTENT_LENGTH`` config key."""
if current_app:
return current_app.config["MAX_CONTENT_LENGTH"]
@ -58,7 +60,7 @@ class Request(RequestBase):
return None
@property
def endpoint(self) -> t.Optional[str]:
def endpoint(self) -> str | None:
"""The endpoint that matched the request URL.
This will be ``None`` if matching failed or has not been
@ -73,7 +75,7 @@ class Request(RequestBase):
return None
@property
def blueprint(self) -> t.Optional[str]:
def blueprint(self) -> str | None:
"""The registered name of the current blueprint.
This will be ``None`` if the endpoint is not part of a
@ -92,7 +94,7 @@ class Request(RequestBase):
return None
@property
def blueprints(self) -> t.List[str]:
def blueprints(self) -> list[str]:
"""The registered names of the current blueprint upwards through
parent blueprints.
@ -123,7 +125,7 @@ class Request(RequestBase):
attach_enctype_error_multidict(self)
def on_json_loading_failed(self, e: t.Optional[ValueError]) -> t.Any:
def on_json_loading_failed(self, e: ValueError | None) -> t.Any:
try:
return super().on_json_loading_failed(e)
except BadRequest as e:
@ -151,7 +153,7 @@ class Response(ResponseBase):
Added :attr:`max_cookie_size`.
"""
default_mimetype: t.Optional[str] = "text/html"
default_mimetype: str | None = "text/html"
json_module = json