Skip to content
Hide whitespace changes
Inline Side-by-side
server-temp/venv/lib/python3.10/site-packages/flask/json/provider.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import dataclasses
import decimal
import json
import typing as t
import uuid
import weakref
from datetime import date
from werkzeug.http import http_date
if t.TYPE_CHECKING: # pragma: no cover
from werkzeug.sansio.response import Response
from ..sansio.app import App
class JSONProvider:
"""A standard set of JSON operations for an application. Subclasses
of this can be used to customize JSON behavior or use different
JSON libraries.
To implement a provider for a specific library, subclass this base
class and implement at least :meth:`dumps` and :meth:`loads`. All
other methods have default implementations.
To use a different provider, either subclass ``Flask`` and set
:attr:`~flask.Flask.json_provider_class` to a provider class, or set
:attr:`app.json <flask.Flask.json>` to an instance of the class.
:param app: An application instance. This will be stored as a
:class:`weakref.proxy` on the :attr:`_app` attribute.
.. versionadded:: 2.2
"""
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.
:param obj: The data to serialize.
:param kwargs: May be passed to the underlying JSON library.
"""
raise NotImplementedError
def dump(self, obj: t.Any, fp: t.IO[str], **kwargs: t.Any) -> None:
"""Serialize data as JSON and write to a file.
:param obj: The data to serialize.
:param fp: A file opened for writing text. Should use the UTF-8
encoding to be valid JSON.
:param kwargs: May be passed to the underlying JSON library.
"""
fp.write(self.dumps(obj, **kwargs))
def loads(self, s: str | bytes, **kwargs: t.Any) -> t.Any:
"""Deserialize data as JSON.
:param s: Text or UTF-8 bytes.
:param kwargs: May be passed to the underlying JSON library.
"""
raise NotImplementedError
def load(self, fp: t.IO[t.AnyStr], **kwargs: t.Any) -> t.Any:
"""Deserialize data as JSON read from a file.
:param fp: A file opened for reading text or UTF-8 bytes.
:param kwargs: May be passed to the underlying JSON library.
"""
return self.loads(fp.read(), **kwargs)
def _prepare_response_obj(
self, args: tuple[t.Any, ...], kwargs: dict[str, t.Any]
) -> t.Any:
if args and kwargs:
raise TypeError("app.json.response() takes either args or kwargs, not both")
if not args and not kwargs:
return None
if len(args) == 1:
return args[0]
return args or kwargs
def response(self, *args: t.Any, **kwargs: t.Any) -> Response:
"""Serialize the given arguments as JSON, and return a
:class:`~flask.Response` object with the ``application/json``
mimetype.
The :func:`~flask.json.jsonify` function calls this method for
the current application.
Either positional or keyword arguments can be given, not both.
If no arguments are given, ``None`` is serialized.
:param args: A single value to serialize, or multiple values to
treat as a list to serialize.
:param kwargs: Treat as a dict to serialize.
"""
obj = self._prepare_response_obj(args, kwargs)
return self._app.response_class(self.dumps(obj), mimetype="application/json")
def _default(o: t.Any) -> t.Any:
if isinstance(o, date):
return http_date(o)
if isinstance(o, (decimal.Decimal, uuid.UUID)):
return str(o)
if dataclasses and dataclasses.is_dataclass(o):
return dataclasses.asdict(o)
if hasattr(o, "__html__"):
return str(o.__html__())
raise TypeError(f"Object of type {type(o).__name__} is not JSON serializable")
class DefaultJSONProvider(JSONProvider):
"""Provide JSON operations using Python's built-in :mod:`json`
library. Serializes the following additional data types:
- :class:`datetime.datetime` and :class:`datetime.date` are
serialized to :rfc:`822` strings. This is the same as the HTTP
date format.
- :class:`uuid.UUID` is serialized to a string.
- :class:`dataclasses.dataclass` is passed to
:func:`dataclasses.asdict`.
- :class:`~markupsafe.Markup` (or any object with a ``__html__``
method) will call the ``__html__`` method to get a string.
"""
default: t.Callable[[t.Any], t.Any] = staticmethod(_default) # type: ignore[assignment]
"""Apply this function to any object that :meth:`json.dumps` does
not know how to serialize. It should return a valid JSON type or
raise a ``TypeError``.
"""
ensure_ascii = True
"""Replace non-ASCII characters with escape sequences. This may be
more compatible with some clients, but can be disabled for better
performance and size.
"""
sort_keys = True
"""Sort the keys in any serialized dicts. This may be useful for
some caching situations, but can be disabled for better performance.
When enabled, keys must all be strings, they are not converted
before sorting.
"""
compact: bool | None = None
"""If ``True``, or ``None`` out of debug mode, the :meth:`response`
output will not add indentation, newlines, or spaces. If ``False``,
or ``None`` in debug mode, it will use a non-compact representation.
"""
mimetype = "application/json"
"""The mimetype set in :meth:`response`."""
def dumps(self, obj: t.Any, **kwargs: t.Any) -> str:
"""Serialize data as JSON to a string.
Keyword arguments are passed to :func:`json.dumps`. Sets some
parameter defaults from the :attr:`default`,
:attr:`ensure_ascii`, and :attr:`sort_keys` attributes.
:param obj: The data to serialize.
:param kwargs: Passed to :func:`json.dumps`.
"""
kwargs.setdefault("default", self.default)
kwargs.setdefault("ensure_ascii", self.ensure_ascii)
kwargs.setdefault("sort_keys", self.sort_keys)
return json.dumps(obj, **kwargs)
def loads(self, s: str | bytes, **kwargs: t.Any) -> t.Any:
"""Deserialize data as JSON from a string or bytes.
:param s: Text or UTF-8 bytes.
:param kwargs: Passed to :func:`json.loads`.
"""
return json.loads(s, **kwargs)
def response(self, *args: t.Any, **kwargs: t.Any) -> Response:
"""Serialize the given arguments as JSON, and return a
:class:`~flask.Response` object with it. The response mimetype
will be "application/json" and can be changed with
:attr:`mimetype`.
If :attr:`compact` is ``False`` or debug mode is enabled, the
output will be formatted to be easier to read.
Either positional or keyword arguments can be given, not both.
If no arguments are given, ``None`` is serialized.
:param args: A single value to serialize, or multiple values to
treat as a list to serialize.
:param kwargs: Treat as a dict to serialize.
"""
obj = self._prepare_response_obj(args, kwargs)
dump_args: dict[str, t.Any] = {}
if (self.compact is None and self._app.debug) or self.compact is False:
dump_args.setdefault("indent", 2)
else:
dump_args.setdefault("separators", (",", ":"))
return self._app.response_class(
f"{self.dumps(obj, **dump_args)}\n", mimetype=self.mimetype
)
server-temp/venv/lib/python3.10/site-packages/flask/json/tag.py deleted 100644 → 0
View file @ 03bf2ef9
"""
Tagged JSON
~~~~~~~~~~~
A compact representation for lossless serialization of non-standard JSON
types. :class:`~flask.sessions.SecureCookieSessionInterface` uses this
to serialize the session data, but it may be useful in other places. It
can be extended to support other types.
.. autoclass:: TaggedJSONSerializer
:members:
.. autoclass:: JSONTag
:members:
Let's see an example that adds support for
:class:`~collections.OrderedDict`. Dicts don't have an order in JSON, so
to handle this we will dump the items as a list of ``[key, value]``
pairs. Subclass :class:`JSONTag` and give it the new key ``' od'`` to
identify the type. The session serializer processes dicts first, so
insert the new tag at the front of the order since ``OrderedDict`` must
be processed before ``dict``.
.. code-block:: python
from flask.json.tag import JSONTag
class TagOrderedDict(JSONTag):
__slots__ = ('serializer',)
key = ' od'
def check(self, value):
return isinstance(value, OrderedDict)
def to_json(self, value):
return [[k, self.serializer.tag(v)] for k, v in iteritems(value)]
def to_python(self, value):
return OrderedDict(value)
app.session_interface.serializer.register(TagOrderedDict, index=0)
"""
from __future__ import annotations
import typing as t
from base64 import b64decode
from base64 import b64encode
from datetime import datetime
from uuid import UUID
from markupsafe import Markup
from werkzeug.http import http_date
from werkzeug.http import parse_date
from ..json import dumps
from ..json import loads
class JSONTag:
"""Base class for defining type tags for :class:`TaggedJSONSerializer`."""
__slots__ = ("serializer",)
#: The tag to mark the serialized object with. If empty, this tag is
#: only used as an intermediate step during tagging.
key: str = ""
def __init__(self, serializer: TaggedJSONSerializer) -> None:
"""Create a tagger for the given serializer."""
self.serializer = serializer
def check(self, value: t.Any) -> bool:
"""Check if the given value should be tagged by this tag."""
raise NotImplementedError
def to_json(self, value: t.Any) -> t.Any:
"""Convert the Python object to an object that is a valid JSON type.
The tag will be added later."""
raise NotImplementedError
def to_python(self, value: t.Any) -> t.Any:
"""Convert the JSON representation back to the correct type. The tag
will already be removed."""
raise NotImplementedError
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)}
class TagDict(JSONTag):
"""Tag for 1-item dicts whose only key matches a registered tag.
Internally, the dict key is suffixed with `__`, and the suffix is removed
when deserializing.
"""
__slots__ = ()
key = " di"
def check(self, value: t.Any) -> bool:
return (
isinstance(value, dict)
and len(value) == 1
and next(iter(value)) in self.serializer.tags
)
def to_json(self, value: t.Any) -> t.Any:
key = next(iter(value))
return {f"{key}__": self.serializer.tag(value[key])}
def to_python(self, value: t.Any) -> t.Any:
key = next(iter(value))
return {key[:-2]: value[key]}
class PassDict(JSONTag):
__slots__ = ()
def check(self, value: t.Any) -> bool:
return isinstance(value, dict)
def to_json(self, value: t.Any) -> t.Any:
# JSON objects may only have string keys, so don't bother tagging the
# key here.
return {k: self.serializer.tag(v) for k, v in value.items()}
tag = to_json
class TagTuple(JSONTag):
__slots__ = ()
key = " t"
def check(self, value: t.Any) -> bool:
return isinstance(value, tuple)
def to_json(self, value: t.Any) -> t.Any:
return [self.serializer.tag(item) for item in value]
def to_python(self, value: t.Any) -> t.Any:
return tuple(value)
class PassList(JSONTag):
__slots__ = ()
def check(self, value: t.Any) -> bool:
return isinstance(value, list)
def to_json(self, value: t.Any) -> t.Any:
return [self.serializer.tag(item) for item in value]
tag = to_json
class TagBytes(JSONTag):
__slots__ = ()
key = " b"
def check(self, value: t.Any) -> bool:
return isinstance(value, bytes)
def to_json(self, value: t.Any) -> t.Any:
return b64encode(value).decode("ascii")
def to_python(self, value: t.Any) -> t.Any:
return b64decode(value)
class TagMarkup(JSONTag):
"""Serialize anything matching the :class:`~markupsafe.Markup` API by
having a ``__html__`` method to the result of that method. Always
deserializes to an instance of :class:`~markupsafe.Markup`."""
__slots__ = ()
key = " m"
def check(self, value: t.Any) -> bool:
return callable(getattr(value, "__html__", None))
def to_json(self, value: t.Any) -> t.Any:
return str(value.__html__())
def to_python(self, value: t.Any) -> t.Any:
return Markup(value)
class TagUUID(JSONTag):
__slots__ = ()
key = " u"
def check(self, value: t.Any) -> bool:
return isinstance(value, UUID)
def to_json(self, value: t.Any) -> t.Any:
return value.hex
def to_python(self, value: t.Any) -> t.Any:
return UUID(value)
class TagDateTime(JSONTag):
__slots__ = ()
key = " d"
def check(self, value: t.Any) -> bool:
return isinstance(value, datetime)
def to_json(self, value: t.Any) -> t.Any:
return http_date(value)
def to_python(self, value: t.Any) -> t.Any:
return parse_date(value)
class TaggedJSONSerializer:
"""Serializer that uses a tag system to compactly represent objects that
are not JSON types. Passed as the intermediate serializer to
:class:`itsdangerous.Serializer`.
The following extra types are supported:
* :class:`dict`
* :class:`tuple`
* :class:`bytes`
* :class:`~markupsafe.Markup`
* :class:`~uuid.UUID`
* :class:`~datetime.datetime`
"""
__slots__ = ("tags", "order")
#: Tag classes to bind when creating the serializer. Other tags can be
#: added later using :meth:`~register`.
default_tags = [
TagDict,
PassDict,
TagTuple,
PassList,
TagBytes,
TagMarkup,
TagUUID,
TagDateTime,
]
def __init__(self) -> None:
self.tags: dict[str, JSONTag] = {}
self.order: list[JSONTag] = []
for cls in self.default_tags:
self.register(cls)
def register(
self,
tag_class: type[JSONTag],
force: bool = False,
index: int | None = None,
) -> None:
"""Register a new tag with this serializer.
:param tag_class: tag class to register. Will be instantiated with this
serializer instance.
:param force: overwrite an existing tag. If false (default), a
:exc:`KeyError` is raised.
:param index: index to insert the new tag in the tag order. Useful when
the new tag is a special case of an existing tag. If ``None``
(default), the tag is appended to the end of the order.
:raise KeyError: if the tag key is already registered and ``force`` is
not true.
"""
tag = tag_class(self)
key = tag.key
if key:
if not force and key in self.tags:
raise KeyError(f"Tag '{key}' is already registered.")
self.tags[key] = tag
if index is None:
self.order.append(tag)
else:
self.order.insert(index, tag)
def tag(self, value: t.Any) -> t.Any:
"""Convert a value to a tagged representation if necessary."""
for tag in self.order:
if tag.check(value):
return tag.tag(value)
return value
def untag(self, value: dict[str, t.Any]) -> t.Any:
"""Convert a tagged representation back to the original type."""
if len(value) != 1:
return value
key = next(iter(value))
if key not in self.tags:
return value
return self.tags[key].to_python(value[key])
def _untag_scan(self, value: t.Any) -> t.Any:
if isinstance(value, dict):
# untag each item recursively
value = {k: self._untag_scan(v) for k, v in value.items()}
# untag the dict itself
value = self.untag(value)
elif isinstance(value, list):
# untag each item recursively
value = [self._untag_scan(item) for item in value]
return value
def dumps(self, value: t.Any) -> str:
"""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 self._untag_scan(loads(value))
server-temp/venv/lib/python3.10/site-packages/flask/logging.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import logging
import sys
import typing as t
from werkzeug.local import LocalProxy
from .globals import request
if t.TYPE_CHECKING: # pragma: no cover
from .sansio.app import App
@LocalProxy
def wsgi_errors_stream() -> t.TextIO:
"""Find the most appropriate error stream for the application. If a request
is active, log to ``wsgi.errors``, otherwise use ``sys.stderr``.
If you configure your own :class:`logging.StreamHandler`, you may want to
use this for the stream. If you are using file or dict configuration and
can't import this directly, you can refer to it as
``ext://flask.logging.wsgi_errors_stream``.
"""
if request:
return request.environ["wsgi.errors"] # type: ignore[no-any-return]
return sys.stderr
def has_level_handler(logger: logging.Logger) -> bool:
"""Check if there is a handler in the logging chain that will handle the
given logger's :meth:`effective level <~logging.Logger.getEffectiveLevel>`.
"""
level = logger.getEffectiveLevel()
current = logger
while current:
if any(handler.level <= level for handler in current.handlers):
return True
if not current.propagate:
break
current = current.parent # type: ignore
return False
#: Log messages to :func:`~flask.logging.wsgi_errors_stream` with the format
#: ``[%(asctime)s] %(levelname)s in %(module)s: %(message)s``.
default_handler = logging.StreamHandler(wsgi_errors_stream) # type: ignore
default_handler.setFormatter(
logging.Formatter("[%(asctime)s] %(levelname)s in %(module)s: %(message)s")
)
def create_logger(app: App) -> logging.Logger:
"""Get the Flask app's logger and configure it if needed.
The logger name will be the same as
:attr:`app.import_name <flask.Flask.name>`.
When :attr:`~flask.Flask.debug` is enabled, set the logger level to
:data:`logging.DEBUG` if it is not set.
If there is no handler for the logger's effective level, add a
:class:`~logging.StreamHandler` for
:func:`~flask.logging.wsgi_errors_stream` with a basic format.
"""
logger = logging.getLogger(app.name)
if app.debug and not logger.level:
logger.setLevel(logging.DEBUG)
if not has_level_handler(logger):
logger.addHandler(default_handler)
return logger
server-temp/venv/lib/python3.10/site-packages/flask/sansio/README.md deleted 100644 → 0
View file @ 03bf2ef9
# Sansio
This folder contains code that can be used by alternative Flask
implementations, for example Quart. The code therefore cannot do any
IO, nor be part of a likely IO path. Finally this code cannot use the
Flask globals.
server-temp/venv/lib/python3.10/site-packages/flask/sansio/app.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import logging
import os
import sys
import typing as t
from datetime import timedelta
from itertools import chain
from werkzeug.exceptions import Aborter
from werkzeug.exceptions import BadRequest
from werkzeug.exceptions import BadRequestKeyError
from werkzeug.routing import BuildError
from werkzeug.routing import Map
from werkzeug.routing import Rule
from werkzeug.sansio.response import Response
from werkzeug.utils import cached_property
from werkzeug.utils import redirect as _wz_redirect
from .. import typing as ft
from ..config import Config
from ..config import ConfigAttribute
from ..ctx import _AppCtxGlobals
from ..helpers import _split_blueprint_path
from ..helpers import get_debug_flag
from ..json.provider import DefaultJSONProvider
from ..json.provider import JSONProvider
from ..logging import create_logger
from ..templating import DispatchingJinjaLoader
from ..templating import Environment
from .scaffold import _endpoint_from_view_func
from .scaffold import find_package
from .scaffold import Scaffold
from .scaffold import setupmethod
if t.TYPE_CHECKING: # pragma: no cover
from werkzeug.wrappers import Response as BaseResponse
from ..testing import FlaskClient
from ..testing import FlaskCliRunner
from .blueprints import Blueprint
T_shell_context_processor = t.TypeVar(
"T_shell_context_processor", bound=ft.ShellContextProcessorCallable
)
T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable)
T_template_filter = t.TypeVar("T_template_filter", bound=ft.TemplateFilterCallable)
T_template_global = t.TypeVar("T_template_global", bound=ft.TemplateGlobalCallable)
T_template_test = t.TypeVar("T_template_test", bound=ft.TemplateTestCallable)
def _make_timedelta(value: timedelta | int | None) -> timedelta | None:
if value is None or isinstance(value, timedelta):
return value
return timedelta(seconds=value)
class App(Scaffold):
"""The flask object implements a WSGI application and acts as the central
object. It is passed the name of the module or package of the
application. Once it is created it will act as a central registry for
the view functions, the URL rules, template configuration and much more.
The name of the package is used to resolve resources from inside the
package or the folder the module is contained in depending on if the
package parameter resolves to an actual python package (a folder with
an :file:`__init__.py` file inside) or a standard module (just a ``.py`` file).
For more information about resource loading, see :func:`open_resource`.
Usually you create a :class:`Flask` instance in your main module or
in the :file:`__init__.py` file of your package like this::
from flask import Flask
app = Flask(__name__)
.. admonition:: About the First Parameter
The idea of the first parameter is to give Flask an idea of what
belongs to your application. This name is used to find resources
on the filesystem, can be used by extensions to improve debugging
information and a lot more.
So it's important what you provide there. If you are using a single
module, `__name__` is always the correct value. If you however are
using a package, it's usually recommended to hardcode the name of
your package there.
For example if your application is defined in :file:`yourapplication/app.py`
you should create it with one of the two versions below::
app = Flask('yourapplication')
app = Flask(__name__.split('.')[0])
Why is that? The application will work even with `__name__`, thanks
to how resources are looked up. However it will make debugging more
painful. Certain extensions can make assumptions based on the
import name of your application. For example the Flask-SQLAlchemy
extension will look for the code in your application that triggered
an SQL query in debug mode. If the import name is not properly set
up, that debugging information is lost. (For example it would only
pick up SQL queries in `yourapplication.app` and not
`yourapplication.views.frontend`)
.. versionadded:: 0.7
The `static_url_path`, `static_folder`, and `template_folder`
parameters were added.
.. versionadded:: 0.8
The `instance_path` and `instance_relative_config` parameters were
added.
.. versionadded:: 0.11
The `root_path` parameter was added.
.. versionadded:: 1.0
The ``host_matching`` and ``static_host`` parameters were added.
.. versionadded:: 1.0
The ``subdomain_matching`` parameter was added. Subdomain
matching needs to be enabled manually now. Setting
:data:`SERVER_NAME` does not implicitly enable it.
:param import_name: the name of the application package
:param static_url_path: can be used to specify a different path for the
static files on the web. Defaults to the name
of the `static_folder` folder.
:param static_folder: The folder with static files that is served at
``static_url_path``. Relative to the application ``root_path``
or an absolute path. Defaults to ``'static'``.
:param static_host: the host to use when adding the static route.
Defaults to None. Required when using ``host_matching=True``
with a ``static_folder`` configured.
:param host_matching: set ``url_map.host_matching`` attribute.
Defaults to False.
:param subdomain_matching: consider the subdomain relative to
:data:`SERVER_NAME` when matching routes. Defaults to False.
:param template_folder: the folder that contains the templates that should
be used by the application. Defaults to
``'templates'`` folder in the root path of the
application.
:param instance_path: An alternative instance path for the application.
By default the folder ``'instance'`` next to the
package or module is assumed to be the instance
path.
:param instance_relative_config: if set to ``True`` relative filenames
for loading the config are assumed to
be relative to the instance path instead
of the application root.
:param root_path: The path to the root of the application files.
This should only be set manually when it can't be detected
automatically, such as for namespace packages.
"""
#: The class of the object assigned to :attr:`aborter`, created by
#: :meth:`create_aborter`. That object is called by
#: :func:`flask.abort` to raise HTTP errors, and can be
#: called directly as well.
#:
#: Defaults to :class:`werkzeug.exceptions.Aborter`.
#:
#: .. versionadded:: 2.2
aborter_class = Aborter
#: The class that is used for the Jinja environment.
#:
#: .. versionadded:: 0.11
jinja_environment = Environment
#: The class that is used for the :data:`~flask.g` instance.
#:
#: Example use cases for a custom class:
#:
#: 1. Store arbitrary attributes on flask.g.
#: 2. Add a property for lazy per-request database connectors.
#: 3. Return None instead of AttributeError on unexpected attributes.
#: 4. Raise exception if an unexpected attr is set, a "controlled" flask.g.
#:
#: In Flask 0.9 this property was called `request_globals_class` but it
#: was changed in 0.10 to :attr:`app_ctx_globals_class` because the
#: flask.g object is now application context scoped.
#:
#: .. versionadded:: 0.10
app_ctx_globals_class = _AppCtxGlobals
#: The class that is used for the ``config`` attribute of this app.
#: Defaults to :class:`~flask.Config`.
#:
#: Example use cases for a custom class:
#:
#: 1. Default values for certain config options.
#: 2. Access to config values through attributes in addition to keys.
#:
#: .. versionadded:: 0.11
config_class = Config
#: The testing flag. Set this to ``True`` to enable the test mode of
#: Flask extensions (and in the future probably also Flask itself).
#: For example this might activate test helpers that have an
#: additional runtime cost which should not be enabled by default.
#:
#: If this is enabled and PROPAGATE_EXCEPTIONS is not changed from the
#: default it's implicitly enabled.
#:
#: This attribute can also be configured from the config with the
#: ``TESTING`` configuration key. Defaults to ``False``.
testing = ConfigAttribute[bool]("TESTING")
#: If a secret key is set, cryptographic components can use this to
#: sign cookies and other things. Set this to a complex random value
#: when you want to use the secure cookie for instance.
#:
#: This attribute can also be configured from the config with the
#: :data:`SECRET_KEY` configuration key. Defaults to ``None``.
secret_key = ConfigAttribute[t.Union[str, bytes, None]]("SECRET_KEY")
#: A :class:`~datetime.timedelta` which is used to set the expiration
#: date of a permanent session. The default is 31 days which makes a
#: permanent session survive for roughly one month.
#:
#: This attribute can also be configured from the config with the
#: ``PERMANENT_SESSION_LIFETIME`` configuration key. Defaults to
#: ``timedelta(days=31)``
permanent_session_lifetime = ConfigAttribute[timedelta](
"PERMANENT_SESSION_LIFETIME",
get_converter=_make_timedelta, # type: ignore[arg-type]
)
json_provider_class: type[JSONProvider] = DefaultJSONProvider
"""A subclass of :class:`~flask.json.provider.JSONProvider`. An
instance is created and assigned to :attr:`app.json` when creating
the app.
The default, :class:`~flask.json.provider.DefaultJSONProvider`, uses
Python's built-in :mod:`json` library. A different provider can use
a different JSON library.
.. versionadded:: 2.2
"""
#: Options that are passed to the Jinja environment in
#: :meth:`create_jinja_environment`. Changing these options after
#: the environment is created (accessing :attr:`jinja_env`) will
#: have no effect.
#:
#: .. versionchanged:: 1.1.0
#: This is a ``dict`` instead of an ``ImmutableDict`` to allow
#: easier configuration.
#:
jinja_options: dict[str, t.Any] = {}
#: The rule object to use for URL rules created. This is used by
#: :meth:`add_url_rule`. Defaults to :class:`werkzeug.routing.Rule`.
#:
#: .. versionadded:: 0.7
url_rule_class = Rule
#: The map object to use for storing the URL rules and routing
#: configuration parameters. Defaults to :class:`werkzeug.routing.Map`.
#:
#: .. versionadded:: 1.1.0
url_map_class = Map
#: The :meth:`test_client` method creates an instance of this test
#: client class. Defaults to :class:`~flask.testing.FlaskClient`.
#:
#: .. versionadded:: 0.7
test_client_class: type[FlaskClient] | None = None
#: The :class:`~click.testing.CliRunner` subclass, by default
#: :class:`~flask.testing.FlaskCliRunner` that is used by
#: :meth:`test_cli_runner`. Its ``__init__`` method should take a
#: Flask app object as the first argument.
#:
#: .. versionadded:: 1.0
test_cli_runner_class: type[FlaskCliRunner] | None = None
default_config: dict[str, t.Any]
response_class: type[Response]
def __init__(
self,
import_name: str,
static_url_path: str | None = None,
static_folder: str | os.PathLike[str] | None = "static",
static_host: str | None = None,
host_matching: bool = False,
subdomain_matching: bool = False,
template_folder: str | os.PathLike[str] | None = "templates",
instance_path: str | None = None,
instance_relative_config: bool = False,
root_path: str | None = None,
):
super().__init__(
import_name=import_name,
static_folder=static_folder,
static_url_path=static_url_path,
template_folder=template_folder,
root_path=root_path,
)
if instance_path is None:
instance_path = self.auto_find_instance_path()
elif not os.path.isabs(instance_path):
raise ValueError(
"If an instance path is provided it must be absolute."
" A relative path was given instead."
)
#: Holds the path to the instance folder.
#:
#: .. versionadded:: 0.8
self.instance_path = instance_path
#: The configuration dictionary as :class:`Config`. This behaves
#: exactly like a regular dictionary but supports additional methods
#: to load a config from files.
self.config = self.make_config(instance_relative_config)
#: An instance of :attr:`aborter_class` created by
#: :meth:`make_aborter`. This is called by :func:`flask.abort`
#: to raise HTTP errors, and can be called directly as well.
#:
#: .. versionadded:: 2.2
#: Moved from ``flask.abort``, which calls this object.
self.aborter = self.make_aborter()
self.json: JSONProvider = self.json_provider_class(self)
"""Provides access to JSON methods. Functions in ``flask.json``
will call methods on this provider when the application context
is active. Used for handling JSON requests and responses.
An instance of :attr:`json_provider_class`. Can be customized by
changing that attribute on a subclass, or by assigning to this
attribute afterwards.
The default, :class:`~flask.json.provider.DefaultJSONProvider`,
uses Python's built-in :mod:`json` library. A different provider
can use a different JSON library.
.. versionadded:: 2.2
"""
#: A list of functions that are called by
#: :meth:`handle_url_build_error` when :meth:`.url_for` raises a
#: :exc:`~werkzeug.routing.BuildError`. Each function is called
#: with ``error``, ``endpoint`` and ``values``. If a function
#: returns ``None`` or raises a ``BuildError``, it is skipped.
#: Otherwise, its return value is returned by ``url_for``.
#:
#: .. versionadded:: 0.9
self.url_build_error_handlers: list[
t.Callable[[Exception, str, dict[str, t.Any]], str]
] = []
#: A list of functions that are called when the application context
#: is destroyed. Since the application context is also torn down
#: if the request ends this is the place to store code that disconnects
#: from databases.
#:
#: .. versionadded:: 0.9
self.teardown_appcontext_funcs: list[ft.TeardownCallable] = []
#: A list of shell context processor functions that should be run
#: when a shell context is created.
#:
#: .. versionadded:: 0.11
self.shell_context_processors: list[ft.ShellContextProcessorCallable] = []
#: Maps registered blueprint names to blueprint objects. The
#: dict retains the order the blueprints were registered in.
#: Blueprints can be registered multiple times, this dict does
#: not track how often they were attached.
#:
#: .. versionadded:: 0.7
self.blueprints: dict[str, Blueprint] = {}
#: a place where extensions can store application specific state. For
#: example this is where an extension could store database engines and
#: similar things.
#:
#: The key must match the name of the extension module. For example in
#: case of a "Flask-Foo" extension in `flask_foo`, the key would be
#: ``'foo'``.
#:
#: .. versionadded:: 0.7
self.extensions: dict[str, t.Any] = {}
#: The :class:`~werkzeug.routing.Map` for this instance. You can use
#: this to change the routing converters after the class was created
#: but before any routes are connected. Example::
#:
#: from werkzeug.routing import BaseConverter
#:
#: class ListConverter(BaseConverter):
#: def to_python(self, value):
#: return value.split(',')
#: def to_url(self, values):
#: return ','.join(super(ListConverter, self).to_url(value)
#: for value in values)
#:
#: app = Flask(__name__)
#: app.url_map.converters['list'] = ListConverter
self.url_map = self.url_map_class(host_matching=host_matching)
self.subdomain_matching = subdomain_matching
# tracks internally if the application already handled at least one
# request.
self._got_first_request = False
# Set the name of the Click group in case someone wants to add
# the app's commands to another CLI tool.
self.cli.name = self.name
def _check_setup_finished(self, f_name: str) -> None:
if self._got_first_request:
raise AssertionError(
f"The setup method '{f_name}' can no longer be called"
" on the application. It has already handled its first"
" request, any changes will not be applied"
" consistently.\n"
"Make sure all imports, decorators, functions, etc."
" needed to set up the application are done before"
" running it."
)
@cached_property
def name(self) -> str: # type: ignore
"""The name of the application. This is usually the import name
with the difference that it's guessed from the run file if the
import name is main. This name is used as a display name when
Flask needs the name of the application. It can be set and overridden
to change the value.
.. versionadded:: 0.8
"""
if self.import_name == "__main__":
fn: str | None = getattr(sys.modules["__main__"], "__file__", None)
if fn is None:
return "__main__"
return os.path.splitext(os.path.basename(fn))[0]
return self.import_name
@cached_property
def logger(self) -> logging.Logger:
"""A standard Python :class:`~logging.Logger` for the app, with
the same name as :attr:`name`.
In debug mode, the logger's :attr:`~logging.Logger.level` will
be set to :data:`~logging.DEBUG`.
If there are no handlers configured, a default handler will be
added. See :doc:`/logging` for more information.
.. versionchanged:: 1.1.0
The logger takes the same name as :attr:`name` rather than
hard-coding ``"flask.app"``.
.. versionchanged:: 1.0.0
Behavior was simplified. The logger is always named
``"flask.app"``. The level is only set during configuration,
it doesn't check ``app.debug`` each time. Only one format is
used, not different ones depending on ``app.debug``. No
handlers are removed, and a handler is only added if no
handlers are already configured.
.. versionadded:: 0.3
"""
return create_logger(self)
@cached_property
def jinja_env(self) -> Environment:
"""The Jinja environment used to load templates.
The environment is created the first time this property is
accessed. Changing :attr:`jinja_options` after that will have no
effect.
"""
return self.create_jinja_environment()
def create_jinja_environment(self) -> Environment:
raise NotImplementedError()
def make_config(self, instance_relative: bool = False) -> Config:
"""Used to create the config attribute by the Flask constructor.
The `instance_relative` parameter is passed in from the constructor
of Flask (there named `instance_relative_config`) and indicates if
the config should be relative to the instance path or the root path
of the application.
.. versionadded:: 0.8
"""
root_path = self.root_path
if instance_relative:
root_path = self.instance_path
defaults = dict(self.default_config)
defaults["DEBUG"] = get_debug_flag()
return self.config_class(root_path, defaults)
def make_aborter(self) -> Aborter:
"""Create the object to assign to :attr:`aborter`. That object
is called by :func:`flask.abort` to raise HTTP errors, and can
be called directly as well.
By default, this creates an instance of :attr:`aborter_class`,
which defaults to :class:`werkzeug.exceptions.Aborter`.
.. versionadded:: 2.2
"""
return self.aborter_class()
def auto_find_instance_path(self) -> str:
"""Tries to locate the instance path if it was not provided to the
constructor of the application class. It will basically calculate
the path to a folder named ``instance`` next to your main file or
the package.
.. versionadded:: 0.8
"""
prefix, package_path = find_package(self.import_name)
if prefix is None:
return os.path.join(package_path, "instance")
return os.path.join(prefix, "var", f"{self.name}-instance")
def create_global_jinja_loader(self) -> DispatchingJinjaLoader:
"""Creates the loader for the Jinja2 environment. Can be used to
override just the loader and keeping the rest unchanged. It's
discouraged to override this function. Instead one should override
the :meth:`jinja_loader` function instead.
The global loader dispatches between the loaders of the application
and the individual blueprints.
.. versionadded:: 0.7
"""
return DispatchingJinjaLoader(self)
def select_jinja_autoescape(self, filename: str) -> bool:
"""Returns ``True`` if autoescaping should be active for the given
template name. If no template name is given, returns `True`.
.. versionchanged:: 2.2
Autoescaping is now enabled by default for ``.svg`` files.
.. versionadded:: 0.5
"""
if filename is None:
return True
return filename.endswith((".html", ".htm", ".xml", ".xhtml", ".svg"))
@property
def debug(self) -> bool:
"""Whether debug mode is enabled. When using ``flask run`` to start the
development server, an interactive debugger will be shown for unhandled
exceptions, and the server will be reloaded when code changes. This maps to the
:data:`DEBUG` config key. It may not behave as expected if set late.
**Do not enable debug mode when deploying in production.**
Default: ``False``
"""
return self.config["DEBUG"] # type: ignore[no-any-return]
@debug.setter
def debug(self, value: bool) -> None:
self.config["DEBUG"] = value
if self.config["TEMPLATES_AUTO_RELOAD"] is None:
self.jinja_env.auto_reload = value
@setupmethod
def register_blueprint(self, blueprint: Blueprint, **options: t.Any) -> None:
"""Register a :class:`~flask.Blueprint` on the application. Keyword
arguments passed to this method will override the defaults set on the
blueprint.
Calls the blueprint's :meth:`~flask.Blueprint.register` method after
recording the blueprint in the application's :attr:`blueprints`.
:param blueprint: The blueprint to register.
:param url_prefix: Blueprint routes will be prefixed with this.
:param subdomain: Blueprint routes will match on this subdomain.
:param url_defaults: Blueprint routes will use these default values for
view arguments.
:param options: Additional keyword arguments are passed to
:class:`~flask.blueprints.BlueprintSetupState`. They can be
accessed in :meth:`~flask.Blueprint.record` callbacks.
.. 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:: 0.7
"""
blueprint.register(self, options)
def iter_blueprints(self) -> t.ValuesView[Blueprint]:
"""Iterates over all blueprints by the order they were registered.
.. versionadded:: 0.11
"""
return self.blueprints.values()
@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:
if endpoint is None:
endpoint = _endpoint_from_view_func(view_func) # type: ignore
options["endpoint"] = endpoint
methods = options.pop("methods", None)
# if the methods are not given and the view_func object knows its
# methods we can use that instead. If neither exists, we go with
# a tuple of only ``GET`` as default.
if methods is None:
methods = getattr(view_func, "methods", None) or ("GET",)
if isinstance(methods, str):
raise TypeError(
"Allowed methods must be a list of strings, for"
' example: @app.route(..., methods=["POST"])'
)
methods = {item.upper() for item in methods}
# Methods that should always be added
required_methods = set(getattr(view_func, "required_methods", ()))
# starting with Flask 0.8 the view_func object can disable and
# force-enable the automatic options handling.
if provide_automatic_options is None:
provide_automatic_options = getattr(
view_func, "provide_automatic_options", None
)
if provide_automatic_options is None:
if "OPTIONS" not in methods:
provide_automatic_options = True
required_methods.add("OPTIONS")
else:
provide_automatic_options = False
# Add the required methods now.
methods |= required_methods
rule_obj = self.url_rule_class(rule, methods=methods, **options)
rule_obj.provide_automatic_options = provide_automatic_options # type: ignore[attr-defined]
self.url_map.add(rule_obj)
if view_func is not None:
old_func = self.view_functions.get(endpoint)
if old_func is not None and old_func != view_func:
raise AssertionError(
"View function mapping is overwriting an existing"
f" endpoint function: {endpoint}"
)
self.view_functions[endpoint] = view_func
@setupmethod
def template_filter(
self, name: str | None = None
) -> t.Callable[[T_template_filter], T_template_filter]:
"""A decorator that is used to register custom template filter.
You can specify a name for the filter, otherwise the function
name will be used. Example::
@app.template_filter()
def reverse(s):
return s[::-1]
: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_template_filter(f, name=name)
return f
return decorator
@setupmethod
def add_template_filter(
self, f: ft.TemplateFilterCallable, name: str | None = None
) -> None:
"""Register a custom template filter. Works exactly like the
:meth:`template_filter` decorator.
:param name: the optional name of the filter, otherwise the
function name will be used.
"""
self.jinja_env.filters[name or f.__name__] = f
@setupmethod
def template_test(
self, name: str | None = None
) -> t.Callable[[T_template_test], T_template_test]:
"""A decorator that is used to register custom template test.
You can specify a name for the test, otherwise the function
name will be used. Example::
@app.template_test()
def is_prime(n):
if n == 2:
return True
for i in range(2, int(math.ceil(math.sqrt(n))) + 1):
if n % i == 0:
return False
return True
.. 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_template_test(f, name=name)
return f
return decorator
@setupmethod
def add_template_test(
self, f: ft.TemplateTestCallable, name: str | None = None
) -> None:
"""Register a custom template test. Works exactly like the
:meth:`template_test` decorator.
.. versionadded:: 0.10
:param name: the optional name of the test, otherwise the
function name will be used.
"""
self.jinja_env.tests[name or f.__name__] = f
@setupmethod
def template_global(
self, name: str | None = None
) -> t.Callable[[T_template_global], T_template_global]:
"""A decorator that is used to register a custom template global function.
You can specify a name for the global function, otherwise the function
name will be used. Example::
@app.template_global()
def double(n):
return 2 * n
.. versionadded:: 0.10
:param name: the optional name of the global function, otherwise the
function name will be used.
"""
def decorator(f: T_template_global) -> T_template_global:
self.add_template_global(f, name=name)
return f
return decorator
@setupmethod
def add_template_global(
self, f: ft.TemplateGlobalCallable, name: str | None = None
) -> None:
"""Register a custom template global function. Works exactly like the
:meth:`template_global` decorator.
.. versionadded:: 0.10
:param name: the optional name of the global function, otherwise the
function name will be used.
"""
self.jinja_env.globals[name or f.__name__] = f
@setupmethod
def teardown_appcontext(self, f: T_teardown) -> T_teardown:
"""Registers a function to be called when the application
context is popped. The application context is typically popped
after the request context for each request, at the end of CLI
commands, or after a manually pushed context ends.
.. code-block:: python
with app.app_context():
...
When the ``with`` block exits (or ``ctx.pop()`` is called), the
teardown functions are called just before the app context is
made inactive. Since a request context typically also manages an
application context it would also be called when you pop a
request context.
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.
.. versionadded:: 0.9
"""
self.teardown_appcontext_funcs.append(f)
return f
@setupmethod
def shell_context_processor(
self, f: T_shell_context_processor
) -> T_shell_context_processor:
"""Registers a shell context processor function.
.. versionadded:: 0.11
"""
self.shell_context_processors.append(f)
return f
def _find_error_handler(
self, e: Exception, blueprints: list[str]
) -> ft.ErrorHandlerCallable | None:
"""Return a registered error handler for an exception in this order:
blueprint handler for a specific code, app handler for a specific code,
blueprint handler for an exception class, app handler for an exception
class, or ``None`` if a suitable handler is not found.
"""
exc_class, code = self._get_exc_class_and_code(type(e))
names = (*blueprints, None)
for c in (code, None) if code is not None else (None,):
for name in names:
handler_map = self.error_handler_spec[name][c]
if not handler_map:
continue
for cls in exc_class.__mro__:
handler = handler_map.get(cls)
if handler is not None:
return handler
return None
def trap_http_exception(self, e: Exception) -> bool:
"""Checks if an HTTP exception should be trapped or not. By default
this will return ``False`` for all exceptions except for a bad request
key error if ``TRAP_BAD_REQUEST_ERRORS`` is set to ``True``. It
also returns ``True`` if ``TRAP_HTTP_EXCEPTIONS`` is set to ``True``.
This is called for all HTTP exceptions raised by a view function.
If it returns ``True`` for any exception the error handler for this
exception is not called and it shows up as regular exception in the
traceback. This is helpful for debugging implicitly raised HTTP
exceptions.
.. versionchanged:: 1.0
Bad request errors are not trapped by default in debug mode.
.. versionadded:: 0.8
"""
if self.config["TRAP_HTTP_EXCEPTIONS"]:
return True
trap_bad_request = self.config["TRAP_BAD_REQUEST_ERRORS"]
# if unset, trap key errors in debug mode
if (
trap_bad_request is None
and self.debug
and isinstance(e, BadRequestKeyError)
):
return True
if trap_bad_request:
return isinstance(e, BadRequest)
return False
def should_ignore_error(self, error: BaseException | None) -> bool:
"""This is called to figure out if an error should be ignored
or not as far as the teardown system is concerned. If this
function returns ``True`` then the teardown handlers will not be
passed the error.
.. versionadded:: 0.10
"""
return False
def redirect(self, location: str, code: int = 302) -> BaseResponse:
"""Create a redirect response object.
This is called by :func:`flask.redirect`, and can be called
directly as well.
:param location: The URL to redirect to.
:param code: The status code for the redirect.
.. versionadded:: 2.2
Moved from ``flask.redirect``, which calls this method.
"""
return _wz_redirect(
location,
code=code,
Response=self.response_class, # type: ignore[arg-type]
)
def inject_url_defaults(self, endpoint: str, values: dict[str, t.Any]) -> None:
"""Injects the URL defaults for the given endpoint directly into
the values dictionary passed. This is used internally and
automatically called on URL building.
.. versionadded:: 0.7
"""
names: t.Iterable[str | None] = (None,)
# url_for may be called outside a request context, parse the
# passed endpoint instead of using request.blueprints.
if "." in endpoint:
names = chain(
names, reversed(_split_blueprint_path(endpoint.rpartition(".")[0]))
)
for name in names:
if name in self.url_default_functions:
for func in self.url_default_functions[name]:
func(endpoint, values)
def handle_url_build_error(
self, error: BuildError, endpoint: str, values: dict[str, t.Any]
) -> str:
"""Called by :meth:`.url_for` if a
:exc:`~werkzeug.routing.BuildError` was raised. If this returns
a value, it will be returned by ``url_for``, otherwise the error
will be re-raised.
Each function in :attr:`url_build_error_handlers` is called with
``error``, ``endpoint`` and ``values``. If a function returns
``None`` or raises a ``BuildError``, it is skipped. Otherwise,
its return value is returned by ``url_for``.
:param error: The active ``BuildError`` being handled.
:param endpoint: The endpoint being built.
:param values: The keyword arguments passed to ``url_for``.
"""
for handler in self.url_build_error_handlers:
try:
rv = handler(error, endpoint, values)
except BuildError as e:
# make error available outside except block
error = e
else:
if rv is not None:
return rv
# Re-raise if called with an active exception, otherwise raise
# the passed in exception.
if error is sys.exc_info()[1]:
raise
raise error
server-temp/venv/lib/python3.10/site-packages/flask/sansio/blueprints.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import os
import typing as t
from collections import defaultdict
from functools import update_wrapper
from .. import typing as ft
from .scaffold import _endpoint_from_view_func
from .scaffold import _sentinel
from .scaffold import Scaffold
from .scaffold import setupmethod
if t.TYPE_CHECKING: # pragma: no cover
from .app import App
DeferredSetupFunction = t.Callable[["BlueprintSetupState"], None]
T_after_request = t.TypeVar("T_after_request", bound=ft.AfterRequestCallable[t.Any])
T_before_request = t.TypeVar("T_before_request", bound=ft.BeforeRequestCallable)
T_error_handler = t.TypeVar("T_error_handler", bound=ft.ErrorHandlerCallable)
T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable)
T_template_context_processor = t.TypeVar(
"T_template_context_processor", bound=ft.TemplateContextProcessorCallable
)
T_template_filter = t.TypeVar("T_template_filter", bound=ft.TemplateFilterCallable)
T_template_global = t.TypeVar("T_template_global", bound=ft.TemplateGlobalCallable)
T_template_test = t.TypeVar("T_template_test", bound=ft.TemplateTestCallable)
T_url_defaults = t.TypeVar("T_url_defaults", bound=ft.URLDefaultCallable)
T_url_value_preprocessor = t.TypeVar(
"T_url_value_preprocessor", bound=ft.URLValuePreprocessorCallable
)
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.
"""
def __init__(
self,
blueprint: Blueprint,
app: App,
options: t.Any,
first_registration: bool,
) -> None:
#: a reference to the current application
self.app = app
#: a reference to the blueprint that created this setup state.
self.blueprint = blueprint
#: a dictionary with all options that were passed to the
#: :meth:`~flask.Flask.register_blueprint` method.
self.options = options
#: as blueprints can be registered multiple times with the
#: application and not everything wants to be registered
#: multiple times on it, this attribute can be used to figure
#: out if the blueprint was registered in the past already.
self.first_registration = first_registration
subdomain = self.options.get("subdomain")
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: ft.RouteCallable | None = None,
**options: t.Any,
) -> None:
"""A helper method to register a rule (and optionally a view function)
to the application. The endpoint is automatically prefixed with the
blueprint's name.
"""
if self.url_prefix is not None:
if rule:
rule = "/".join((self.url_prefix.rstrip("/"), rule.lstrip("/")))
else:
rule = self.url_prefix
options.setdefault("subdomain", self.subdomain)
if endpoint is None:
endpoint = _endpoint_from_view_func(view_func) # type: ignore
defaults = self.url_defaults
if "defaults" in options:
defaults = dict(defaults, **options.pop("defaults"))
self.app.add_url_rule(
rule,
f"{self.name_prefix}.{self.name}.{endpoint}".lstrip("."),
view_func,
defaults=defaults,
**options,
)
class Blueprint(Scaffold):
"""Represents a blueprint, a collection of routes and other
app-related functions that can be registered on a real application
later.
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.
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.
See :doc:`/blueprints` for more information.
: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.
.. 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[str] | None = None,
static_url_path: str | None = None,
template_folder: str | os.PathLike[str] | None = None,
url_prefix: str | None = None,
subdomain: str | None = None,
url_defaults: dict[str, t.Any] | None = None,
root_path: str | None = None,
cli_group: str | None = _sentinel, # type: ignore[assignment]
):
super().__init__(
import_name=import_name,
static_folder=static_folder,
static_url_path=static_url_path,
template_folder=template_folder,
root_path=root_path,
)
if not name:
raise ValueError("'name' may not be empty.")
if "." in name:
raise ValueError("'name' may not contain a dot '.' character.")
self.name = name
self.url_prefix = url_prefix
self.subdomain = subdomain
self.deferred_functions: list[DeferredSetupFunction] = []
if url_defaults is None:
url_defaults = {}
self.url_values_defaults = url_defaults
self.cli_group = cli_group
self._blueprints: list[tuple[Blueprint, dict[str, t.Any]]] = []
def _check_setup_finished(self, f_name: str) -> None:
if self._got_registered_once:
raise AssertionError(
f"The setup method '{f_name}' can no longer be called on the blueprint"
f" '{self.name}'. It has already been registered at least once, any"
" changes will not be applied consistently.\n"
"Make sure all imports, decorators, functions, etc. needed to set up"
" the blueprint are done before registering it."
)
@setupmethod
def record(self, func: DeferredSetupFunction) -> None:
"""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)
@setupmethod
def record_once(self, func: DeferredSetupFunction) -> None:
"""Works like :meth:`record` but wraps the function in another
function that will ensure the function is only called once. If the
blueprint is registered a second time on the application, the
function passed is not called.
"""
def wrapper(state: BlueprintSetupState) -> None:
if state.first_registration:
func(state)
self.record(update_wrapper(wrapper, func))
def make_setup_state(
self, app: App, options: dict[str, t.Any], first_registration: bool = False
) -> BlueprintSetupState:
"""Creates an instance of :meth:`~flask.blueprints.BlueprintSetupState`
object that is later passed to the register callback functions.
Subclasses can override this to return a subclass of the setup state.
"""
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: App, options: dict[str, t.Any]) -> None:
"""Called by :meth:`Flask.register_blueprint` to register all
views and callbacks registered on the blueprint with the
application. Creates a :class:`.BlueprintSetupState` and calls
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, # type: ignore[attr-defined]
endpoint="static",
)
# Merge blueprint data into parent.
if first_bp_registration or first_name_registration:
self._merge_blueprint_funcs(app, name)
for deferred in self.deferred_functions:
deferred(state)
cli_resolved_group = options.get("cli_group", self.cli_group)
if self.cli.commands:
if cli_resolved_group is None:
app.cli.commands.update(self.cli.commands)
elif cli_resolved_group is _sentinel:
self.cli.name = name
app.cli.add_command(self.cli)
else:
self.cli.name = cli_resolved_group
app.cli.add_command(self.cli)
for blueprint, bp_options in self._blueprints:
bp_options = bp_options.copy()
bp_url_prefix = bp_options.get("url_prefix")
bp_subdomain = bp_options.get("subdomain")
if bp_subdomain is None:
bp_subdomain = blueprint.subdomain
if state.subdomain is not None and bp_subdomain is not None:
bp_options["subdomain"] = bp_subdomain + "." + state.subdomain
elif bp_subdomain is not None:
bp_options["subdomain"] = bp_subdomain
elif state.subdomain is not None:
bp_options["subdomain"] = state.subdomain
if bp_url_prefix is None:
bp_url_prefix = blueprint.url_prefix
if state.url_prefix is not None and bp_url_prefix is not None:
bp_options["url_prefix"] = (
state.url_prefix.rstrip("/") + "/" + bp_url_prefix.lstrip("/")
)
elif bp_url_prefix is not None:
bp_options["url_prefix"] = bp_url_prefix
elif state.url_prefix is not None:
bp_options["url_prefix"] = state.url_prefix
bp_options["name_prefix"] = name
blueprint.register(app, bp_options)
def _merge_blueprint_funcs(self, app: App, name: str) -> None:
def extend(
bp_dict: dict[ft.AppOrBlueprintKey, list[t.Any]],
parent_dict: dict[ft.AppOrBlueprintKey, list[t.Any]],
) -> None:
for key, values in bp_dict.items():
key = name if key is None else f"{name}.{key}"
parent_dict[key].extend(values)
for key, value in self.error_handler_spec.items():
key = name if key is None else f"{name}.{key}"
value = defaultdict(
dict,
{
code: {exc_class: func for exc_class, func in code_values.items()}
for code, code_values in value.items()
},
)
app.error_handler_spec[key] = value
for endpoint, func in self.view_functions.items():
app.view_functions[endpoint] = func
extend(self.before_request_funcs, app.before_request_funcs)
extend(self.after_request_funcs, app.after_request_funcs)
extend(
self.teardown_request_funcs,
app.teardown_request_funcs,
)
extend(self.url_default_functions, app.url_default_functions)
extend(self.url_value_preprocessors, app.url_value_preprocessors)
extend(self.template_context_processors, app.template_context_processors)
@setupmethod
def add_url_rule(
self,
rule: str,
endpoint: str | None = None,
view_func: ft.RouteCallable | None = None,
provide_automatic_options: bool | None = None,
**options: t.Any,
) -> None:
"""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:
def from_blueprint(state: BlueprintSetupState) -> None:
state.app.errorhandler(code)(f)
self.record_once(from_blueprint)
return f
return decorator
@setupmethod
def app_url_value_preprocessor(
self, f: T_url_value_preprocessor
) -> T_url_value_preprocessor:
"""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
server-temp/venv/lib/python3.10/site-packages/flask/sansio/scaffold.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import importlib.util
import os
import pathlib
import sys
import typing as t
from collections import defaultdict
from functools import update_wrapper
import click
from jinja2 import BaseLoader
from jinja2 import FileSystemLoader
from werkzeug.exceptions import default_exceptions
from werkzeug.exceptions import HTTPException
from werkzeug.utils import cached_property
from .. import typing as ft
from ..cli import AppGroup
from ..helpers import get_root_path
from ..templating import _default_template_ctx_processor
# a singleton sentinel value for parameter defaults
_sentinel = object()
F = t.TypeVar("F", bound=t.Callable[..., t.Any])
T_after_request = t.TypeVar("T_after_request", bound=ft.AfterRequestCallable[t.Any])
T_before_request = t.TypeVar("T_before_request", bound=ft.BeforeRequestCallable)
T_error_handler = t.TypeVar("T_error_handler", bound=ft.ErrorHandlerCallable)
T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable)
T_template_context_processor = t.TypeVar(
"T_template_context_processor", bound=ft.TemplateContextProcessorCallable
)
T_url_defaults = t.TypeVar("T_url_defaults", bound=ft.URLDefaultCallable)
T_url_value_preprocessor = t.TypeVar(
"T_url_value_preprocessor", bound=ft.URLValuePreprocessorCallable
)
T_route = t.TypeVar("T_route", bound=ft.RouteCallable)
def setupmethod(f: F) -> F:
f_name = f.__name__
def wrapper_func(self: Scaffold, *args: t.Any, **kwargs: t.Any) -> t.Any:
self._check_setup_finished(f_name)
return f(self, *args, **kwargs)
return t.cast(F, update_wrapper(wrapper_func, f))
class Scaffold:
"""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[str] | None = None,
static_url_path: str | None = None,
template_folder: str | os.PathLike[str] | None = None,
root_path: str | None = None,
):
#: The name of the package or module that this object belongs
#: to. Do not change this once it is set by the constructor.
self.import_name = import_name
self.static_folder = static_folder # 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: click.Group = 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, ft.RouteCallable] = {}
#: A data structure of registered error handlers, in the format
#: ``{scope: {code: {class: handler}}}``. The ``scope`` key is
#: the name of a blueprint the handlers are active for, or
#: ``None`` for all requests. The ``code`` key is the HTTP
#: status code for ``HTTPException``, or ``None`` for
#: other exceptions. The innermost dictionary maps exception
#: classes to handler functions.
#:
#: To register an error handler, use the :meth:`errorhandler`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.error_handler_spec: dict[
ft.AppOrBlueprintKey,
dict[int | None, dict[type[Exception], ft.ErrorHandlerCallable]],
] = defaultdict(lambda: defaultdict(dict))
#: A data structure of functions to call at the beginning of
#: each request, in the format ``{scope: [functions]}``. The
#: ``scope`` key is the name of a blueprint the functions are
#: active for, or ``None`` for all requests.
#:
#: To register a function, use the :meth:`before_request`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.before_request_funcs: dict[
ft.AppOrBlueprintKey, list[ft.BeforeRequestCallable]
] = defaultdict(list)
#: A data structure of functions to call at the end of each
#: request, in the format ``{scope: [functions]}``. The
#: ``scope`` key is the name of a blueprint the functions are
#: active for, or ``None`` for all requests.
#:
#: To register a function, use the :meth:`after_request`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.after_request_funcs: dict[
ft.AppOrBlueprintKey, list[ft.AfterRequestCallable[t.Any]]
] = defaultdict(list)
#: A data structure of functions to call at the end of each
#: request even if an exception is raised, in the format
#: ``{scope: [functions]}``. The ``scope`` key is the name of a
#: blueprint the functions are active for, or ``None`` for all
#: requests.
#:
#: To register a function, use the :meth:`teardown_request`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.teardown_request_funcs: dict[
ft.AppOrBlueprintKey, list[ft.TeardownCallable]
] = defaultdict(list)
#: A data structure of functions to call to pass extra context
#: values when rendering templates, in the format
#: ``{scope: [functions]}``. The ``scope`` key is the name of a
#: blueprint the functions are active for, or ``None`` for all
#: requests.
#:
#: To register a function, use the :meth:`context_processor`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.template_context_processors: dict[
ft.AppOrBlueprintKey, list[ft.TemplateContextProcessorCallable]
] = defaultdict(list, {None: [_default_template_ctx_processor]})
#: A data structure of functions to call to modify the keyword
#: arguments passed to the view function, in the format
#: ``{scope: [functions]}``. The ``scope`` key is the name of a
#: blueprint the functions are active for, or ``None`` for all
#: requests.
#:
#: To register a function, use the
#: :meth:`url_value_preprocessor` decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.url_value_preprocessors: dict[
ft.AppOrBlueprintKey,
list[ft.URLValuePreprocessorCallable],
] = defaultdict(list)
#: A data structure of functions to call to modify the keyword
#: arguments when generating URLs, in the format
#: ``{scope: [functions]}``. The ``scope`` key is the name of a
#: blueprint the functions are active for, or ``None`` for all
#: requests.
#:
#: To register a function, use the :meth:`url_defaults`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.url_default_functions: dict[
ft.AppOrBlueprintKey, list[ft.URLDefaultCallable]
] = defaultdict(list)
def __repr__(self) -> str:
return f"<{type(self).__name__} {self.name!r}>"
def _check_setup_finished(self, f_name: str) -> None:
raise NotImplementedError
@property
def static_folder(self) -> str | None:
"""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[str] | None) -> None:
if value is not None:
value = os.fspath(value).rstrip(r"\/")
self._static_folder = value
@property
def has_static_folder(self) -> bool:
"""``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
@cached_property
def jinja_loader(self) -> BaseLoader | None:
"""The Jinja loader for this object's templates. By default this
is a class :class:`jinja2.loaders.FileSystemLoader` to
:attr:`template_folder` if it is set.
.. versionadded:: 0.5
"""
if self.template_folder is not None:
return FileSystemLoader(os.path.join(self.root_path, self.template_folder))
else:
return None
def _method_route(
self,
method: str,
rule: str,
options: dict[str, t.Any],
) -> t.Callable[[T_route], T_route]:
if "methods" in options:
raise TypeError("Use the 'route' decorator to use the 'methods' argument.")
return self.route(rule, methods=[method], **options)
@setupmethod
def get(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""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: ft.RouteCallable) -> 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: str) -> str:
"""Find the path that contains the package or module."""
root_mod_name, _, _ = import_name.partition(".")
try:
root_spec = importlib.util.find_spec(root_mod_name)
if root_spec is None:
raise ValueError("not found")
except (ImportError, ValueError):
# ImportError: the machinery told us it does not exist
# ValueError:
# - the module name was invalid
# - the module name is __main__
# - we raised `ValueError` due to `root_spec` being `None`
return os.getcwd()
if root_spec.submodule_search_locations:
if root_spec.origin is None or root_spec.origin == "namespace":
# namespace package
package_spec = importlib.util.find_spec(import_name)
if package_spec is not None and package_spec.submodule_search_locations:
# Pick the path in the namespace that contains the submodule.
package_path = pathlib.Path(
os.path.commonpath(package_spec.submodule_search_locations)
)
search_location = next(
location
for location in root_spec.submodule_search_locations
if _path_is_relative_to(package_path, location)
)
else:
# Pick the first path.
search_location = root_spec.submodule_search_locations[0]
return os.path.dirname(search_location)
else:
# package with __init__.py
return os.path.dirname(os.path.dirname(root_spec.origin))
else:
# module
return os.path.dirname(root_spec.origin) # type: ignore[type-var, return-value]
def find_package(import_name: str) -> tuple[str | None, str]:
"""Find the prefix that a package is installed under, and the path
that it would be imported from.
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
server-temp/venv/lib/python3.10/site-packages/flask/sessions.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import hashlib
import typing as t
from collections.abc import MutableMapping
from datetime import datetime
from datetime import timezone
from itsdangerous import BadSignature
from itsdangerous import URLSafeTimedSerializer
from werkzeug.datastructures import CallbackDict
from .json.tag import TaggedJSONSerializer
if t.TYPE_CHECKING: # pragma: no cover
import typing_extensions as te
from .app import Flask
from .wrappers import Request
from .wrappers import Response
# TODO generic when Python > 3.8
class SessionMixin(MutableMapping): # type: ignore[type-arg]
"""Expands a basic dictionary with session attributes."""
@property
def permanent(self) -> bool:
"""This reflects the ``'_permanent'`` key in the dict."""
return self.get("_permanent", False)
@permanent.setter
def permanent(self, value: bool) -> None:
self["_permanent"] = bool(value)
#: Some implementations can detect whether a session is newly
#: created, but that is not guaranteed. Use with caution. The mixin
# default is hard-coded ``False``.
new = False
#: Some implementations can detect changes to the session and set
#: this when that happens. The mixin default is hard coded to
#: ``True``.
modified = True
#: Some implementations can detect when session data is read or
#: written and set this when that happens. The mixin default is hard
#: coded to ``True``.
accessed = True
# 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
:attr:`accessed` attributes. It cannot reliably track whether a
session is new (vs. empty), so :attr:`new` remains hard coded to
``False``.
"""
#: When data is changed, this is set to ``True``. Only the session
#: dictionary itself is tracked; if the session contains mutable
#: data (for example a nested dict) then this must be set to
#: ``True`` manually when modifying that data. The session cookie
#: will only be written to the response if this is ``True``.
modified = False
#: When data is read or written, this is set to ``True``. Used by
# :class:`.SecureCookieSessionInterface` to add a ``Vary: Cookie``
#: header, which allows caching proxies to cache different pages for
#: different users.
accessed = False
def __init__(self, initial: t.Any = None) -> None:
def on_update(self: te.Self) -> None:
self.modified = True
self.accessed = True
super().__init__(initial, on_update)
def __getitem__(self, key: str) -> t.Any:
self.accessed = True
return super().__getitem__(key)
def get(self, key: str, default: t.Any = None) -> t.Any:
self.accessed = True
return super().get(key, default)
def setdefault(self, key: str, default: t.Any = None) -> t.Any:
self.accessed = True
return super().setdefault(key, default)
class NullSession(SecureCookieSession):
"""Class used to generate nicer error messages if sessions are not
available. Will still allow read-only access to the empty session
but fail on setting.
"""
def _fail(self, *args: t.Any, **kwargs: t.Any) -> t.NoReturn:
raise RuntimeError(
"The session is unavailable because no secret "
"key was set. Set the secret_key on the "
"application to something unique and secret."
)
__setitem__ = __delitem__ = clear = pop = popitem = update = setdefault = _fail # type: ignore # noqa: B950
del _fail
class SessionInterface:
"""The basic interface you have to implement in order to replace the
default session interface which uses werkzeug's securecookie
implementation. The only methods you have to implement are
:meth:`open_session` and :meth:`save_session`, the others have
useful defaults which you don't need to change.
The session object returned by the :meth:`open_session` method has to
provide a dictionary like interface plus the properties and methods
from the :class:`SessionMixin`. We recommend just subclassing a dict
and adding that mixin::
class Session(dict, SessionMixin):
pass
If :meth:`open_session` returns ``None`` Flask will call into
:meth:`make_null_session` to create a session that acts as replacement
if the session support cannot work because some requirement is not
fulfilled. The default :class:`NullSession` class that is created
will complain that the secret key was not set.
To replace the session interface on an application all you have to do
is to assign :attr:`flask.Flask.session_interface`::
app = Flask(__name__)
app.session_interface = MySessionInterface()
Multiple requests with the same session may be sent and handled
concurrently. When implementing a new session interface, consider
whether reads or writes to the backing store must be synchronized.
There is no guarantee on the order in which the session for each
request is opened or saved, it will occur in the order that requests
begin and end processing.
.. versionadded:: 0.8
"""
#: :meth:`make_null_session` will look here for the class that should
#: be created when a null session is requested. Likewise the
#: :meth:`is_null_session` method will perform a typecheck against
#: this type.
null_session_class = NullSession
#: A flag that indicates if the session interface is pickle based.
#: This can be used by Flask extensions to make a decision in regards
#: to how to deal with the session object.
#:
#: .. versionadded:: 0.10
pickle_based = False
def make_null_session(self, app: Flask) -> NullSession:
"""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
null session is to still support lookup without complaining but
modifications are answered with a helpful error message of what
failed.
This creates an instance of :attr:`null_session_class` by default.
"""
return self.null_session_class()
def is_null_session(self, obj: object) -> bool:
"""Checks if a given object is a null session. Null sessions are
not asked to be saved.
This checks if the object is an instance of :attr:`null_session_class`
by default.
"""
return isinstance(obj, self.null_session_class)
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"] # 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,
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.
Uses the :data:`SESSION_COOKIE_DOMAIN` config.
.. versionchanged:: 2.3
Not set by default, does not fall back to ``SERVER_NAME``.
"""
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
default implementation uses the value from the ``SESSION_COOKIE_PATH``
config var if it's set, and falls back to ``APPLICATION_ROOT`` or
uses ``/`` if it's ``None``.
"""
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"] # 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"] # type: ignore[no-any-return]
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"] # 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
or ``None`` if the session is linked to the browser session. The
default implementation returns now + the permanent session
lifetime configured on the application.
"""
if session.permanent:
return datetime.now(timezone.utc) + app.permanent_session_lifetime
return None
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
the ``SESSION_REFRESH_EACH_REQUEST`` config is true, the cookie is
always set.
This check is usually skipped if the session was deleted.
.. versionadded:: 0.11
"""
return session.modified or (
session.permanent and app.config["SESSION_REFRESH_EACH_REQUEST"]
)
def open_session(self, app: Flask, request: Request) -> SessionMixin | None:
"""This is called at the beginning of each request, after
pushing the request context, before matching the URL.
This must return an object which implements a dictionary-like
interface as well as the :class:`SessionMixin` interface.
This will return ``None`` to indicate that loading failed in
some way that is not immediately an error. The request
context will fall back to using :meth:`make_null_session`
in this case.
"""
raise NotImplementedError()
def save_session(
self, app: Flask, session: SessionMixin, response: Response
) -> None:
"""This is called at the end of each request, after generating
a response, before removing the request context. It is skipped
if :meth:`is_null_session` returns ``True``.
"""
raise NotImplementedError()
session_json_serializer = TaggedJSONSerializer()
class SecureCookieSessionInterface(SessionInterface):
"""The default session interface that stores sessions in signed cookies
through the :mod:`itsdangerous` module.
"""
#: the salt that should be applied on top of the secret key for the
#: signing of cookie based sessions.
salt = "cookie-session"
#: the hash function to use for the signature. The default is sha1
digest_method = staticmethod(hashlib.sha1)
#: the name of the itsdangerous supported key derivation. The default
#: is hmac.
key_derivation = "hmac"
#: A python serializer for the payload. The default is a compact
#: JSON derived serializer with support for some extra Python types
#: such as datetime objects or tuples.
serializer = session_json_serializer
session_class = SecureCookieSession
def get_signing_serializer(self, app: Flask) -> URLSafeTimedSerializer | None:
if not app.secret_key:
return None
signer_kwargs = dict(
key_derivation=self.key_derivation, digest_method=self.digest_method
)
return URLSafeTimedSerializer(
app.secret_key,
salt=self.salt,
serializer=self.serializer,
signer_kwargs=signer_kwargs,
)
def open_session(self, app: Flask, request: Request) -> SecureCookieSession | None:
s = self.get_signing_serializer(app)
if s is None:
return None
val = request.cookies.get(self.get_cookie_name(app))
if not val:
return self.session_class()
max_age = int(app.permanent_session_lifetime.total_seconds())
try:
data = s.loads(val, max_age=max_age)
return self.session_class(data)
except BadSignature:
return self.session_class()
def save_session(
self, app: Flask, session: SessionMixin, response: Response
) -> None:
name = self.get_cookie_name(app)
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
secure = self.get_cookie_secure(app)
samesite = self.get_cookie_samesite(app)
httponly = self.get_cookie_httponly(app)
# Add a "Vary: Cookie" header if the session was accessed at all.
if session.accessed:
response.vary.add("Cookie")
# If the session is modified to be empty, remove the cookie.
# If the session is empty, return without setting the cookie.
if not session:
if session.modified:
response.delete_cookie(
name,
domain=domain,
path=path,
secure=secure,
samesite=samesite,
httponly=httponly,
)
response.vary.add("Cookie")
return
if not self.should_set_cookie(app, session):
return
expires = self.get_expiration_time(app, session)
val = self.get_signing_serializer(app).dumps(dict(session)) # type: ignore
response.set_cookie(
name,
val, # type: ignore
expires=expires,
httponly=httponly,
domain=domain,
path=path,
secure=secure,
samesite=samesite,
)
response.vary.add("Cookie")
server-temp/venv/lib/python3.10/site-packages/flask/signals.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
from blinker import Namespace
# This namespace is only for signals provided by Flask itself.
_signals = Namespace()
template_rendered = _signals.signal("template-rendered")
before_render_template = _signals.signal("before-render-template")
request_started = _signals.signal("request-started")
request_finished = _signals.signal("request-finished")
request_tearing_down = _signals.signal("request-tearing-down")
got_request_exception = _signals.signal("got-request-exception")
appcontext_tearing_down = _signals.signal("appcontext-tearing-down")
appcontext_pushed = _signals.signal("appcontext-pushed")
appcontext_popped = _signals.signal("appcontext-popped")
message_flashed = _signals.signal("message-flashed")
server-temp/venv/lib/python3.10/site-packages/flask/templating.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import typing as t
from jinja2 import BaseLoader
from jinja2 import Environment as BaseEnvironment
from jinja2 import Template
from jinja2 import TemplateNotFound
from .globals import _cv_app
from .globals import _cv_request
from .globals import current_app
from .globals import request
from .helpers import stream_with_context
from .signals import before_render_template
from .signals import template_rendered
if t.TYPE_CHECKING: # pragma: no cover
from .app import Flask
from .sansio.app import App
from .sansio.scaffold import Scaffold
def _default_template_ctx_processor() -> dict[str, t.Any]:
"""Default template context processor. Injects `request`,
`session` and `g`.
"""
appctx = _cv_app.get(None)
reqctx = _cv_request.get(None)
rv: dict[str, t.Any] = {}
if appctx is not None:
rv["g"] = appctx.g
if reqctx is not None:
rv["request"] = reqctx.request
rv["session"] = reqctx.session
return rv
class Environment(BaseEnvironment):
"""Works like a regular Jinja2 environment but has some additional
knowledge of how Flask's blueprint works so that it can prepend the
name of the blueprint to referenced templates if necessary.
"""
def __init__(self, app: App, **options: t.Any) -> None:
if "loader" not in options:
options["loader"] = app.create_global_jinja_loader()
BaseEnvironment.__init__(self, **options)
self.app = app
class DispatchingJinjaLoader(BaseLoader):
"""A loader that looks for templates in the application and all
the blueprint folders.
"""
def __init__(self, app: App) -> None:
self.app = app
def get_source(
self, environment: BaseEnvironment, template: str
) -> tuple[str, str | None, t.Callable[[], bool] | None]:
if self.app.config["EXPLAIN_TEMPLATE_LOADING"]:
return self._get_source_explained(environment, template)
return self._get_source_fast(environment, template)
def _get_source_explained(
self, environment: BaseEnvironment, template: str
) -> tuple[str, str | None, t.Callable[[], bool] | None]:
attempts = []
rv: tuple[str, str | None, t.Callable[[], bool] | None] | None
trv: None | (tuple[str, str | None, t.Callable[[], bool] | None]) = None
for srcobj, loader in self._iter_loaders(template):
try:
rv = loader.get_source(environment, template)
if trv is None:
trv = rv
except TemplateNotFound:
rv = None
attempts.append((loader, srcobj, rv))
from .debughelpers import explain_template_loading_attempts
explain_template_loading_attempts(self.app, template, attempts)
if trv is not None:
return trv
raise TemplateNotFound(template)
def _get_source_fast(
self, environment: BaseEnvironment, template: str
) -> tuple[str, str | None, t.Callable[[], bool] | None]:
for _srcobj, loader in self._iter_loaders(template):
try:
return loader.get_source(environment, template)
except TemplateNotFound:
continue
raise TemplateNotFound(template)
def _iter_loaders(self, template: str) -> t.Iterator[tuple[Scaffold, BaseLoader]]:
loader = self.app.jinja_loader
if loader is not None:
yield self.app, loader
for blueprint in self.app.iter_blueprints():
loader = blueprint.jinja_loader
if loader is not None:
yield blueprint, loader
def list_templates(self) -> list[str]:
result = set()
loader = self.app.jinja_loader
if loader is not None:
result.update(loader.list_templates())
for blueprint in self.app.iter_blueprints():
loader = blueprint.jinja_loader
if loader is not None:
for template in loader.list_templates():
result.add(template)
return list(result)
def _render(app: Flask, template: Template, context: dict[str, t.Any]) -> str:
app.update_template_context(context)
before_render_template.send(
app, _async_wrapper=app.ensure_sync, template=template, context=context
)
rv = template.render(context)
template_rendered.send(
app, _async_wrapper=app.ensure_sync, template=template, context=context
)
return rv
def render_template(
template_name_or_list: str | Template | list[str | Template],
**context: t.Any,
) -> str:
"""Render a template by name with the given context.
:param template_name_or_list: The name of the template to render. If
a list is given, the first name to exist will be rendered.
:param context: The variables to make available in the template.
"""
app = current_app._get_current_object() # type: ignore[attr-defined]
template = app.jinja_env.get_or_select_template(template_name_or_list)
return _render(app, template, context)
def render_template_string(source: str, **context: t.Any) -> str:
"""Render a template from the given source string with the given
context.
:param source: The source code of the template to render.
:param context: The variables to make available in the template.
"""
app = current_app._get_current_object() # type: ignore[attr-defined]
template = app.jinja_env.from_string(source)
return _render(app, template, context)
def _stream(
app: Flask, template: Template, context: dict[str, t.Any]
) -> t.Iterator[str]:
app.update_template_context(context)
before_render_template.send(
app, _async_wrapper=app.ensure_sync, template=template, context=context
)
def generate() -> t.Iterator[str]:
yield from template.generate(context)
template_rendered.send(
app, _async_wrapper=app.ensure_sync, template=template, context=context
)
rv = generate()
# If a request context is active, keep it while generating.
if request:
rv = stream_with_context(rv)
return rv
def stream_template(
template_name_or_list: str | Template | list[str | Template],
**context: t.Any,
) -> t.Iterator[str]:
"""Render a template by name with the given context as a stream.
This returns an iterator of strings, which can be used as a
streaming response from a view.
:param template_name_or_list: The name of the template to render. If
a list is given, the first name to exist will be rendered.
:param context: The variables to make available in the template.
.. versionadded:: 2.2
"""
app = current_app._get_current_object() # type: ignore[attr-defined]
template = app.jinja_env.get_or_select_template(template_name_or_list)
return _stream(app, template, context)
def stream_template_string(source: str, **context: t.Any) -> t.Iterator[str]:
"""Render a template from the given source string with the given
context as a stream. This returns an iterator of strings, which can
be used as a streaming response from a view.
:param source: The source code of the template to render.
:param context: The variables to make available in the template.
.. versionadded:: 2.2
"""
app = current_app._get_current_object() # type: ignore[attr-defined]
template = app.jinja_env.from_string(source)
return _stream(app, template, context)
server-temp/venv/lib/python3.10/site-packages/flask/testing.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import importlib.metadata
import typing as t
from contextlib import contextmanager
from contextlib import ExitStack
from copy import copy
from types import TracebackType
from urllib.parse import urlsplit
import werkzeug.test
from click.testing import CliRunner
from werkzeug.test import Client
from werkzeug.wrappers import Request as BaseRequest
from .cli import ScriptInfo
from .sessions import SessionMixin
if t.TYPE_CHECKING: # pragma: no cover
from _typeshed.wsgi import WSGIEnvironment
from werkzeug.test import TestResponse
from .app import Flask
class EnvironBuilder(werkzeug.test.EnvironBuilder):
"""An :class:`~werkzeug.test.EnvironBuilder`, that takes defaults from the
application.
:param app: The Flask application to configure the environment from.
:param path: URL path being requested.
:param base_url: Base URL where the app is being served, which
``path`` is relative to. If not given, built from
:data:`PREFERRED_URL_SCHEME`, ``subdomain``,
:data:`SERVER_NAME`, and :data:`APPLICATION_ROOT`.
:param subdomain: Subdomain name to append to :data:`SERVER_NAME`.
:param url_scheme: Scheme to use instead of
:data:`PREFERRED_URL_SCHEME`.
:param json: If given, this is serialized as JSON and passed as
``data``. Also defaults ``content_type`` to
``application/json``.
:param args: other positional arguments passed to
:class:`~werkzeug.test.EnvironBuilder`.
:param kwargs: other keyword arguments passed to
:class:`~werkzeug.test.EnvironBuilder`.
"""
def __init__(
self,
app: Flask,
path: str = "/",
base_url: str | None = None,
subdomain: str | None = None,
url_scheme: str | None = None,
*args: t.Any,
**kwargs: t.Any,
) -> None:
assert not (base_url or subdomain or url_scheme) or (
base_url is not None
) != bool(
subdomain or url_scheme
), 'Cannot pass "subdomain" or "url_scheme" with "base_url".'
if base_url is None:
http_host = app.config.get("SERVER_NAME") or "localhost"
app_root = app.config["APPLICATION_ROOT"]
if subdomain:
http_host = f"{subdomain}.{http_host}"
if url_scheme is None:
url_scheme = app.config["PREFERRED_URL_SCHEME"]
url = urlsplit(path)
base_url = (
f"{url.scheme or url_scheme}://{url.netloc or http_host}"
f"/{app_root.lstrip('/')}"
)
path = url.path
if url.query:
sep = b"?" if isinstance(url.query, bytes) else "?"
path += sep + url.query
self.app = app
super().__init__(path, base_url, *args, **kwargs)
def json_dumps(self, obj: t.Any, **kwargs: t.Any) -> str: # type: ignore
"""Serialize ``obj`` to a JSON-formatted string.
The serialization will be configured according to the config associated
with this EnvironBuilder's ``app``.
"""
return self.app.json.dumps(obj, **kwargs)
_werkzeug_version = ""
def _get_werkzeug_version() -> str:
global _werkzeug_version
if not _werkzeug_version:
_werkzeug_version = importlib.metadata.version("werkzeug")
return _werkzeug_version
class FlaskClient(Client):
"""Works like a regular Werkzeug test client but has knowledge about
Flask's contexts to defer the cleanup of the request context until
the end of a ``with`` block. For general information about how to
use this class refer to :class:`werkzeug.test.Client`.
.. versionchanged:: 0.12
`app.test_client()` includes preset default environment, which can be
set after instantiation of the `app.test_client()` object in
`client.environ_base`.
Basic usage is outlined in the :doc:`/testing` chapter.
"""
application: Flask
def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
super().__init__(*args, **kwargs)
self.preserve_context = False
self._new_contexts: list[t.ContextManager[t.Any]] = []
self._context_stack = ExitStack()
self.environ_base = {
"REMOTE_ADDR": "127.0.0.1",
"HTTP_USER_AGENT": f"Werkzeug/{_get_werkzeug_version()}",
}
@contextmanager
def session_transaction(
self, *args: t.Any, **kwargs: t.Any
) -> t.Iterator[SessionMixin]:
"""When used in combination with a ``with`` statement this opens a
session transaction. This can be used to modify the session that
the test client uses. Once the ``with`` block is left the session is
stored back.
::
with client.session_transaction() as session:
session['value'] = 42
Internally this is implemented by going through a temporary test
request context and since session handling could depend on
request variables this function accepts the same arguments as
:meth:`~flask.Flask.test_request_context` which are directly
passed through.
"""
if self._cookies is None:
raise TypeError(
"Cookies are disabled. Create a client with 'use_cookies=True'."
)
app = self.application
ctx = app.test_request_context(*args, **kwargs)
self._add_cookies_to_wsgi(ctx.request.environ)
with ctx:
sess = app.session_interface.open_session(app, ctx.request)
if sess is None:
raise RuntimeError("Session backend did not open a session.")
yield sess
resp = app.response_class()
if app.session_interface.is_null_session(sess):
return
with ctx:
app.session_interface.save_session(app, sess, resp)
self._update_cookies_from_response(
ctx.request.host.partition(":")[0],
ctx.request.path,
resp.headers.getlist("Set-Cookie"),
)
def _copy_environ(self, other: WSGIEnvironment) -> WSGIEnvironment:
out = {**self.environ_base, **other}
if self.preserve_context:
out["werkzeug.debug.preserve_context"] = self._new_contexts.append
return out
def _request_from_builder_args(
self, args: tuple[t.Any, ...], kwargs: dict[str, t.Any]
) -> BaseRequest:
kwargs["environ_base"] = self._copy_environ(kwargs.get("environ_base", {}))
builder = EnvironBuilder(self.application, *args, **kwargs)
try:
return builder.get_request()
finally:
builder.close()
def open(
self,
*args: t.Any,
buffered: bool = False,
follow_redirects: bool = False,
**kwargs: t.Any,
) -> TestResponse:
if args and isinstance(
args[0], (werkzeug.test.EnvironBuilder, dict, BaseRequest)
):
if isinstance(args[0], werkzeug.test.EnvironBuilder):
builder = copy(args[0])
builder.environ_base = self._copy_environ(builder.environ_base or {}) # type: ignore[arg-type]
request = builder.get_request()
elif isinstance(args[0], dict):
request = EnvironBuilder.from_environ(
args[0], app=self.application, environ_base=self._copy_environ({})
).get_request()
else:
# isinstance(args[0], BaseRequest)
request = copy(args[0])
request.environ = self._copy_environ(request.environ)
else:
# request is None
request = self._request_from_builder_args(args, kwargs)
# Pop any previously preserved contexts. This prevents contexts
# from being preserved across redirects or multiple requests
# within a single block.
self._context_stack.close()
response = super().open(
request,
buffered=buffered,
follow_redirects=follow_redirects,
)
response.json_module = self.application.json # type: ignore[assignment]
# Re-push contexts that were preserved during the request.
while self._new_contexts:
cm = self._new_contexts.pop()
self._context_stack.enter_context(cm)
return response
def __enter__(self) -> FlaskClient:
if self.preserve_context:
raise RuntimeError("Cannot nest client invocations")
self.preserve_context = True
return self
def __exit__(
self,
exc_type: type | None,
exc_value: BaseException | None,
tb: TracebackType | None,
) -> None:
self.preserve_context = False
self._context_stack.close()
class FlaskCliRunner(CliRunner):
"""A :class:`~click.testing.CliRunner` for testing a Flask app's
CLI commands. Typically created using
:meth:`~flask.Flask.test_cli_runner`. See :ref:`testing-cli`.
"""
def __init__(self, app: Flask, **kwargs: t.Any) -> None:
self.app = app
super().__init__(**kwargs)
def invoke( # type: ignore
self, cli: t.Any = None, args: t.Any = None, **kwargs: t.Any
) -> t.Any:
"""Invokes a CLI command in an isolated environment. See
:meth:`CliRunner.invoke <click.testing.CliRunner.invoke>` for
full method documentation. See :ref:`testing-cli` for examples.
If the ``obj`` argument is not given, passes an instance of
:class:`~flask.cli.ScriptInfo` that knows how to load the Flask
app being tested.
:param cli: Command object to invoke. Default is the app's
:attr:`~flask.app.Flask.cli` group.
:param args: List of strings to invoke the command with.
:return: a :class:`~click.testing.Result` object.
"""
if cli is None:
cli = self.app.cli
if "obj" not in kwargs:
kwargs["obj"] = ScriptInfo(create_app=lambda: self.app)
return super().invoke(cli, args, **kwargs)
server-temp/venv/lib/python3.10/site-packages/flask/typing.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
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.sansio.response import Response # noqa: F401
# The possible types that are directly convertible or are a Response object.
ResponseValue = t.Union[
"Response",
str,
bytes,
t.List[t.Any],
# Only dict is actually accepted, but Mapping allows for TypedDict.
t.Mapping[str, t.Any],
t.Iterator[str],
t.Iterator[bytes],
]
# the possible types for an individual HTTP header
# This should be a Union, but mypy doesn't pass unless it's a TypeVar.
HeaderValue = t.Union[str, t.List[str], t.Tuple[str, ...]]
# the possible types for HTTP headers
HeadersValue = t.Union[
"Headers",
t.Mapping[str, HeaderValue],
t.Sequence[t.Tuple[str, HeaderValue]],
]
# The possible types returned by a route function.
ResponseReturnValue = t.Union[
ResponseValue,
t.Tuple[ResponseValue, HeadersValue],
t.Tuple[ResponseValue, int],
t.Tuple[ResponseValue, int, HeadersValue],
"WSGIApplication",
]
# Allow any subclass of werkzeug.Response, such as the one from Flask,
# as a callback argument. Using werkzeug.Response directly makes a
# callback annotated with flask.Response fail type checking.
ResponseClass = t.TypeVar("ResponseClass", bound="Response")
AppOrBlueprintKey = t.Optional[str] # The App key is None, whereas blueprints are named
AfterRequestCallable = t.Union[
t.Callable[[ResponseClass], ResponseClass],
t.Callable[[ResponseClass], t.Awaitable[ResponseClass]],
]
BeforeFirstRequestCallable = t.Union[
t.Callable[[], None], t.Callable[[], t.Awaitable[None]]
]
BeforeRequestCallable = t.Union[
t.Callable[[], t.Optional[ResponseReturnValue]],
t.Callable[[], t.Awaitable[t.Optional[ResponseReturnValue]]],
]
ShellContextProcessorCallable = t.Callable[[], t.Dict[str, t.Any]]
TeardownCallable = t.Union[
t.Callable[[t.Optional[BaseException]], None],
t.Callable[[t.Optional[BaseException]], t.Awaitable[None]],
]
TemplateContextProcessorCallable = t.Union[
t.Callable[[], t.Dict[str, t.Any]],
t.Callable[[], t.Awaitable[t.Dict[str, t.Any]]],
]
TemplateFilterCallable = t.Callable[..., t.Any]
TemplateGlobalCallable = t.Callable[..., t.Any]
TemplateTestCallable = t.Callable[..., bool]
URLDefaultCallable = t.Callable[[str, t.Dict[str, t.Any]], None]
URLValuePreprocessorCallable = t.Callable[
[t.Optional[str], t.Optional[t.Dict[str, t.Any]]], None
]
# This should take Exception, but that either breaks typing the argument
# with a specific exception, or decorating multiple times with different
# exceptions (and using a union type on the argument).
# https://github.com/pallets/flask/issues/4095
# https://github.com/pallets/flask/issues/4295
# https://github.com/pallets/flask/issues/4297
ErrorHandlerCallable = t.Union[
t.Callable[[t.Any], ResponseReturnValue],
t.Callable[[t.Any], t.Awaitable[ResponseReturnValue]],
]
RouteCallable = t.Union[
t.Callable[..., ResponseReturnValue],
t.Callable[..., t.Awaitable[ResponseReturnValue]],
]
server-temp/venv/lib/python3.10/site-packages/flask/views.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import typing as t
from . import typing as ft
from .globals import current_app
from .globals import request
F = t.TypeVar("F", bound=t.Callable[..., t.Any])
http_method_funcs = frozenset(
["get", "post", "head", "options", "delete", "put", "trace", "patch"]
)
class View:
"""Subclass this class and override :meth:`dispatch_request` to
create a generic class-based view. Call :meth:`as_view` to create a
view function that creates an instance of the class with the given
arguments and calls its ``dispatch_request`` method with any URL
variables.
See :doc:`views` for a detailed guide.
.. code-block:: python
class Hello(View):
init_every_request = False
def dispatch_request(self, name):
return f"Hello, {name}!"
app.add_url_rule(
"/hello/<name>", view_func=Hello.as_view("hello")
)
Set :attr:`methods` on the class to change what methods the view
accepts.
Set :attr:`decorators` on the class to apply a list of decorators to
the generated view function. Decorators applied to the class itself
will not be applied to the generated view function!
Set :attr:`init_every_request` to ``False`` for efficiency, unless
you need to store request-global data on ``self``.
"""
#: The methods this view is registered for. Uses the same default
#: (``["GET", "HEAD", "OPTIONS"]``) as ``route`` and
#: ``add_url_rule`` by default.
methods: t.ClassVar[t.Collection[str] | None] = None
#: Control whether the ``OPTIONS`` method is handled automatically.
#: Uses the same default (``True``) as ``route`` and
#: ``add_url_rule`` by default.
provide_automatic_options: t.ClassVar[bool | None] = None
#: A list of decorators to apply, in order, to the generated view
#: function. Remember that ``@decorator`` syntax is applied bottom
#: to top, so the first decorator in the list would be the bottom
#: decorator.
#:
#: .. versionadded:: 0.8
decorators: t.ClassVar[list[t.Callable[[F], F]]] = []
#: Create a new instance of this view class for every request by
#: default. If a view subclass sets this to ``False``, the same
#: instance is used for every request.
#:
#: A single instance is more efficient, especially if complex setup
#: is done during init. However, storing data on ``self`` is no
#: longer safe across requests, and :data:`~flask.g` should be used
#: instead.
#:
#: .. versionadded:: 2.2
init_every_request: t.ClassVar[bool] = True
def dispatch_request(self) -> ft.ResponseReturnValue:
"""The actual view function behavior. Subclasses must override
this and return a valid response. Any variables from the URL
rule are passed as keyword arguments.
"""
raise NotImplementedError()
@classmethod
def as_view(
cls, name: str, *class_args: t.Any, **class_kwargs: t.Any
) -> ft.RouteCallable:
"""Convert the class into a view function that can be registered
for a route.
By default, the generated view will create a new instance of the
view class for every request and call its
:meth:`dispatch_request` method. If the view class sets
:attr:`init_every_request` to ``False``, the same instance will
be used for every request.
Except for ``name``, all other arguments passed to this method
are forwarded to the view class ``__init__`` method.
.. versionchanged:: 2.2
Added the ``init_every_request`` class attribute.
"""
if cls.init_every_request:
def view(**kwargs: t.Any) -> ft.ResponseReturnValue:
self = view.view_class( # type: ignore[attr-defined]
*class_args, **class_kwargs
)
return current_app.ensure_sync(self.dispatch_request)(**kwargs) # type: ignore[no-any-return]
else:
self = cls(*class_args, **class_kwargs)
def view(**kwargs: t.Any) -> ft.ResponseReturnValue:
return current_app.ensure_sync(self.dispatch_request)(**kwargs) # type: ignore[no-any-return]
if cls.decorators:
view.__name__ = name
view.__module__ = cls.__module__
for decorator in cls.decorators:
view = decorator(view)
# We attach the view class to the view function for two reasons:
# first of all it allows us to easily figure out what class-based
# view this thing came from, secondly it's also used for instantiating
# the view class so you can actually replace it with something else
# for testing purposes and debugging.
view.view_class = cls # type: ignore
view.__name__ = name
view.__doc__ = cls.__doc__
view.__module__ = cls.__module__
view.methods = cls.methods # type: ignore
view.provide_automatic_options = cls.provide_automatic_options # type: ignore
return view
class MethodView(View):
"""Dispatches request methods to the corresponding instance methods.
For example, if you implement a ``get`` method, it will be used to
handle ``GET`` requests.
This can be useful for defining a REST API.
:attr:`methods` is automatically set based on the methods defined on
the class.
See :doc:`views` for a detailed guide.
.. code-block:: python
class CounterAPI(MethodView):
def get(self):
return str(session.get("counter", 0))
def post(self):
session["counter"] = session.get("counter", 0) + 1
return redirect(url_for("counter"))
app.add_url_rule(
"/counter", view_func=CounterAPI.as_view("counter")
)
"""
def __init_subclass__(cls, **kwargs: t.Any) -> None:
super().__init_subclass__(**kwargs)
if "methods" not in cls.__dict__:
methods = set()
for base in cls.__bases__:
if getattr(base, "methods", None):
methods.update(base.methods) # type: ignore[attr-defined]
for key in http_method_funcs:
if hasattr(cls, key):
methods.add(key.upper())
if methods:
cls.methods = methods
def dispatch_request(self, **kwargs: t.Any) -> ft.ResponseReturnValue:
meth = getattr(self, request.method.lower(), None)
# If the request method is HEAD and we don't have a handler for it
# retry with GET.
if meth is None and request.method == "HEAD":
meth = getattr(self, "get", None)
assert meth is not None, f"Unimplemented method {request.method!r}"
return current_app.ensure_sync(meth)(**kwargs) # type: ignore[no-any-return]