Skip to content
Hide whitespace changes
Inline Side-by-side
server-temp/venv/lib/python3.10/site-packages/flask/app.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import collections.abc as cabc
import os
import sys
import typing as t
import weakref
from datetime import timedelta
from inspect import iscoroutinefunction
from itertools import chain
from types import TracebackType
from urllib.parse import quote as _url_quote
import click
from werkzeug.datastructures import Headers
from werkzeug.datastructures import ImmutableDict
from werkzeug.exceptions import BadRequestKeyError
from werkzeug.exceptions import HTTPException
from werkzeug.exceptions import InternalServerError
from werkzeug.routing import BuildError
from werkzeug.routing import MapAdapter
from werkzeug.routing import RequestRedirect
from werkzeug.routing import RoutingException
from werkzeug.routing import Rule
from werkzeug.serving import is_running_from_reloader
from werkzeug.wrappers import Response as BaseResponse
from . import cli
from . import typing as ft
from .ctx import AppContext
from .ctx import RequestContext
from .globals import _cv_app
from .globals import _cv_request
from .globals import current_app
from .globals import g
from .globals import request
from .globals import request_ctx
from .globals import session
from .helpers import get_debug_flag
from .helpers import get_flashed_messages
from .helpers import get_load_dotenv
from .helpers import send_from_directory
from .sansio.app import App
from .sansio.scaffold import _sentinel
from .sessions import SecureCookieSessionInterface
from .sessions import SessionInterface
from .signals import appcontext_tearing_down
from .signals import got_request_exception
from .signals import request_finished
from .signals import request_started
from .signals import request_tearing_down
from .templating import Environment
from .wrappers import Request
from .wrappers import Response
if t.TYPE_CHECKING: # pragma: no cover
from _typeshed.wsgi import StartResponse
from _typeshed.wsgi import WSGIEnvironment
from .testing import FlaskClient
from .testing import FlaskCliRunner
T_shell_context_processor = t.TypeVar(
"T_shell_context_processor", bound=ft.ShellContextProcessorCallable
)
T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable)
T_template_filter = t.TypeVar("T_template_filter", bound=ft.TemplateFilterCallable)
T_template_global = t.TypeVar("T_template_global", bound=ft.TemplateGlobalCallable)
T_template_test = t.TypeVar("T_template_test", bound=ft.TemplateTestCallable)
def _make_timedelta(value: timedelta | int | None) -> timedelta | None:
if value is None or isinstance(value, timedelta):
return value
return timedelta(seconds=value)
class Flask(App):
"""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.
"""
default_config = ImmutableDict(
{
"DEBUG": None,
"TESTING": False,
"PROPAGATE_EXCEPTIONS": None,
"SECRET_KEY": None,
"PERMANENT_SESSION_LIFETIME": timedelta(days=31),
"USE_X_SENDFILE": False,
"SERVER_NAME": None,
"APPLICATION_ROOT": "/",
"SESSION_COOKIE_NAME": "session",
"SESSION_COOKIE_DOMAIN": None,
"SESSION_COOKIE_PATH": None,
"SESSION_COOKIE_HTTPONLY": True,
"SESSION_COOKIE_SECURE": False,
"SESSION_COOKIE_SAMESITE": None,
"SESSION_REFRESH_EACH_REQUEST": True,
"MAX_CONTENT_LENGTH": None,
"SEND_FILE_MAX_AGE_DEFAULT": None,
"TRAP_BAD_REQUEST_ERRORS": None,
"TRAP_HTTP_EXCEPTIONS": False,
"EXPLAIN_TEMPLATE_LOADING": False,
"PREFERRED_URL_SCHEME": "http",
"TEMPLATES_AUTO_RELOAD": None,
"MAX_COOKIE_SIZE": 4093,
}
)
#: The class that is used for request objects. See :class:`~flask.Request`
#: for more information.
request_class: type[Request] = Request
#: The class that is used for response objects. See
#: :class:`~flask.Response` for more information.
response_class: type[Response] = Response
#: the session interface to use. By default an instance of
#: :class:`~flask.sessions.SecureCookieSessionInterface` is used here.
#:
#: .. versionadded:: 0.8
session_interface: SessionInterface = SecureCookieSessionInterface()
def __init__(
self,
import_name: str,
static_url_path: str | None = None,
static_folder: str | os.PathLike[str] | None = "static",
static_host: str | None = None,
host_matching: bool = False,
subdomain_matching: bool = False,
template_folder: str | os.PathLike[str] | None = "templates",
instance_path: str | None = None,
instance_relative_config: bool = False,
root_path: str | None = None,
):
super().__init__(
import_name=import_name,
static_url_path=static_url_path,
static_folder=static_folder,
static_host=static_host,
host_matching=host_matching,
subdomain_matching=subdomain_matching,
template_folder=template_folder,
instance_path=instance_path,
instance_relative_config=instance_relative_config,
root_path=root_path,
)
# Add a static route using the provided static_url_path, static_host,
# and static_folder if there is a configured static_folder.
# Note we do this without checking if static_folder exists.
# For one, it might be created while the server is running (e.g. during
# development). Also, Google App Engine stores static files somewhere
if self.has_static_folder:
assert (
bool(static_host) == host_matching
), "Invalid static_host/host_matching combination"
# Use a weakref to avoid creating a reference cycle between the app
# and the view function (see #3761).
self_ref = weakref.ref(self)
self.add_url_rule(
f"{self.static_url_path}/<path:filename>",
endpoint="static",
host=static_host,
view_func=lambda **kw: self_ref().send_static_file(**kw), # type: ignore # noqa: B950
)
def get_send_file_max_age(self, filename: str | None) -> int | None:
"""Used by :func:`send_file` to determine the ``max_age`` cache
value for a given file path if it wasn't passed.
By default, this returns :data:`SEND_FILE_MAX_AGE_DEFAULT` from
the configuration of :data:`~flask.current_app`. This defaults
to ``None``, which tells the browser to use conditional requests
instead of a timed cache, which is usually preferable.
Note this is a duplicate of the same method in the Flask
class.
.. versionchanged:: 2.0
The default configuration is ``None`` instead of 12 hours.
.. versionadded:: 0.9
"""
value = current_app.config["SEND_FILE_MAX_AGE_DEFAULT"]
if value is None:
return None
if isinstance(value, timedelta):
return int(value.total_seconds())
return value # type: ignore[no-any-return]
def send_static_file(self, filename: str) -> Response:
"""The view function used to serve files from
:attr:`static_folder`. A route is automatically registered for
this view at :attr:`static_url_path` if :attr:`static_folder` is
set.
Note this is a duplicate of the same method in the Flask
class.
.. versionadded:: 0.5
"""
if not self.has_static_folder:
raise RuntimeError("'static_folder' must be set to serve static_files.")
# send_file only knows to call get_send_file_max_age on the app,
# call it here so it works for blueprints too.
max_age = self.get_send_file_max_age(filename)
return send_from_directory(
t.cast(str, self.static_folder), filename, max_age=max_age
)
def open_resource(self, resource: str, mode: str = "rb") -> t.IO[t.AnyStr]:
"""Open a resource file relative to :attr:`root_path` for
reading.
For example, if the file ``schema.sql`` is next to the file
``app.py`` where the ``Flask`` app is defined, it can be opened
with:
.. code-block:: python
with app.open_resource("schema.sql") as f:
conn.executescript(f.read())
:param resource: Path to the resource relative to
:attr:`root_path`.
:param mode: Open the file in this mode. Only reading is
supported, valid values are "r" (or "rt") and "rb".
Note this is a duplicate of the same method in the Flask
class.
"""
if mode not in {"r", "rt", "rb"}:
raise ValueError("Resources can only be opened for reading.")
return open(os.path.join(self.root_path, resource), mode)
def open_instance_resource(self, resource: str, mode: str = "rb") -> t.IO[t.AnyStr]:
"""Opens a resource from the application's instance folder
(:attr:`instance_path`). Otherwise works like
:meth:`open_resource`. Instance resources can also be opened for
writing.
:param resource: the name of the resource. To access resources within
subfolders use forward slashes as separator.
:param mode: resource file opening mode, default is 'rb'.
"""
return open(os.path.join(self.instance_path, resource), mode)
def create_jinja_environment(self) -> Environment:
"""Create the Jinja environment based on :attr:`jinja_options`
and the various Jinja-related methods of the app. Changing
:attr:`jinja_options` after this will have no effect. Also adds
Flask-related globals and filters to the environment.
.. versionchanged:: 0.11
``Environment.auto_reload`` set in accordance with
``TEMPLATES_AUTO_RELOAD`` configuration option.
.. versionadded:: 0.5
"""
options = dict(self.jinja_options)
if "autoescape" not in options:
options["autoescape"] = self.select_jinja_autoescape
if "auto_reload" not in options:
auto_reload = self.config["TEMPLATES_AUTO_RELOAD"]
if auto_reload is None:
auto_reload = self.debug
options["auto_reload"] = auto_reload
rv = self.jinja_environment(self, **options)
rv.globals.update(
url_for=self.url_for,
get_flashed_messages=get_flashed_messages,
config=self.config,
# request, session and g are normally added with the
# context processor for efficiency reasons but for imported
# templates we also want the proxies in there.
request=request,
session=session,
g=g,
)
rv.policies["json.dumps_function"] = self.json.dumps
return rv
def create_url_adapter(self, request: Request | None) -> MapAdapter | None:
"""Creates a URL adapter for the given request. The URL adapter
is created at a point where the request context is not yet set
up so the request is passed explicitly.
.. versionadded:: 0.6
.. versionchanged:: 0.9
This can now also be called without a request object when the
URL adapter is created for the application context.
.. versionchanged:: 1.0
:data:`SERVER_NAME` no longer implicitly enables subdomain
matching. Use :attr:`subdomain_matching` instead.
"""
if request is not None:
# If subdomain matching is disabled (the default), use the
# default subdomain in all cases. This should be the default
# in Werkzeug but it currently does not have that feature.
if not self.subdomain_matching:
subdomain = self.url_map.default_subdomain or None
else:
subdomain = None
return self.url_map.bind_to_environ(
request.environ,
server_name=self.config["SERVER_NAME"],
subdomain=subdomain,
)
# We need at the very least the server name to be set for this
# to work.
if self.config["SERVER_NAME"] is not None:
return self.url_map.bind(
self.config["SERVER_NAME"],
script_name=self.config["APPLICATION_ROOT"],
url_scheme=self.config["PREFERRED_URL_SCHEME"],
)
return None
def raise_routing_exception(self, request: Request) -> t.NoReturn:
"""Intercept routing exceptions and possibly do something else.
In debug mode, intercept a routing redirect and replace it with
an error if the body will be discarded.
With modern Werkzeug this shouldn't occur, since it now uses a
308 status which tells the browser to resend the method and
body.
.. versionchanged:: 2.1
Don't intercept 307 and 308 redirects.
:meta private:
:internal:
"""
if (
not self.debug
or not isinstance(request.routing_exception, RequestRedirect)
or request.routing_exception.code in {307, 308}
or request.method in {"GET", "HEAD", "OPTIONS"}
):
raise request.routing_exception # type: ignore[misc]
from .debughelpers import FormDataRoutingRedirect
raise FormDataRoutingRedirect(request)
def update_template_context(self, context: dict[str, t.Any]) -> None:
"""Update the template context with some commonly used variables.
This injects request, session, config and g into the template
context as well as everything template context processors want
to inject. Note that the as of Flask 0.6, the original values
in the context will not be overridden if a context processor
decides to return a value with the same key.
:param context: the context as a dictionary that is updated in place
to add extra variables.
"""
names: t.Iterable[str | None] = (None,)
# A template may be rendered outside a request context.
if request:
names = chain(names, reversed(request.blueprints))
# The values passed to render_template take precedence. Keep a
# copy to re-apply after all context functions.
orig_ctx = context.copy()
for name in names:
if name in self.template_context_processors:
for func in self.template_context_processors[name]:
context.update(self.ensure_sync(func)())
context.update(orig_ctx)
def make_shell_context(self) -> dict[str, t.Any]:
"""Returns the shell context for an interactive shell for this
application. This runs all the registered shell context
processors.
.. versionadded:: 0.11
"""
rv = {"app": self, "g": g}
for processor in self.shell_context_processors:
rv.update(processor())
return rv
def run(
self,
host: str | None = None,
port: int | None = None,
debug: bool | None = None,
load_dotenv: bool = True,
**options: t.Any,
) -> None:
"""Runs the application on a local development server.
Do not use ``run()`` in a production setting. It is not intended to
meet security and performance requirements for a production server.
Instead, see :doc:`/deploying/index` for WSGI server recommendations.
If the :attr:`debug` flag is set the server will automatically reload
for code changes and show a debugger in case an exception happened.
If you want to run the application in debug mode, but disable the
code execution on the interactive debugger, you can pass
``use_evalex=False`` as parameter. This will keep the debugger's
traceback screen active, but disable code execution.
It is not recommended to use this function for development with
automatic reloading as this is badly supported. Instead you should
be using the :command:`flask` command line script's ``run`` support.
.. admonition:: Keep in Mind
Flask will suppress any server error with a generic error page
unless it is in debug mode. As such to enable just the
interactive debugger without the code reloading, you have to
invoke :meth:`run` with ``debug=True`` and ``use_reloader=False``.
Setting ``use_debugger`` to ``True`` without being in debug mode
won't catch any exceptions because there won't be any to
catch.
:param host: the hostname to listen on. Set this to ``'0.0.0.0'`` to
have the server available externally as well. Defaults to
``'127.0.0.1'`` or the host in the ``SERVER_NAME`` config variable
if present.
:param port: the port of the webserver. Defaults to ``5000`` or the
port defined in the ``SERVER_NAME`` config variable if present.
:param debug: if given, enable or disable debug mode. See
:attr:`debug`.
:param load_dotenv: Load the nearest :file:`.env` and :file:`.flaskenv`
files to set environment variables. Will also change the working
directory to the directory containing the first file found.
:param options: the options to be forwarded to the underlying Werkzeug
server. See :func:`werkzeug.serving.run_simple` for more
information.
.. versionchanged:: 1.0
If installed, python-dotenv will be used to load environment
variables from :file:`.env` and :file:`.flaskenv` files.
The :envvar:`FLASK_DEBUG` environment variable will override :attr:`debug`.
Threaded mode is enabled by default.
.. versionchanged:: 0.10
The default port is now picked from the ``SERVER_NAME``
variable.
"""
# Ignore this call so that it doesn't start another server if
# the 'flask run' command is used.
if os.environ.get("FLASK_RUN_FROM_CLI") == "true":
if not is_running_from_reloader():
click.secho(
" * Ignoring a call to 'app.run()' that would block"
" the current 'flask' CLI command.\n"
" Only call 'app.run()' in an 'if __name__ =="
' "__main__"\' guard.',
fg="red",
)
return
if get_load_dotenv(load_dotenv):
cli.load_dotenv()
# if set, env var overrides existing value
if "FLASK_DEBUG" in os.environ:
self.debug = get_debug_flag()
# debug passed to method overrides all other sources
if debug is not None:
self.debug = bool(debug)
server_name = self.config.get("SERVER_NAME")
sn_host = sn_port = None
if server_name:
sn_host, _, sn_port = server_name.partition(":")
if not host:
if sn_host:
host = sn_host
else:
host = "127.0.0.1"
if port or port == 0:
port = int(port)
elif sn_port:
port = int(sn_port)
else:
port = 5000
options.setdefault("use_reloader", self.debug)
options.setdefault("use_debugger", self.debug)
options.setdefault("threaded", True)
cli.show_server_banner(self.debug, self.name)
from werkzeug.serving import run_simple
try:
run_simple(t.cast(str, host), port, self, **options)
finally:
# reset the first request information if the development server
# reset normally. This makes it possible to restart the server
# without reloader and that stuff from an interactive shell.
self._got_first_request = False
def test_client(self, use_cookies: bool = True, **kwargs: t.Any) -> FlaskClient:
"""Creates a test client for this application. For information
about unit testing head over to :doc:`/testing`.
Note that if you are testing for assertions or exceptions in your
application code, you must set ``app.testing = True`` in order for the
exceptions to propagate to the test client. Otherwise, the exception
will be handled by the application (not visible to the test client) and
the only indication of an AssertionError or other exception will be a
500 status code response to the test client. See the :attr:`testing`
attribute. For example::
app.testing = True
client = app.test_client()
The test client can be used in a ``with`` block to defer the closing down
of the context until the end of the ``with`` block. This is useful if
you want to access the context locals for testing::
with app.test_client() as c:
rv = c.get('/?vodka=42')
assert request.args['vodka'] == '42'
Additionally, you may pass optional keyword arguments that will then
be passed to the application's :attr:`test_client_class` constructor.
For example::
from flask.testing import FlaskClient
class CustomClient(FlaskClient):
def __init__(self, *args, **kwargs):
self._authentication = kwargs.pop("authentication")
super(CustomClient,self).__init__( *args, **kwargs)
app.test_client_class = CustomClient
client = app.test_client(authentication='Basic ....')
See :class:`~flask.testing.FlaskClient` for more information.
.. versionchanged:: 0.4
added support for ``with`` block usage for the client.
.. versionadded:: 0.7
The `use_cookies` parameter was added as well as the ability
to override the client to be used by setting the
:attr:`test_client_class` attribute.
.. versionchanged:: 0.11
Added `**kwargs` to support passing additional keyword arguments to
the constructor of :attr:`test_client_class`.
"""
cls = self.test_client_class
if cls is None:
from .testing import FlaskClient as cls
return cls( # type: ignore
self, self.response_class, use_cookies=use_cookies, **kwargs
)
def test_cli_runner(self, **kwargs: t.Any) -> FlaskCliRunner:
"""Create a CLI runner for testing CLI commands.
See :ref:`testing-cli`.
Returns an instance of :attr:`test_cli_runner_class`, by default
:class:`~flask.testing.FlaskCliRunner`. The Flask app object is
passed as the first argument.
.. versionadded:: 1.0
"""
cls = self.test_cli_runner_class
if cls is None:
from .testing import FlaskCliRunner as cls
return cls(self, **kwargs) # type: ignore
def handle_http_exception(
self, e: HTTPException
) -> HTTPException | ft.ResponseReturnValue:
"""Handles an HTTP exception. By default this will invoke the
registered error handlers and fall back to returning the
exception as response.
.. versionchanged:: 1.0.3
``RoutingException``, used internally for actions such as
slash redirects during routing, is not passed to error
handlers.
.. versionchanged:: 1.0
Exceptions are looked up by code *and* by MRO, so
``HTTPException`` subclasses can be handled with a catch-all
handler for the base ``HTTPException``.
.. versionadded:: 0.3
"""
# Proxy exceptions don't have error codes. We want to always return
# those unchanged as errors
if e.code is None:
return e
# RoutingExceptions are used internally to trigger routing
# actions, such as slash redirects raising RequestRedirect. They
# are not raised or handled in user code.
if isinstance(e, RoutingException):
return e
handler = self._find_error_handler(e, request.blueprints)
if handler is None:
return e
return self.ensure_sync(handler)(e) # type: ignore[no-any-return]
def handle_user_exception(
self, e: Exception
) -> HTTPException | ft.ResponseReturnValue:
"""This method is called whenever an exception occurs that
should be handled. A special case is :class:`~werkzeug
.exceptions.HTTPException` which is forwarded to the
:meth:`handle_http_exception` method. This function will either
return a response value or reraise the exception with the same
traceback.
.. versionchanged:: 1.0
Key errors raised from request data like ``form`` show the
bad key in debug mode rather than a generic bad request
message.
.. versionadded:: 0.7
"""
if isinstance(e, BadRequestKeyError) and (
self.debug or self.config["TRAP_BAD_REQUEST_ERRORS"]
):
e.show_exception = True
if isinstance(e, HTTPException) and not self.trap_http_exception(e):
return self.handle_http_exception(e)
handler = self._find_error_handler(e, request.blueprints)
if handler is None:
raise
return self.ensure_sync(handler)(e) # type: ignore[no-any-return]
def handle_exception(self, e: Exception) -> Response:
"""Handle an exception that did not have an error handler
associated with it, or that was raised from an error handler.
This always causes a 500 ``InternalServerError``.
Always sends the :data:`got_request_exception` signal.
If :data:`PROPAGATE_EXCEPTIONS` is ``True``, such as in debug
mode, the error will be re-raised so that the debugger can
display it. Otherwise, the original exception is logged, and
an :exc:`~werkzeug.exceptions.InternalServerError` is returned.
If an error handler is registered for ``InternalServerError`` or
``500``, it will be used. For consistency, the handler will
always receive the ``InternalServerError``. The original
unhandled exception is available as ``e.original_exception``.
.. versionchanged:: 1.1.0
Always passes the ``InternalServerError`` instance to the
handler, setting ``original_exception`` to the unhandled
error.
.. versionchanged:: 1.1.0
``after_request`` functions and other finalization is done
even for the default 500 response when there is no handler.
.. versionadded:: 0.3
"""
exc_info = sys.exc_info()
got_request_exception.send(self, _async_wrapper=self.ensure_sync, exception=e)
propagate = self.config["PROPAGATE_EXCEPTIONS"]
if propagate is None:
propagate = self.testing or self.debug
if propagate:
# Re-raise if called with an active exception, otherwise
# raise the passed in exception.
if exc_info[1] is e:
raise
raise e
self.log_exception(exc_info)
server_error: InternalServerError | ft.ResponseReturnValue
server_error = InternalServerError(original_exception=e)
handler = self._find_error_handler(server_error, request.blueprints)
if handler is not None:
server_error = self.ensure_sync(handler)(server_error)
return self.finalize_request(server_error, from_error_handler=True)
def log_exception(
self,
exc_info: (tuple[type, BaseException, TracebackType] | tuple[None, None, None]),
) -> None:
"""Logs an exception. This is called by :meth:`handle_exception`
if debugging is disabled and right before the handler is called.
The default implementation logs the exception as error on the
:attr:`logger`.
.. versionadded:: 0.8
"""
self.logger.error(
f"Exception on {request.path} [{request.method}]", exc_info=exc_info
)
def dispatch_request(self) -> ft.ResponseReturnValue:
"""Does the request dispatching. Matches the URL and returns the
return value of the view or error handler. This does not have to
be a response object. In order to convert the return value to a
proper response object, call :func:`make_response`.
.. versionchanged:: 0.7
This no longer does the exception handling, this code was
moved to the new :meth:`full_dispatch_request`.
"""
req = request_ctx.request
if req.routing_exception is not None:
self.raise_routing_exception(req)
rule: Rule = req.url_rule # type: ignore[assignment]
# if we provide automatic options for this URL and the
# request came with the OPTIONS method, reply automatically
if (
getattr(rule, "provide_automatic_options", False)
and req.method == "OPTIONS"
):
return self.make_default_options_response()
# otherwise dispatch to the handler for that endpoint
view_args: dict[str, t.Any] = req.view_args # type: ignore[assignment]
return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args) # type: ignore[no-any-return]
def full_dispatch_request(self) -> Response:
"""Dispatches the request and on top of that performs request
pre and postprocessing as well as HTTP exception catching and
error handling.
.. versionadded:: 0.7
"""
self._got_first_request = True
try:
request_started.send(self, _async_wrapper=self.ensure_sync)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
def finalize_request(
self,
rv: ft.ResponseReturnValue | HTTPException,
from_error_handler: bool = False,
) -> Response:
"""Given the return value from a view function this finalizes
the request by converting it into a response and invoking the
postprocessing functions. This is invoked for both normal
request dispatching as well as error handlers.
Because this means that it might be called as a result of a
failure a special safe mode is available which can be enabled
with the `from_error_handler` flag. If enabled, failures in
response processing will be logged and otherwise ignored.
:internal:
"""
response = self.make_response(rv)
try:
response = self.process_response(response)
request_finished.send(
self, _async_wrapper=self.ensure_sync, response=response
)
except Exception:
if not from_error_handler:
raise
self.logger.exception(
"Request finalizing failed with an error while handling an error"
)
return response
def make_default_options_response(self) -> Response:
"""This method is called to create the default ``OPTIONS`` response.
This can be changed through subclassing to change the default
behavior of ``OPTIONS`` responses.
.. versionadded:: 0.7
"""
adapter = request_ctx.url_adapter
methods = adapter.allowed_methods() # type: ignore[union-attr]
rv = self.response_class()
rv.allow.update(methods)
return rv
def ensure_sync(self, func: t.Callable[..., t.Any]) -> t.Callable[..., t.Any]:
"""Ensure that the function is synchronous for WSGI workers.
Plain ``def`` functions are returned as-is. ``async def``
functions are wrapped to run and wait for the response.
Override this method to change how the app runs async views.
.. versionadded:: 2.0
"""
if iscoroutinefunction(func):
return self.async_to_sync(func)
return func
def async_to_sync(
self, func: t.Callable[..., t.Coroutine[t.Any, t.Any, t.Any]]
) -> t.Callable[..., t.Any]:
"""Return a sync function that will run the coroutine function.
.. code-block:: python
result = app.async_to_sync(func)(*args, **kwargs)
Override this method to change how the app converts async code
to be synchronously callable.
.. versionadded:: 2.0
"""
try:
from asgiref.sync import async_to_sync as asgiref_async_to_sync
except ImportError:
raise RuntimeError(
"Install Flask with the 'async' extra in order to use async views."
) from None
return asgiref_async_to_sync(func)
def url_for(
self,
/,
endpoint: str,
*,
_anchor: str | None = None,
_method: str | None = None,
_scheme: str | None = None,
_external: bool | None = None,
**values: t.Any,
) -> str:
"""Generate a URL to the given endpoint with the given values.
This is called by :func:`flask.url_for`, and can be called
directly as well.
An *endpoint* is the name of a URL rule, usually added with
:meth:`@app.route() <route>`, and usually the same name as the
view function. A route defined in a :class:`~flask.Blueprint`
will prepend the blueprint's name separated by a ``.`` to the
endpoint.
In some cases, such as email messages, you want URLs to include
the scheme and domain, like ``https://example.com/hello``. When
not in an active request, URLs will be external by default, but
this requires setting :data:`SERVER_NAME` so Flask knows what
domain to use. :data:`APPLICATION_ROOT` and
:data:`PREFERRED_URL_SCHEME` should also be configured as
needed. This config is only used when not in an active request.
Functions can be decorated with :meth:`url_defaults` to modify
keyword arguments before the URL is built.
If building fails for some reason, such as an unknown endpoint
or incorrect values, the app's :meth:`handle_url_build_error`
method is called. If that returns a string, that is returned,
otherwise a :exc:`~werkzeug.routing.BuildError` is raised.
:param endpoint: The endpoint name associated with the URL to
generate. If this starts with a ``.``, the current blueprint
name (if any) will be used.
:param _anchor: If given, append this as ``#anchor`` to the URL.
:param _method: If given, generate the URL associated with this
method for the endpoint.
:param _scheme: If given, the URL will have this scheme if it
is external.
:param _external: If given, prefer the URL to be internal
(False) or require it to be external (True). External URLs
include the scheme and domain. When not in an active
request, URLs are external by default.
:param values: Values to use for the variable parts of the URL
rule. Unknown keys are appended as query string arguments,
like ``?a=b&c=d``.
.. versionadded:: 2.2
Moved from ``flask.url_for``, which calls this method.
"""
req_ctx = _cv_request.get(None)
if req_ctx is not None:
url_adapter = req_ctx.url_adapter
blueprint_name = req_ctx.request.blueprint
# If the endpoint starts with "." and the request matches a
# blueprint, the endpoint is relative to the blueprint.
if endpoint[:1] == ".":
if blueprint_name is not None:
endpoint = f"{blueprint_name}{endpoint}"
else:
endpoint = endpoint[1:]
# When in a request, generate a URL without scheme and
# domain by default, unless a scheme is given.
if _external is None:
_external = _scheme is not None
else:
app_ctx = _cv_app.get(None)
# If called by helpers.url_for, an app context is active,
# use its url_adapter. Otherwise, app.url_for was called
# directly, build an adapter.
if app_ctx is not None:
url_adapter = app_ctx.url_adapter
else:
url_adapter = self.create_url_adapter(None)
if url_adapter is None:
raise RuntimeError(
"Unable to build URLs outside an active request"
" without 'SERVER_NAME' configured. Also configure"
" 'APPLICATION_ROOT' and 'PREFERRED_URL_SCHEME' as"
" needed."
)
# When outside a request, generate a URL with scheme and
# domain by default.
if _external is None:
_external = True
# It is an error to set _scheme when _external=False, in order
# to avoid accidental insecure URLs.
if _scheme is not None and not _external:
raise ValueError("When specifying '_scheme', '_external' must be True.")
self.inject_url_defaults(endpoint, values)
try:
rv = url_adapter.build( # type: ignore[union-attr]
endpoint,
values,
method=_method,
url_scheme=_scheme,
force_external=_external,
)
except BuildError as error:
values.update(
_anchor=_anchor, _method=_method, _scheme=_scheme, _external=_external
)
return self.handle_url_build_error(error, endpoint, values)
if _anchor is not None:
_anchor = _url_quote(_anchor, safe="%!#$&'()*+,/:;=?@")
rv = f"{rv}#{_anchor}"
return rv
def make_response(self, rv: ft.ResponseReturnValue) -> Response:
"""Convert the return value from a view function to an instance of
:attr:`response_class`.
:param rv: the return value from the view function. The view function
must return a response. Returning ``None``, or the view ending
without returning, is not allowed. The following types are allowed
for ``view_rv``:
``str``
A response object is created with the string encoded to UTF-8
as the body.
``bytes``
A response object is created with the bytes as the body.
``dict``
A dictionary that will be jsonify'd before being returned.
``list``
A list that will be jsonify'd before being returned.
``generator`` or ``iterator``
A generator that returns ``str`` or ``bytes`` to be
streamed as the response.
``tuple``
Either ``(body, status, headers)``, ``(body, status)``, or
``(body, headers)``, where ``body`` is any of the other types
allowed here, ``status`` is a string or an integer, and
``headers`` is a dictionary or a list of ``(key, value)``
tuples. If ``body`` is a :attr:`response_class` instance,
``status`` overwrites the exiting value and ``headers`` are
extended.
:attr:`response_class`
The object is returned unchanged.
other :class:`~werkzeug.wrappers.Response` class
The object is coerced to :attr:`response_class`.
:func:`callable`
The function is called as a WSGI application. The result is
used to create a response object.
.. versionchanged:: 2.2
A generator will be converted to a streaming response.
A list will be converted to a JSON response.
.. versionchanged:: 1.1
A dict will be converted to a JSON response.
.. versionchanged:: 0.9
Previously a tuple was interpreted as the arguments for the
response object.
"""
status = headers = None
# unpack tuple returns
if isinstance(rv, tuple):
len_rv = len(rv)
# a 3-tuple is unpacked directly
if len_rv == 3:
rv, status, headers = rv # type: ignore[misc]
# decide if a 2-tuple has status or headers
elif len_rv == 2:
if isinstance(rv[1], (Headers, dict, tuple, list)):
rv, headers = rv
else:
rv, status = rv # type: ignore[assignment,misc]
# other sized tuples are not allowed
else:
raise TypeError(
"The view function did not return a valid response tuple."
" The tuple must have the form (body, status, headers),"
" (body, status), or (body, headers)."
)
# the body must not be None
if rv is None:
raise TypeError(
f"The view function for {request.endpoint!r} did not"
" return a valid response. The function either returned"
" None or ended without a return statement."
)
# make sure the body is an instance of the response class
if not isinstance(rv, self.response_class):
if isinstance(rv, (str, bytes, bytearray)) or isinstance(rv, cabc.Iterator):
# let the response class set the status and headers instead of
# waiting to do it manually, so that the class can handle any
# special logic
rv = self.response_class(
rv,
status=status,
headers=headers, # type: ignore[arg-type]
)
status = headers = None
elif isinstance(rv, (dict, list)):
rv = self.json.response(rv)
elif isinstance(rv, BaseResponse) or callable(rv):
# evaluate a WSGI callable, or coerce a different response
# class to the correct type
try:
rv = self.response_class.force_type(
rv, # type: ignore[arg-type]
request.environ,
)
except TypeError as e:
raise TypeError(
f"{e}\nThe view function did not return a valid"
" response. The return type must be a string,"
" dict, list, tuple with headers or status,"
" Response instance, or WSGI callable, but it"
f" was a {type(rv).__name__}."
).with_traceback(sys.exc_info()[2]) from None
else:
raise TypeError(
"The view function did not return a valid"
" response. The return type must be a string,"
" dict, list, tuple with headers or status,"
" Response instance, or WSGI callable, but it was a"
f" {type(rv).__name__}."
)
rv = t.cast(Response, rv)
# prefer the status if it was provided
if status is not None:
if isinstance(status, (str, bytes, bytearray)):
rv.status = status
else:
rv.status_code = status
# extend existing headers with provided headers
if headers:
rv.headers.update(headers) # type: ignore[arg-type]
return rv
def preprocess_request(self) -> ft.ResponseReturnValue | None:
"""Called before the request is dispatched. Calls
:attr:`url_value_preprocessors` registered with the app and the
current blueprint (if any). Then calls :attr:`before_request_funcs`
registered with the app and the blueprint.
If any :meth:`before_request` handler 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.
"""
names = (None, *reversed(request.blueprints))
for name in names:
if name in self.url_value_preprocessors:
for url_func in self.url_value_preprocessors[name]:
url_func(request.endpoint, request.view_args)
for name in names:
if name in self.before_request_funcs:
for before_func in self.before_request_funcs[name]:
rv = self.ensure_sync(before_func)()
if rv is not None:
return rv # type: ignore[no-any-return]
return None
def process_response(self, response: Response) -> Response:
"""Can be overridden in order to modify the response object
before it's sent to the WSGI server. By default this will
call all the :meth:`after_request` decorated functions.
.. versionchanged:: 0.5
As of Flask 0.5 the functions registered for after request
execution are called in reverse order of registration.
:param response: a :attr:`response_class` object.
:return: a new response object or the same, has to be an
instance of :attr:`response_class`.
"""
ctx = request_ctx._get_current_object() # type: ignore[attr-defined]
for func in ctx._after_request_functions:
response = self.ensure_sync(func)(response)
for name in chain(request.blueprints, (None,)):
if name in self.after_request_funcs:
for func in reversed(self.after_request_funcs[name]):
response = self.ensure_sync(func)(response)
if not self.session_interface.is_null_session(ctx.session):
self.session_interface.save_session(self, ctx.session, response)
return response
def do_teardown_request(
self,
exc: BaseException | None = _sentinel, # type: ignore[assignment]
) -> None:
"""Called after the request is dispatched and the response is
returned, right before the request context is popped.
This calls all functions decorated with
:meth:`teardown_request`, and :meth:`Blueprint.teardown_request`
if a blueprint handled the request. Finally, the
:data:`request_tearing_down` signal is sent.
This is called by
:meth:`RequestContext.pop() <flask.ctx.RequestContext.pop>`,
which may be delayed during testing to maintain access to
resources.
:param exc: An unhandled exception raised while dispatching the
request. Detected from the current exception information if
not passed. Passed to each teardown function.
.. versionchanged:: 0.9
Added the ``exc`` argument.
"""
if exc is _sentinel:
exc = sys.exc_info()[1]
for name in chain(request.blueprints, (None,)):
if name in self.teardown_request_funcs:
for func in reversed(self.teardown_request_funcs[name]):
self.ensure_sync(func)(exc)
request_tearing_down.send(self, _async_wrapper=self.ensure_sync, exc=exc)
def do_teardown_appcontext(
self,
exc: BaseException | None = _sentinel, # type: ignore[assignment]
) -> None:
"""Called right before the application context is popped.
When handling a request, the application context is popped
after the request context. See :meth:`do_teardown_request`.
This calls all functions decorated with
:meth:`teardown_appcontext`. Then the
:data:`appcontext_tearing_down` signal is sent.
This is called by
:meth:`AppContext.pop() <flask.ctx.AppContext.pop>`.
.. versionadded:: 0.9
"""
if exc is _sentinel:
exc = sys.exc_info()[1]
for func in reversed(self.teardown_appcontext_funcs):
self.ensure_sync(func)(exc)
appcontext_tearing_down.send(self, _async_wrapper=self.ensure_sync, exc=exc)
def app_context(self) -> AppContext:
"""Create an :class:`~flask.ctx.AppContext`. Use as a ``with``
block to push the context, which will make :data:`current_app`
point at this application.
An application context is automatically pushed by
:meth:`RequestContext.push() <flask.ctx.RequestContext.push>`
when handling a request, and when running a CLI command. Use
this to manually create a context outside of these situations.
::
with app.app_context():
init_db()
See :doc:`/appcontext`.
.. versionadded:: 0.9
"""
return AppContext(self)
def request_context(self, environ: WSGIEnvironment) -> RequestContext:
"""Create a :class:`~flask.ctx.RequestContext` representing a
WSGI environment. Use a ``with`` block to push the context,
which will make :data:`request` point at this request.
See :doc:`/reqcontext`.
Typically you should not call this from your own code. A request
context is automatically pushed by the :meth:`wsgi_app` when
handling a request. Use :meth:`test_request_context` to create
an environment and context instead of this method.
:param environ: a WSGI environment
"""
return RequestContext(self, environ)
def test_request_context(self, *args: t.Any, **kwargs: t.Any) -> RequestContext:
"""Create a :class:`~flask.ctx.RequestContext` for a WSGI
environment created from the given values. This is mostly useful
during testing, where you may want to run a function that uses
request data without dispatching a full request.
See :doc:`/reqcontext`.
Use a ``with`` block to push the context, which will make
:data:`request` point at the request for the created
environment. ::
with app.test_request_context(...):
generate_report()
When using the shell, it may be easier to push and pop the
context manually to avoid indentation. ::
ctx = app.test_request_context(...)
ctx.push()
...
ctx.pop()
Takes the same arguments as Werkzeug's
:class:`~werkzeug.test.EnvironBuilder`, with some defaults from
the application. See the linked Werkzeug docs for most of the
available arguments. Flask-specific behavior is listed here.
: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 data: The request body, either as a string or a dict of
form keys and values.
: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`.
"""
from .testing import EnvironBuilder
builder = EnvironBuilder(self, *args, **kwargs)
try:
return self.request_context(builder.get_environ())
finally:
builder.close()
def wsgi_app(
self, environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable[bytes]:
"""The actual WSGI application. This is not implemented in
:meth:`__call__` so that middlewares can be applied without
losing a reference to the app object. Instead of doing this::
app = MyMiddleware(app)
It's a better idea to do this instead::
app.wsgi_app = MyMiddleware(app.wsgi_app)
Then you still have the original application object around and
can continue to call methods on it.
.. versionchanged:: 0.7
Teardown events for the request and app contexts are called
even if an unhandled error occurs. Other events may not be
called depending on when an error occurs during dispatch.
See :ref:`callbacks-and-errors`.
:param environ: A WSGI environment.
:param start_response: A callable accepting a status code,
a list of headers, and an optional exception context to
start the response.
"""
ctx = self.request_context(environ)
error: BaseException | None = None
try:
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if "werkzeug.debug.preserve_context" in environ:
environ["werkzeug.debug.preserve_context"](_cv_app.get())
environ["werkzeug.debug.preserve_context"](_cv_request.get())
if error is not None and self.should_ignore_error(error):
error = None
ctx.pop(error)
def __call__(
self, environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable[bytes]:
"""The WSGI server calls the Flask application object as the
WSGI application. This calls :meth:`wsgi_app`, which can be
wrapped to apply middleware.
"""
return self.wsgi_app(environ, start_response)
server-temp/venv/lib/python3.10/site-packages/flask/blueprints.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import os
import typing as t
from datetime import timedelta
from .globals import current_app
from .helpers import send_from_directory
from .sansio.blueprints import Blueprint as SansioBlueprint
from .sansio.blueprints import BlueprintSetupState as BlueprintSetupState # noqa
if t.TYPE_CHECKING: # pragma: no cover
from .wrappers import Response
class Blueprint(SansioBlueprint):
def get_send_file_max_age(self, filename: str | None) -> int | None:
"""Used by :func:`send_file` to determine the ``max_age`` cache
value for a given file path if it wasn't passed.
By default, this returns :data:`SEND_FILE_MAX_AGE_DEFAULT` from
the configuration of :data:`~flask.current_app`. This defaults
to ``None``, which tells the browser to use conditional requests
instead of a timed cache, which is usually preferable.
Note this is a duplicate of the same method in the Flask
class.
.. versionchanged:: 2.0
The default configuration is ``None`` instead of 12 hours.
.. versionadded:: 0.9
"""
value = current_app.config["SEND_FILE_MAX_AGE_DEFAULT"]
if value is None:
return None
if isinstance(value, timedelta):
return int(value.total_seconds())
return value # type: ignore[no-any-return]
def send_static_file(self, filename: str) -> Response:
"""The view function used to serve files from
:attr:`static_folder`. A route is automatically registered for
this view at :attr:`static_url_path` if :attr:`static_folder` is
set.
Note this is a duplicate of the same method in the Flask
class.
.. versionadded:: 0.5
"""
if not self.has_static_folder:
raise RuntimeError("'static_folder' must be set to serve static_files.")
# send_file only knows to call get_send_file_max_age on the app,
# call it here so it works for blueprints too.
max_age = self.get_send_file_max_age(filename)
return send_from_directory(
t.cast(str, self.static_folder), filename, max_age=max_age
)
def open_resource(self, resource: str, mode: str = "rb") -> t.IO[t.AnyStr]:
"""Open a resource file relative to :attr:`root_path` for
reading.
For example, if the file ``schema.sql`` is next to the file
``app.py`` where the ``Flask`` app is defined, it can be opened
with:
.. code-block:: python
with app.open_resource("schema.sql") as f:
conn.executescript(f.read())
:param resource: Path to the resource relative to
:attr:`root_path`.
:param mode: Open the file in this mode. Only reading is
supported, valid values are "r" (or "rt") and "rb".
Note this is a duplicate of the same method in the Flask
class.
"""
if mode not in {"r", "rt", "rb"}:
raise ValueError("Resources can only be opened for reading.")
return open(os.path.join(self.root_path, resource), mode)
server-temp/venv/lib/python3.10/site-packages/flask/cli.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import ast
import collections.abc as cabc
import importlib.metadata
import inspect
import os
import platform
import re
import sys
import traceback
import typing as t
from functools import update_wrapper
from operator import itemgetter
from types import ModuleType
import click
from click.core import ParameterSource
from werkzeug import run_simple
from werkzeug.serving import is_running_from_reloader
from werkzeug.utils import import_string
from .globals import current_app
from .helpers import get_debug_flag
from .helpers import get_load_dotenv
if t.TYPE_CHECKING:
import ssl
from _typeshed.wsgi import StartResponse
from _typeshed.wsgi import WSGIApplication
from _typeshed.wsgi import WSGIEnvironment
from .app import Flask
class NoAppException(click.UsageError):
"""Raised if an application cannot be found or loaded."""
def find_best_app(module: ModuleType) -> Flask:
"""Given a module instance this tries to find the best possible
application in the module or raises an exception.
"""
from . import Flask
# Search for the most common names first.
for attr_name in ("app", "application"):
app = getattr(module, attr_name, None)
if isinstance(app, Flask):
return app
# Otherwise find the only object that is a Flask instance.
matches = [v for v in module.__dict__.values() if isinstance(v, Flask)]
if len(matches) == 1:
return matches[0]
elif len(matches) > 1:
raise NoAppException(
"Detected multiple Flask applications in module"
f" '{module.__name__}'. Use '{module.__name__}:name'"
" to specify the correct one."
)
# Search for app factory functions.
for attr_name in ("create_app", "make_app"):
app_factory = getattr(module, attr_name, None)
if inspect.isfunction(app_factory):
try:
app = app_factory()
if isinstance(app, Flask):
return app
except TypeError as e:
if not _called_with_wrong_args(app_factory):
raise
raise NoAppException(
f"Detected factory '{attr_name}' in module '{module.__name__}',"
" but could not call it without arguments. Use"
f" '{module.__name__}:{attr_name}(args)'"
" to specify arguments."
) from e
raise NoAppException(
"Failed to find Flask application or factory in module"
f" '{module.__name__}'. Use '{module.__name__}:name'"
" to specify one."
)
def _called_with_wrong_args(f: t.Callable[..., Flask]) -> bool:
"""Check whether calling a function raised a ``TypeError`` because
the call failed or because something in the factory raised the
error.
:param f: The function that was called.
:return: ``True`` if the call failed.
"""
tb = sys.exc_info()[2]
try:
while tb is not None:
if tb.tb_frame.f_code is f.__code__:
# In the function, it was called successfully.
return False
tb = tb.tb_next
# Didn't reach the function.
return True
finally:
# Delete tb to break a circular reference.
# https://docs.python.org/2/library/sys.html#sys.exc_info
del tb
def find_app_by_string(module: ModuleType, app_name: str) -> Flask:
"""Check if the given string is a variable name or a function. Call
a function to get the app instance, or return the variable directly.
"""
from . import Flask
# Parse app_name as a single expression to determine if it's a valid
# attribute name or function call.
try:
expr = ast.parse(app_name.strip(), mode="eval").body
except SyntaxError:
raise NoAppException(
f"Failed to parse {app_name!r} as an attribute name or function call."
) from None
if isinstance(expr, ast.Name):
name = expr.id
args = []
kwargs = {}
elif isinstance(expr, ast.Call):
# Ensure the function name is an attribute name only.
if not isinstance(expr.func, ast.Name):
raise NoAppException(
f"Function reference must be a simple name: {app_name!r}."
)
name = expr.func.id
# Parse the positional and keyword arguments as literals.
try:
args = [ast.literal_eval(arg) for arg in expr.args]
kwargs = {
kw.arg: ast.literal_eval(kw.value)
for kw in expr.keywords
if kw.arg is not None
}
except ValueError:
# literal_eval gives cryptic error messages, show a generic
# message with the full expression instead.
raise NoAppException(
f"Failed to parse arguments as literal values: {app_name!r}."
) from None
else:
raise NoAppException(
f"Failed to parse {app_name!r} as an attribute name or function call."
)
try:
attr = getattr(module, name)
except AttributeError as e:
raise NoAppException(
f"Failed to find attribute {name!r} in {module.__name__!r}."
) from e
# If the attribute is a function, call it with any args and kwargs
# to get the real application.
if inspect.isfunction(attr):
try:
app = attr(*args, **kwargs)
except TypeError as e:
if not _called_with_wrong_args(attr):
raise
raise NoAppException(
f"The factory {app_name!r} in module"
f" {module.__name__!r} could not be called with the"
" specified arguments."
) from e
else:
app = attr
if isinstance(app, Flask):
return app
raise NoAppException(
"A valid Flask application was not obtained from"
f" '{module.__name__}:{app_name}'."
)
def prepare_import(path: str) -> str:
"""Given a filename this will try to calculate the python path, add it
to the search path and return the actual module name that is expected.
"""
path = os.path.realpath(path)
fname, ext = os.path.splitext(path)
if ext == ".py":
path = fname
if os.path.basename(path) == "__init__":
path = os.path.dirname(path)
module_name = []
# move up until outside package structure (no __init__.py)
while True:
path, name = os.path.split(path)
module_name.append(name)
if not os.path.exists(os.path.join(path, "__init__.py")):
break
if sys.path[0] != path:
sys.path.insert(0, path)
return ".".join(module_name[::-1])
@t.overload
def locate_app(
module_name: str, app_name: str | None, raise_if_not_found: t.Literal[True] = True
) -> Flask:
...
@t.overload
def locate_app(
module_name: str, app_name: str | None, raise_if_not_found: t.Literal[False] = ...
) -> Flask | None:
...
def locate_app(
module_name: str, app_name: str | None, raise_if_not_found: bool = True
) -> Flask | None:
try:
__import__(module_name)
except ImportError:
# Reraise the ImportError if it occurred within the imported module.
# Determine this by checking whether the trace has a depth > 1.
if sys.exc_info()[2].tb_next: # type: ignore[union-attr]
raise NoAppException(
f"While importing {module_name!r}, an ImportError was"
f" raised:\n\n{traceback.format_exc()}"
) from None
elif raise_if_not_found:
raise NoAppException(f"Could not import {module_name!r}.") from None
else:
return None
module = sys.modules[module_name]
if app_name is None:
return find_best_app(module)
else:
return find_app_by_string(module, app_name)
def get_version(ctx: click.Context, param: click.Parameter, value: t.Any) -> None:
if not value or ctx.resilient_parsing:
return
flask_version = importlib.metadata.version("flask")
werkzeug_version = importlib.metadata.version("werkzeug")
click.echo(
f"Python {platform.python_version()}\n"
f"Flask {flask_version}\n"
f"Werkzeug {werkzeug_version}",
color=ctx.color,
)
ctx.exit()
version_option = click.Option(
["--version"],
help="Show the Flask version.",
expose_value=False,
callback=get_version,
is_flag=True,
is_eager=True,
)
class ScriptInfo:
"""Helper object to deal with Flask applications. This is usually not
necessary to interface with as it's used internally in the dispatching
to click. In future versions of Flask this object will most likely play
a bigger role. Typically it's created automatically by the
:class:`FlaskGroup` but you can also manually create it and pass it
onwards as click object.
"""
def __init__(
self,
app_import_path: str | None = None,
create_app: t.Callable[..., Flask] | None = None,
set_debug_flag: bool = True,
) -> None:
#: Optionally the import path for the Flask application.
self.app_import_path = app_import_path
#: Optionally a function that is passed the script info to create
#: the instance of the application.
self.create_app = create_app
#: A dictionary with arbitrary data that can be associated with
#: this script info.
self.data: dict[t.Any, t.Any] = {}
self.set_debug_flag = set_debug_flag
self._loaded_app: Flask | None = None
def load_app(self) -> Flask:
"""Loads the Flask app (if not yet loaded) and returns it. Calling
this multiple times will just result in the already loaded app to
be returned.
"""
if self._loaded_app is not None:
return self._loaded_app
if self.create_app is not None:
app: Flask | None = self.create_app()
else:
if self.app_import_path:
path, name = (
re.split(r":(?![\\/])", self.app_import_path, maxsplit=1) + [None]
)[:2]
import_name = prepare_import(path)
app = locate_app(import_name, name)
else:
for path in ("wsgi.py", "app.py"):
import_name = prepare_import(path)
app = locate_app(import_name, None, raise_if_not_found=False)
if app is not None:
break
if app is None:
raise NoAppException(
"Could not locate a Flask application. Use the"
" 'flask --app' option, 'FLASK_APP' environment"
" variable, or a 'wsgi.py' or 'app.py' file in the"
" current directory."
)
if self.set_debug_flag:
# Update the app's debug flag through the descriptor so that
# other values repopulate as well.
app.debug = get_debug_flag()
self._loaded_app = app
return app
pass_script_info = click.make_pass_decorator(ScriptInfo, ensure=True)
F = t.TypeVar("F", bound=t.Callable[..., t.Any])
def with_appcontext(f: F) -> F:
"""Wraps a callback so that it's guaranteed to be executed with the
script's application context.
Custom commands (and their options) registered under ``app.cli`` or
``blueprint.cli`` will always have an app context available, this
decorator is not required in that case.
.. versionchanged:: 2.2
The app context is active for subcommands as well as the
decorated callback. The app context is always available to
``app.cli`` command and parameter callbacks.
"""
@click.pass_context
def decorator(ctx: click.Context, /, *args: t.Any, **kwargs: t.Any) -> t.Any:
if not current_app:
app = ctx.ensure_object(ScriptInfo).load_app()
ctx.with_resource(app.app_context())
return ctx.invoke(f, *args, **kwargs)
return update_wrapper(decorator, f) # type: ignore[return-value]
class AppGroup(click.Group):
"""This works similar to a regular click :class:`~click.Group` but it
changes the behavior of the :meth:`command` decorator so that it
automatically wraps the functions in :func:`with_appcontext`.
Not to be confused with :class:`FlaskGroup`.
"""
def command( # type: ignore[override]
self, *args: t.Any, **kwargs: t.Any
) -> t.Callable[[t.Callable[..., t.Any]], click.Command]:
"""This works exactly like the method of the same name on a regular
:class:`click.Group` but it wraps callbacks in :func:`with_appcontext`
unless it's disabled by passing ``with_appcontext=False``.
"""
wrap_for_ctx = kwargs.pop("with_appcontext", True)
def decorator(f: t.Callable[..., t.Any]) -> click.Command:
if wrap_for_ctx:
f = with_appcontext(f)
return super(AppGroup, self).command(*args, **kwargs)(f) # type: ignore[no-any-return]
return decorator
def group( # type: ignore[override]
self, *args: t.Any, **kwargs: t.Any
) -> t.Callable[[t.Callable[..., t.Any]], click.Group]:
"""This works exactly like the method of the same name on a regular
:class:`click.Group` but it defaults the group class to
:class:`AppGroup`.
"""
kwargs.setdefault("cls", AppGroup)
return super().group(*args, **kwargs) # type: ignore[no-any-return]
def _set_app(ctx: click.Context, param: click.Option, value: str | None) -> str | None:
if value is None:
return None
info = ctx.ensure_object(ScriptInfo)
info.app_import_path = value
return value
# This option is eager so the app will be available if --help is given.
# --help is also eager, so --app must be before it in the param list.
# no_args_is_help bypasses eager processing, so this option must be
# processed manually in that case to ensure FLASK_APP gets picked up.
_app_option = click.Option(
["-A", "--app"],
metavar="IMPORT",
help=(
"The Flask application or factory function to load, in the form 'module:name'."
" Module can be a dotted import or file path. Name is not required if it is"
" 'app', 'application', 'create_app', or 'make_app', and can be 'name(args)' to"
" pass arguments."
),
is_eager=True,
expose_value=False,
callback=_set_app,
)
def _set_debug(ctx: click.Context, param: click.Option, value: bool) -> bool | None:
# If the flag isn't provided, it will default to False. Don't use
# that, let debug be set by env in that case.
source = ctx.get_parameter_source(param.name) # type: ignore[arg-type]
if source is not None and source in (
ParameterSource.DEFAULT,
ParameterSource.DEFAULT_MAP,
):
return None
# Set with env var instead of ScriptInfo.load so that it can be
# accessed early during a factory function.
os.environ["FLASK_DEBUG"] = "1" if value else "0"
return value
_debug_option = click.Option(
["--debug/--no-debug"],
help="Set debug mode.",
expose_value=False,
callback=_set_debug,
)
def _env_file_callback(
ctx: click.Context, param: click.Option, value: str | None
) -> str | None:
if value is None:
return None
import importlib
try:
importlib.import_module("dotenv")
except ImportError:
raise click.BadParameter(
"python-dotenv must be installed to load an env file.",
ctx=ctx,
param=param,
) from None
# Don't check FLASK_SKIP_DOTENV, that only disables automatically
# loading .env and .flaskenv files.
load_dotenv(value)
return value
# This option is eager so env vars are loaded as early as possible to be
# used by other options.
_env_file_option = click.Option(
["-e", "--env-file"],
type=click.Path(exists=True, dir_okay=False),
help="Load environment variables from this file. python-dotenv must be installed.",
is_eager=True,
expose_value=False,
callback=_env_file_callback,
)
class FlaskGroup(AppGroup):
"""Special subclass of the :class:`AppGroup` group that supports
loading more commands from the configured Flask app. Normally a
developer does not have to interface with this class but there are
some very advanced use cases for which it makes sense to create an
instance of this. see :ref:`custom-scripts`.
:param add_default_commands: if this is True then the default run and
shell commands will be added.
:param add_version_option: adds the ``--version`` option.
:param create_app: an optional callback that is passed the script info and
returns the loaded app.
:param load_dotenv: Load the nearest :file:`.env` and :file:`.flaskenv`
files to set environment variables. Will also change the working
directory to the directory containing the first file found.
:param set_debug_flag: Set the app's debug flag.
.. versionchanged:: 2.2
Added the ``-A/--app``, ``--debug/--no-debug``, ``-e/--env-file`` options.
.. versionchanged:: 2.2
An app context is pushed when running ``app.cli`` commands, so
``@with_appcontext`` is no longer required for those commands.
.. versionchanged:: 1.0
If installed, python-dotenv will be used to load environment variables
from :file:`.env` and :file:`.flaskenv` files.
"""
def __init__(
self,
add_default_commands: bool = True,
create_app: t.Callable[..., Flask] | None = None,
add_version_option: bool = True,
load_dotenv: bool = True,
set_debug_flag: bool = True,
**extra: t.Any,
) -> None:
params = list(extra.pop("params", None) or ())
# Processing is done with option callbacks instead of a group
# callback. This allows users to make a custom group callback
# without losing the behavior. --env-file must come first so
# that it is eagerly evaluated before --app.
params.extend((_env_file_option, _app_option, _debug_option))
if add_version_option:
params.append(version_option)
if "context_settings" not in extra:
extra["context_settings"] = {}
extra["context_settings"].setdefault("auto_envvar_prefix", "FLASK")
super().__init__(params=params, **extra)
self.create_app = create_app
self.load_dotenv = load_dotenv
self.set_debug_flag = set_debug_flag
if add_default_commands:
self.add_command(run_command)
self.add_command(shell_command)
self.add_command(routes_command)
self._loaded_plugin_commands = False
def _load_plugin_commands(self) -> None:
if self._loaded_plugin_commands:
return
if sys.version_info >= (3, 10):
from importlib import metadata
else:
# Use a backport on Python < 3.10. We technically have
# importlib.metadata on 3.8+, but the API changed in 3.10,
# so use the backport for consistency.
import importlib_metadata as metadata
for ep in metadata.entry_points(group="flask.commands"):
self.add_command(ep.load(), ep.name)
self._loaded_plugin_commands = True
def get_command(self, ctx: click.Context, name: str) -> click.Command | None:
self._load_plugin_commands()
# Look up built-in and plugin commands, which should be
# available even if the app fails to load.
rv = super().get_command(ctx, name)
if rv is not None:
return rv
info = ctx.ensure_object(ScriptInfo)
# Look up commands provided by the app, showing an error and
# continuing if the app couldn't be loaded.
try:
app = info.load_app()
except NoAppException as e:
click.secho(f"Error: {e.format_message()}\n", err=True, fg="red")
return None
# Push an app context for the loaded app unless it is already
# active somehow. This makes the context available to parameter
# and command callbacks without needing @with_appcontext.
if not current_app or current_app._get_current_object() is not app: # type: ignore[attr-defined]
ctx.with_resource(app.app_context())
return app.cli.get_command(ctx, name)
def list_commands(self, ctx: click.Context) -> list[str]:
self._load_plugin_commands()
# Start with the built-in and plugin commands.
rv = set(super().list_commands(ctx))
info = ctx.ensure_object(ScriptInfo)
# Add commands provided by the app, showing an error and
# continuing if the app couldn't be loaded.
try:
rv.update(info.load_app().cli.list_commands(ctx))
except NoAppException as e:
# When an app couldn't be loaded, show the error message
# without the traceback.
click.secho(f"Error: {e.format_message()}\n", err=True, fg="red")
except Exception:
# When any other errors occurred during loading, show the
# full traceback.
click.secho(f"{traceback.format_exc()}\n", err=True, fg="red")
return sorted(rv)
def make_context(
self,
info_name: str | None,
args: list[str],
parent: click.Context | None = None,
**extra: t.Any,
) -> click.Context:
# Set a flag to tell app.run to become a no-op. If app.run was
# not in a __name__ == __main__ guard, it would start the server
# when importing, blocking whatever command is being called.
os.environ["FLASK_RUN_FROM_CLI"] = "true"
# Attempt to load .env and .flask env files. The --env-file
# option can cause another file to be loaded.
if get_load_dotenv(self.load_dotenv):
load_dotenv()
if "obj" not in extra and "obj" not in self.context_settings:
extra["obj"] = ScriptInfo(
create_app=self.create_app, set_debug_flag=self.set_debug_flag
)
return super().make_context(info_name, args, parent=parent, **extra)
def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
if not args and self.no_args_is_help:
# Attempt to load --env-file and --app early in case they
# were given as env vars. Otherwise no_args_is_help will not
# see commands from app.cli.
_env_file_option.handle_parse_result(ctx, {}, [])
_app_option.handle_parse_result(ctx, {}, [])
return super().parse_args(ctx, args)
def _path_is_ancestor(path: str, other: str) -> bool:
"""Take ``other`` and remove the length of ``path`` from it. Then join it
to ``path``. If it is the original value, ``path`` is an ancestor of
``other``."""
return os.path.join(path, other[len(path) :].lstrip(os.sep)) == other
def load_dotenv(path: str | os.PathLike[str] | None = None) -> bool:
"""Load "dotenv" files in order of precedence to set environment variables.
If an env var is already set it is not overwritten, so earlier files in the
list are preferred over later files.
This is a no-op if `python-dotenv`_ is not installed.
.. _python-dotenv: https://github.com/theskumar/python-dotenv#readme
:param path: Load the file at this location instead of searching.
:return: ``True`` if a file was loaded.
.. versionchanged:: 2.0
The current directory is not changed to the location of the
loaded file.
.. versionchanged:: 2.0
When loading the env files, set the default encoding to UTF-8.
.. versionchanged:: 1.1.0
Returns ``False`` when python-dotenv is not installed, or when
the given path isn't a file.
.. versionadded:: 1.0
"""
try:
import dotenv
except ImportError:
if path or os.path.isfile(".env") or os.path.isfile(".flaskenv"):
click.secho(
" * Tip: There are .env or .flaskenv files present."
' Do "pip install python-dotenv" to use them.',
fg="yellow",
err=True,
)
return False
# Always return after attempting to load a given path, don't load
# the default files.
if path is not None:
if os.path.isfile(path):
return dotenv.load_dotenv(path, encoding="utf-8")
return False
loaded = False
for name in (".env", ".flaskenv"):
path = dotenv.find_dotenv(name, usecwd=True)
if not path:
continue
dotenv.load_dotenv(path, encoding="utf-8")
loaded = True
return loaded # True if at least one file was located and loaded.
def show_server_banner(debug: bool, app_import_path: str | None) -> None:
"""Show extra startup messages the first time the server is run,
ignoring the reloader.
"""
if is_running_from_reloader():
return
if app_import_path is not None:
click.echo(f" * Serving Flask app '{app_import_path}'")
if debug is not None:
click.echo(f" * Debug mode: {'on' if debug else 'off'}")
class CertParamType(click.ParamType):
"""Click option type for the ``--cert`` option. Allows either an
existing file, the string ``'adhoc'``, or an import for a
:class:`~ssl.SSLContext` object.
"""
name = "path"
def __init__(self) -> None:
self.path_type = click.Path(exists=True, dir_okay=False, resolve_path=True)
def convert(
self, value: t.Any, param: click.Parameter | None, ctx: click.Context | None
) -> t.Any:
try:
import ssl
except ImportError:
raise click.BadParameter(
'Using "--cert" requires Python to be compiled with SSL support.',
ctx,
param,
) from None
try:
return self.path_type(value, param, ctx)
except click.BadParameter:
value = click.STRING(value, param, ctx).lower()
if value == "adhoc":
try:
import cryptography # noqa: F401
except ImportError:
raise click.BadParameter(
"Using ad-hoc certificates requires the cryptography library.",
ctx,
param,
) from None
return value
obj = import_string(value, silent=True)
if isinstance(obj, ssl.SSLContext):
return obj
raise
def _validate_key(ctx: click.Context, param: click.Parameter, value: t.Any) -> t.Any:
"""The ``--key`` option must be specified when ``--cert`` is a file.
Modifies the ``cert`` param to be a ``(cert, key)`` pair if needed.
"""
cert = ctx.params.get("cert")
is_adhoc = cert == "adhoc"
try:
import ssl
except ImportError:
is_context = False
else:
is_context = isinstance(cert, ssl.SSLContext)
if value is not None:
if is_adhoc:
raise click.BadParameter(
'When "--cert" is "adhoc", "--key" is not used.', ctx, param
)
if is_context:
raise click.BadParameter(
'When "--cert" is an SSLContext object, "--key" is not used.',
ctx,
param,
)
if not cert:
raise click.BadParameter('"--cert" must also be specified.', ctx, param)
ctx.params["cert"] = cert, value
else:
if cert and not (is_adhoc or is_context):
raise click.BadParameter('Required when using "--cert".', ctx, param)
return value
class SeparatedPathType(click.Path):
"""Click option type that accepts a list of values separated by the
OS's path separator (``:``, ``;`` on Windows). Each value is
validated as a :class:`click.Path` type.
"""
def convert(
self, value: t.Any, param: click.Parameter | None, ctx: click.Context | None
) -> t.Any:
items = self.split_envvar_value(value)
# can't call no-arg super() inside list comprehension until Python 3.12
super_convert = super().convert
return [super_convert(item, param, ctx) for item in items]
@click.command("run", short_help="Run a development server.")
@click.option("--host", "-h", default="127.0.0.1", help="The interface to bind to.")
@click.option("--port", "-p", default=5000, help="The port to bind to.")
@click.option(
"--cert",
type=CertParamType(),
help="Specify a certificate file to use HTTPS.",
is_eager=True,
)
@click.option(
"--key",
type=click.Path(exists=True, dir_okay=False, resolve_path=True),
callback=_validate_key,
expose_value=False,
help="The key file to use when specifying a certificate.",
)
@click.option(
"--reload/--no-reload",
default=None,
help="Enable or disable the reloader. By default the reloader "
"is active if debug is enabled.",
)
@click.option(
"--debugger/--no-debugger",
default=None,
help="Enable or disable the debugger. By default the debugger "
"is active if debug is enabled.",
)
@click.option(
"--with-threads/--without-threads",
default=True,
help="Enable or disable multithreading.",
)
@click.option(
"--extra-files",
default=None,
type=SeparatedPathType(),
help=(
"Extra files that trigger a reload on change. Multiple paths"
f" are separated by {os.path.pathsep!r}."
),
)
@click.option(
"--exclude-patterns",
default=None,
type=SeparatedPathType(),
help=(
"Files matching these fnmatch patterns will not trigger a reload"
" on change. Multiple patterns are separated by"
f" {os.path.pathsep!r}."
),
)
@pass_script_info
def run_command(
info: ScriptInfo,
host: str,
port: int,
reload: bool,
debugger: bool,
with_threads: bool,
cert: ssl.SSLContext | tuple[str, str | None] | t.Literal["adhoc"] | None,
extra_files: list[str] | None,
exclude_patterns: list[str] | None,
) -> None:
"""Run a local development server.
This server is for development purposes only. It does not provide
the stability, security, or performance of production WSGI servers.
The reloader and debugger are enabled by default with the '--debug'
option.
"""
try:
app: WSGIApplication = info.load_app()
except Exception as e:
if is_running_from_reloader():
# When reloading, print out the error immediately, but raise
# it later so the debugger or server can handle it.
traceback.print_exc()
err = e
def app(
environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable[bytes]:
raise err from None
else:
# When not reloading, raise the error immediately so the
# command fails.
raise e from None
debug = get_debug_flag()
if reload is None:
reload = debug
if debugger is None:
debugger = debug
show_server_banner(debug, info.app_import_path)
run_simple(
host,
port,
app,
use_reloader=reload,
use_debugger=debugger,
threaded=with_threads,
ssl_context=cert,
extra_files=extra_files,
exclude_patterns=exclude_patterns,
)
run_command.params.insert(0, _debug_option)
@click.command("shell", short_help="Run a shell in the app context.")
@with_appcontext
def shell_command() -> None:
"""Run an interactive Python shell in the context of a given
Flask application. The application will populate the default
namespace of this shell according to its configuration.
This is useful for executing small snippets of management code
without having to manually configure the application.
"""
import code
banner = (
f"Python {sys.version} on {sys.platform}\n"
f"App: {current_app.import_name}\n"
f"Instance: {current_app.instance_path}"
)
ctx: dict[str, t.Any] = {}
# Support the regular Python interpreter startup script if someone
# is using it.
startup = os.environ.get("PYTHONSTARTUP")
if startup and os.path.isfile(startup):
with open(startup) as f:
eval(compile(f.read(), startup, "exec"), ctx)
ctx.update(current_app.make_shell_context())
# Site, customize, or startup script can set a hook to call when
# entering interactive mode. The default one sets up readline with
# tab and history completion.
interactive_hook = getattr(sys, "__interactivehook__", None)
if interactive_hook is not None:
try:
import readline
from rlcompleter import Completer
except ImportError:
pass
else:
# rlcompleter uses __main__.__dict__ by default, which is
# flask.__main__. Use the shell context instead.
readline.set_completer(Completer(ctx).complete)
interactive_hook()
code.interact(banner=banner, local=ctx)
@click.command("routes", short_help="Show the routes for the app.")
@click.option(
"--sort",
"-s",
type=click.Choice(("endpoint", "methods", "domain", "rule", "match")),
default="endpoint",
help=(
"Method to sort routes by. 'match' is the order that Flask will match routes"
" when dispatching a request."
),
)
@click.option("--all-methods", is_flag=True, help="Show HEAD and OPTIONS methods.")
@with_appcontext
def routes_command(sort: str, all_methods: bool) -> None:
"""Show all registered routes with endpoints and methods."""
rules = list(current_app.url_map.iter_rules())
if not rules:
click.echo("No routes were registered.")
return
ignored_methods = set() if all_methods else {"HEAD", "OPTIONS"}
host_matching = current_app.url_map.host_matching
has_domain = any(rule.host if host_matching else rule.subdomain for rule in rules)
rows = []
for rule in rules:
row = [
rule.endpoint,
", ".join(sorted((rule.methods or set()) - ignored_methods)),
]
if has_domain:
row.append((rule.host if host_matching else rule.subdomain) or "")
row.append(rule.rule)
rows.append(row)
headers = ["Endpoint", "Methods"]
sorts = ["endpoint", "methods"]
if has_domain:
headers.append("Host" if host_matching else "Subdomain")
sorts.append("domain")
headers.append("Rule")
sorts.append("rule")
try:
rows.sort(key=itemgetter(sorts.index(sort)))
except ValueError:
pass
rows.insert(0, headers)
widths = [max(len(row[i]) for row in rows) for i in range(len(headers))]
rows.insert(1, ["-" * w for w in widths])
template = " ".join(f"{{{i}:<{w}}}" for i, w in enumerate(widths))
for row in rows:
click.echo(template.format(*row))
cli = FlaskGroup(
name="flask",
help="""\
A general utility script for Flask applications.
An application to load must be given with the '--app' option,
'FLASK_APP' environment variable, or with a 'wsgi.py' or 'app.py' file
in the current directory.
""",
)
def main() -> None:
cli.main()
if __name__ == "__main__":
main()
server-temp/venv/lib/python3.10/site-packages/flask/config.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import errno
import json
import os
import types
import typing as t
from werkzeug.utils import import_string
if t.TYPE_CHECKING:
import typing_extensions as te
from .sansio.app import App
T = t.TypeVar("T")
class ConfigAttribute(t.Generic[T]):
"""Makes an attribute forward to the config"""
def __init__(
self, name: str, get_converter: t.Callable[[t.Any], T] | None = None
) -> None:
self.__name__ = name
self.get_converter = get_converter
@t.overload
def __get__(self, obj: None, owner: None) -> te.Self:
...
@t.overload
def __get__(self, obj: App, owner: type[App]) -> T:
...
def __get__(self, obj: App | None, owner: type[App] | None = None) -> T | te.Self:
if obj is None:
return self
rv = obj.config[self.__name__]
if self.get_converter is not None:
rv = self.get_converter(rv)
return rv # type: ignore[no-any-return]
def __set__(self, obj: App, value: t.Any) -> None:
obj.config[self.__name__] = value
class Config(dict): # type: ignore[type-arg]
"""Works exactly like a dict but provides ways to fill it from files
or special dictionaries. There are two common patterns to populate the
config.
Either you can fill the config from a config file::
app.config.from_pyfile('yourconfig.cfg')
Or alternatively you can define the configuration options in the
module that calls :meth:`from_object` or provide an import path to
a module that should be loaded. It is also possible to tell it to
use the same module and with that provide the configuration values
just before the call::
DEBUG = True
SECRET_KEY = 'development key'
app.config.from_object(__name__)
In both cases (loading from any Python file or loading from modules),
only uppercase keys are added to the config. This makes it possible to use
lowercase values in the config file for temporary values that are not added
to the config or to define the config keys in the same file that implements
the application.
Probably the most interesting way to load configurations is from an
environment variable pointing to a file::
app.config.from_envvar('YOURAPPLICATION_SETTINGS')
In this case before launching the application you have to set this
environment variable to the file you want to use. On Linux and OS X
use the export statement::
export YOURAPPLICATION_SETTINGS='/path/to/config/file'
On windows use `set` instead.
:param root_path: path to which files are read relative from. When the
config object is created by the application, this is
the application's :attr:`~flask.Flask.root_path`.
:param defaults: an optional dictionary of default values
"""
def __init__(
self,
root_path: str | os.PathLike[str],
defaults: dict[str, t.Any] | None = None,
) -> None:
super().__init__(defaults or {})
self.root_path = root_path
def from_envvar(self, variable_name: str, silent: bool = False) -> bool:
"""Loads a configuration from an environment variable pointing to
a configuration file. This is basically just a shortcut with nicer
error messages for this line of code::
app.config.from_pyfile(os.environ['YOURAPPLICATION_SETTINGS'])
:param variable_name: name of the environment variable
:param silent: set to ``True`` if you want silent failure for missing
files.
:return: ``True`` if the file was loaded successfully.
"""
rv = os.environ.get(variable_name)
if not rv:
if silent:
return False
raise RuntimeError(
f"The environment variable {variable_name!r} is not set"
" and as such configuration could not be loaded. Set"
" this variable and make it point to a configuration"
" file"
)
return self.from_pyfile(rv, silent=silent)
def from_prefixed_env(
self, prefix: str = "FLASK", *, loads: t.Callable[[str], t.Any] = json.loads
) -> bool:
"""Load any environment variables that start with ``FLASK_``,
dropping the prefix from the env key for the config key. Values
are passed through a loading function to attempt to convert them
to more specific types than strings.
Keys are loaded in :func:`sorted` order.
The default loading function attempts to parse values as any
valid JSON type, including dicts and lists.
Specific items in nested dicts can be set by separating the
keys with double underscores (``__``). If an intermediate key
doesn't exist, it will be initialized to an empty dict.
:param prefix: Load env vars that start with this prefix,
separated with an underscore (``_``).
:param loads: Pass each string value to this function and use
the returned value as the config value. If any error is
raised it is ignored and the value remains a string. The
default is :func:`json.loads`.
.. versionadded:: 2.1
"""
prefix = f"{prefix}_"
len_prefix = len(prefix)
for key in sorted(os.environ):
if not key.startswith(prefix):
continue
value = os.environ[key]
try:
value = loads(value)
except Exception:
# Keep the value as a string if loading failed.
pass
# Change to key.removeprefix(prefix) on Python >= 3.9.
key = key[len_prefix:]
if "__" not in key:
# A non-nested key, set directly.
self[key] = value
continue
# Traverse nested dictionaries with keys separated by "__".
current = self
*parts, tail = key.split("__")
for part in parts:
# If an intermediate dict does not exist, create it.
if part not in current:
current[part] = {}
current = current[part]
current[tail] = value
return True
def from_pyfile(
self, filename: str | os.PathLike[str], silent: bool = False
) -> bool:
"""Updates the values in the config from a Python file. This function
behaves as if the file was imported as module with the
:meth:`from_object` function.
:param filename: the filename of the config. This can either be an
absolute filename or a filename relative to the
root path.
:param silent: set to ``True`` if you want silent failure for missing
files.
:return: ``True`` if the file was loaded successfully.
.. versionadded:: 0.7
`silent` parameter.
"""
filename = os.path.join(self.root_path, filename)
d = types.ModuleType("config")
d.__file__ = filename
try:
with open(filename, mode="rb") as config_file:
exec(compile(config_file.read(), filename, "exec"), d.__dict__)
except OSError as e:
if silent and e.errno in (errno.ENOENT, errno.EISDIR, errno.ENOTDIR):
return False
e.strerror = f"Unable to load configuration file ({e.strerror})"
raise
self.from_object(d)
return True
def from_object(self, obj: object | str) -> None:
"""Updates the values from the given object. An object can be of one
of the following two types:
- a string: in this case the object with that name will be imported
- an actual object reference: that object is used directly
Objects are usually either modules or classes. :meth:`from_object`
loads only the uppercase attributes of the module/class. A ``dict``
object will not work with :meth:`from_object` because the keys of a
``dict`` are not attributes of the ``dict`` class.
Example of module-based configuration::
app.config.from_object('yourapplication.default_config')
from yourapplication import default_config
app.config.from_object(default_config)
Nothing is done to the object before loading. If the object is a
class and has ``@property`` attributes, it needs to be
instantiated before being passed to this method.
You should not use this function to load the actual configuration but
rather configuration defaults. The actual config should be loaded
with :meth:`from_pyfile` and ideally from a location not within the
package because the package might be installed system wide.
See :ref:`config-dev-prod` for an example of class-based configuration
using :meth:`from_object`.
:param obj: an import name or object
"""
if isinstance(obj, str):
obj = import_string(obj)
for key in dir(obj):
if key.isupper():
self[key] = getattr(obj, key)
def from_file(
self,
filename: str | os.PathLike[str],
load: t.Callable[[t.IO[t.Any]], t.Mapping[str, t.Any]],
silent: bool = False,
text: bool = True,
) -> bool:
"""Update the values in the config from a file that is loaded
using the ``load`` parameter. The loaded data is passed to the
:meth:`from_mapping` method.
.. code-block:: python
import json
app.config.from_file("config.json", load=json.load)
import tomllib
app.config.from_file("config.toml", load=tomllib.load, text=False)
:param filename: The path to the data file. This can be an
absolute path or relative to the config root path.
:param load: A callable that takes a file handle and returns a
mapping of loaded data from the file.
:type load: ``Callable[[Reader], Mapping]`` where ``Reader``
implements a ``read`` method.
:param silent: Ignore the file if it doesn't exist.
:param text: Open the file in text or binary mode.
:return: ``True`` if the file was loaded successfully.
.. versionchanged:: 2.3
The ``text`` parameter was added.
.. versionadded:: 2.0
"""
filename = os.path.join(self.root_path, filename)
try:
with open(filename, "r" if text else "rb") as f:
obj = load(f)
except OSError as e:
if silent and e.errno in (errno.ENOENT, errno.EISDIR):
return False
e.strerror = f"Unable to load configuration file ({e.strerror})"
raise
return self.from_mapping(obj)
def from_mapping(
self, mapping: t.Mapping[str, t.Any] | None = None, **kwargs: t.Any
) -> bool:
"""Updates the config like :meth:`update` ignoring items with
non-upper keys.
:return: Always returns ``True``.
.. versionadded:: 0.11
"""
mappings: dict[str, t.Any] = {}
if mapping is not None:
mappings.update(mapping)
mappings.update(kwargs)
for key, value in mappings.items():
if key.isupper():
self[key] = value
return True
def get_namespace(
self, namespace: str, lowercase: bool = True, trim_namespace: bool = True
) -> dict[str, t.Any]:
"""Returns a dictionary containing a subset of configuration options
that match the specified namespace/prefix. Example usage::
app.config['IMAGE_STORE_TYPE'] = 'fs'
app.config['IMAGE_STORE_PATH'] = '/var/app/images'
app.config['IMAGE_STORE_BASE_URL'] = 'http://img.website.com'
image_store_config = app.config.get_namespace('IMAGE_STORE_')
The resulting dictionary `image_store_config` would look like::
{
'type': 'fs',
'path': '/var/app/images',
'base_url': 'http://img.website.com'
}
This is often useful when configuration options map directly to
keyword arguments in functions or class constructors.
:param namespace: a configuration namespace
:param lowercase: a flag indicating if the keys of the resulting
dictionary should be lowercase
:param trim_namespace: a flag indicating if the keys of the resulting
dictionary should not include the namespace
.. versionadded:: 0.11
"""
rv = {}
for k, v in self.items():
if not k.startswith(namespace):
continue
if trim_namespace:
key = k[len(namespace) :]
else:
key = k
if lowercase:
key = key.lower()
rv[key] = v
return rv
def __repr__(self) -> str:
return f"<{type(self).__name__} {dict.__repr__(self)}>"
server-temp/venv/lib/python3.10/site-packages/flask/ctx.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import contextvars
import sys
import typing as t
from functools import update_wrapper
from types import TracebackType
from werkzeug.exceptions import HTTPException
from . import typing as ft
from .globals import _cv_app
from .globals import _cv_request
from .signals import appcontext_popped
from .signals import appcontext_pushed
if t.TYPE_CHECKING: # pragma: no cover
from _typeshed.wsgi import WSGIEnvironment
from .app import Flask
from .sessions import SessionMixin
from .wrappers import Request
# a singleton sentinel value for parameter defaults
_sentinel = object()
class _AppCtxGlobals:
"""A plain object. Used as a namespace for storing data during an
application context.
Creating an app context automatically creates this object, which is
made available as the :data:`g` proxy.
.. describe:: 'key' in g
Check whether an attribute is present.
.. versionadded:: 0.10
.. describe:: iter(g)
Return an iterator over the attribute names.
.. versionadded:: 0.10
"""
# Define attr methods to let mypy know this is a namespace object
# that has arbitrary attributes.
def __getattr__(self, name: str) -> t.Any:
try:
return self.__dict__[name]
except KeyError:
raise AttributeError(name) from None
def __setattr__(self, name: str, value: t.Any) -> None:
self.__dict__[name] = value
def __delattr__(self, name: str) -> None:
try:
del self.__dict__[name]
except KeyError:
raise AttributeError(name) from None
def get(self, name: str, default: t.Any | None = None) -> t.Any:
"""Get an attribute by name, or a default value. Like
:meth:`dict.get`.
:param name: Name of attribute to get.
:param default: Value to return if the attribute is not present.
.. versionadded:: 0.10
"""
return self.__dict__.get(name, default)
def pop(self, name: str, default: t.Any = _sentinel) -> t.Any:
"""Get and remove an attribute by name. Like :meth:`dict.pop`.
:param name: Name of attribute to pop.
:param default: Value to return if the attribute is not present,
instead of raising a ``KeyError``.
.. versionadded:: 0.11
"""
if default is _sentinel:
return self.__dict__.pop(name)
else:
return self.__dict__.pop(name, default)
def setdefault(self, name: str, default: t.Any = None) -> t.Any:
"""Get the value of an attribute if it is present, otherwise
set and return a default value. Like :meth:`dict.setdefault`.
:param name: Name of attribute to get.
:param default: Value to set and return if the attribute is not
present.
.. versionadded:: 0.11
"""
return self.__dict__.setdefault(name, default)
def __contains__(self, item: str) -> bool:
return item in self.__dict__
def __iter__(self) -> t.Iterator[str]:
return iter(self.__dict__)
def __repr__(self) -> str:
ctx = _cv_app.get(None)
if ctx is not None:
return f"<flask.g of '{ctx.app.name}'>"
return object.__repr__(self)
def after_this_request(
f: ft.AfterRequestCallable[t.Any],
) -> ft.AfterRequestCallable[t.Any]:
"""Executes a function after this request. This is useful to modify
response objects. The function is passed the response object and has
to return the same or a new one.
Example::
@app.route('/')
def index():
@after_this_request
def add_header(response):
response.headers['X-Foo'] = 'Parachute'
return response
return 'Hello World!'
This is more useful if a function other than the view function wants to
modify a response. For instance think of a decorator that wants to add
some headers without converting the return value into a response object.
.. versionadded:: 0.9
"""
ctx = _cv_request.get(None)
if ctx is None:
raise RuntimeError(
"'after_this_request' can only be used when a request"
" context is active, such as in a view function."
)
ctx._after_request_functions.append(f)
return f
F = t.TypeVar("F", bound=t.Callable[..., t.Any])
def copy_current_request_context(f: F) -> F:
"""A helper function that decorates a function to retain the current
request context. This is useful when working with greenlets. The moment
the function is decorated a copy of the request context is created and
then pushed when the function is called. The current session is also
included in the copied request context.
Example::
import gevent
from flask import copy_current_request_context
@app.route('/')
def index():
@copy_current_request_context
def do_some_work():
# do some work here, it can access flask.request or
# flask.session like you would otherwise in the view function.
...
gevent.spawn(do_some_work)
return 'Regular response'
.. versionadded:: 0.10
"""
ctx = _cv_request.get(None)
if ctx is None:
raise RuntimeError(
"'copy_current_request_context' can only be used when a"
" request context is active, such as in a view function."
)
ctx = ctx.copy()
def wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any:
with ctx: # type: ignore[union-attr]
return ctx.app.ensure_sync(f)(*args, **kwargs) # type: ignore[union-attr]
return update_wrapper(wrapper, f) # type: ignore[return-value]
def has_request_context() -> bool:
"""If you have code that wants to test if a request context is there or
not this function can be used. For instance, you may want to take advantage
of request information if the request object is available, but fail
silently if it is unavailable.
::
class User(db.Model):
def __init__(self, username, remote_addr=None):
self.username = username
if remote_addr is None and has_request_context():
remote_addr = request.remote_addr
self.remote_addr = remote_addr
Alternatively you can also just test any of the context bound objects
(such as :class:`request` or :class:`g`) for truthness::
class User(db.Model):
def __init__(self, username, remote_addr=None):
self.username = username
if remote_addr is None and request:
remote_addr = request.remote_addr
self.remote_addr = remote_addr
.. versionadded:: 0.7
"""
return _cv_request.get(None) is not None
def has_app_context() -> bool:
"""Works like :func:`has_request_context` but for the application
context. You can also just do a boolean check on the
:data:`current_app` object instead.
.. versionadded:: 0.9
"""
return _cv_app.get(None) is not None
class AppContext:
"""The app context contains application-specific information. An app
context is created and pushed at the beginning of each request if
one is not already active. An app context is also pushed when
running CLI commands.
"""
def __init__(self, app: Flask) -> None:
self.app = app
self.url_adapter = app.create_url_adapter(None)
self.g: _AppCtxGlobals = app.app_ctx_globals_class()
self._cv_tokens: list[contextvars.Token[AppContext]] = []
def push(self) -> None:
"""Binds the app context to the current context."""
self._cv_tokens.append(_cv_app.set(self))
appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync)
def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore
"""Pops the app context."""
try:
if len(self._cv_tokens) == 1:
if exc is _sentinel:
exc = sys.exc_info()[1]
self.app.do_teardown_appcontext(exc)
finally:
ctx = _cv_app.get()
_cv_app.reset(self._cv_tokens.pop())
if ctx is not self:
raise AssertionError(
f"Popped wrong app context. ({ctx!r} instead of {self!r})"
)
appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync)
def __enter__(self) -> AppContext:
self.push()
return self
def __exit__(
self,
exc_type: type | None,
exc_value: BaseException | None,
tb: TracebackType | None,
) -> None:
self.pop(exc_value)
class RequestContext:
"""The request context contains per-request information. The Flask
app creates and pushes it at the beginning of the request, then pops
it at the end of the request. It will create the URL adapter and
request object for the WSGI environment provided.
Do not attempt to use this class directly, instead use
:meth:`~flask.Flask.test_request_context` and
:meth:`~flask.Flask.request_context` to create this object.
When the request context is popped, it will evaluate all the
functions registered on the application for teardown execution
(:meth:`~flask.Flask.teardown_request`).
The request context is automatically popped at the end of the
request. When using the interactive debugger, the context will be
restored so ``request`` is still accessible. Similarly, the test
client can preserve the context after the request ends. However,
teardown functions may already have closed some resources such as
database connections.
"""
def __init__(
self,
app: Flask,
environ: WSGIEnvironment,
request: Request | None = None,
session: SessionMixin | None = None,
) -> None:
self.app = app
if request is None:
request = app.request_class(environ)
request.json_module = app.json
self.request: Request = request
self.url_adapter = None
try:
self.url_adapter = app.create_url_adapter(self.request)
except HTTPException as e:
self.request.routing_exception = e
self.flashes: list[tuple[str, str]] | None = None
self.session: SessionMixin | None = session
# Functions that should be executed after the request on the response
# object. These will be called before the regular "after_request"
# functions.
self._after_request_functions: list[ft.AfterRequestCallable[t.Any]] = []
self._cv_tokens: list[
tuple[contextvars.Token[RequestContext], AppContext | None]
] = []
def copy(self) -> RequestContext:
"""Creates a copy of this request context with the same request object.
This can be used to move a request context to a different greenlet.
Because the actual request object is the same this cannot be used to
move a request context to a different thread unless access to the
request object is locked.
.. versionadded:: 0.10
.. versionchanged:: 1.1
The current session object is used instead of reloading the original
data. This prevents `flask.session` pointing to an out-of-date object.
"""
return self.__class__(
self.app,
environ=self.request.environ,
request=self.request,
session=self.session,
)
def match_request(self) -> None:
"""Can be overridden by a subclass to hook into the matching
of the request.
"""
try:
result = self.url_adapter.match(return_rule=True) # type: ignore
self.request.url_rule, self.request.view_args = result # type: ignore
except HTTPException as e:
self.request.routing_exception = e
def push(self) -> None:
# Before we push the request context we have to ensure that there
# is an application context.
app_ctx = _cv_app.get(None)
if app_ctx is None or app_ctx.app is not self.app:
app_ctx = self.app.app_context()
app_ctx.push()
else:
app_ctx = None
self._cv_tokens.append((_cv_request.set(self), app_ctx))
# Open the session at the moment that the request context is available.
# This allows a custom open_session method to use the request context.
# Only open a new session if this is the first time the request was
# pushed, otherwise stream_with_context loses the session.
if self.session is None:
session_interface = self.app.session_interface
self.session = session_interface.open_session(self.app, self.request)
if self.session is None:
self.session = session_interface.make_null_session(self.app)
# Match the request URL after loading the session, so that the
# session is available in custom URL converters.
if self.url_adapter is not None:
self.match_request()
def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore
"""Pops the request context and unbinds it by doing that. This will
also trigger the execution of functions registered by the
:meth:`~flask.Flask.teardown_request` decorator.
.. versionchanged:: 0.9
Added the `exc` argument.
"""
clear_request = len(self._cv_tokens) == 1
try:
if clear_request:
if exc is _sentinel:
exc = sys.exc_info()[1]
self.app.do_teardown_request(exc)
request_close = getattr(self.request, "close", None)
if request_close is not None:
request_close()
finally:
ctx = _cv_request.get()
token, app_ctx = self._cv_tokens.pop()
_cv_request.reset(token)
# get rid of circular dependencies at the end of the request
# so that we don't require the GC to be active.
if clear_request:
ctx.request.environ["werkzeug.request"] = None
if app_ctx is not None:
app_ctx.pop(exc)
if ctx is not self:
raise AssertionError(
f"Popped wrong request context. ({ctx!r} instead of {self!r})"
)
def __enter__(self) -> RequestContext:
self.push()
return self
def __exit__(
self,
exc_type: type | None,
exc_value: BaseException | None,
tb: TracebackType | None,
) -> None:
self.pop(exc_value)
def __repr__(self) -> str:
return (
f"<{type(self).__name__} {self.request.url!r}"
f" [{self.request.method}] of {self.app.name}>"
)
server-temp/venv/lib/python3.10/site-packages/flask/debughelpers.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import typing as t
from jinja2.loaders import BaseLoader
from werkzeug.routing import RequestRedirect
from .blueprints import Blueprint
from .globals import request_ctx
from .sansio.app import App
if t.TYPE_CHECKING:
from .sansio.scaffold import Scaffold
from .wrappers import Request
class UnexpectedUnicodeError(AssertionError, UnicodeError):
"""Raised in places where we want some better error reporting for
unexpected unicode or binary data.
"""
class DebugFilesKeyError(KeyError, AssertionError):
"""Raised from request.files during debugging. The idea is that it can
provide a better error message than just a generic KeyError/BadRequest.
"""
def __init__(self, request: Request, key: str) -> None:
form_matches = request.form.getlist(key)
buf = [
f"You tried to access the file {key!r} in the request.files"
" dictionary but it does not exist. The mimetype for the"
f" request is {request.mimetype!r} instead of"
" 'multipart/form-data' which means that no file contents"
" were transmitted. To fix this error you should provide"
' enctype="multipart/form-data" in your form.'
]
if form_matches:
names = ", ".join(repr(x) for x in form_matches)
buf.append(
"\n\nThe browser instead transmitted some file names. "
f"This was submitted: {names}"
)
self.msg = "".join(buf)
def __str__(self) -> str:
return self.msg
class FormDataRoutingRedirect(AssertionError):
"""This exception is raised in debug mode if a routing redirect
would cause the browser to drop the method or body. This happens
when method is not GET, HEAD or OPTIONS and the status code is not
307 or 308.
"""
def __init__(self, request: Request) -> None:
exc = request.routing_exception
assert isinstance(exc, RequestRedirect)
buf = [
f"A request was sent to '{request.url}', but routing issued"
f" a redirect to the canonical URL '{exc.new_url}'."
]
if f"{request.base_url}/" == exc.new_url.partition("?")[0]:
buf.append(
" The URL was defined with a trailing slash. Flask"
" will redirect to the URL with a trailing slash if it"
" was accessed without one."
)
buf.append(
" Send requests to the canonical URL, or use 307 or 308 for"
" routing redirects. Otherwise, browsers will drop form"
" data.\n\n"
"This exception is only raised in debug mode."
)
super().__init__("".join(buf))
def attach_enctype_error_multidict(request: Request) -> None:
"""Patch ``request.files.__getitem__`` to raise a descriptive error
about ``enctype=multipart/form-data``.
:param request: The request to patch.
:meta private:
"""
oldcls = request.files.__class__
class newcls(oldcls): # type: ignore[valid-type, misc]
def __getitem__(self, key: str) -> t.Any:
try:
return super().__getitem__(key)
except KeyError as e:
if key not in request.form:
raise
raise DebugFilesKeyError(request, key).with_traceback(
e.__traceback__
) from None
newcls.__name__ = oldcls.__name__
newcls.__module__ = oldcls.__module__
request.files.__class__ = newcls
def _dump_loader_info(loader: BaseLoader) -> t.Iterator[str]:
yield f"class: {type(loader).__module__}.{type(loader).__name__}"
for key, value in sorted(loader.__dict__.items()):
if key.startswith("_"):
continue
if isinstance(value, (tuple, list)):
if not all(isinstance(x, str) for x in value):
continue
yield f"{key}:"
for item in value:
yield f" - {item}"
continue
elif not isinstance(value, (str, int, float, bool)):
continue
yield f"{key}: {value!r}"
def explain_template_loading_attempts(
app: App,
template: str,
attempts: list[
tuple[
BaseLoader,
Scaffold,
tuple[str, str | None, t.Callable[[], bool] | None] | None,
]
],
) -> None:
"""This should help developers understand what failed"""
info = [f"Locating template {template!r}:"]
total_found = 0
blueprint = None
if request_ctx and request_ctx.request.blueprint is not None:
blueprint = request_ctx.request.blueprint
for idx, (loader, srcobj, triple) in enumerate(attempts):
if isinstance(srcobj, App):
src_info = f"application {srcobj.import_name!r}"
elif isinstance(srcobj, Blueprint):
src_info = f"blueprint {srcobj.name!r} ({srcobj.import_name})"
else:
src_info = repr(srcobj)
info.append(f"{idx + 1:5}: trying loader of {src_info}")
for line in _dump_loader_info(loader):
info.append(f" {line}")
if triple is None:
detail = "no match"
else:
detail = f"found ({triple[1] or '<string>'!r})"
total_found += 1
info.append(f" -> {detail}")
seems_fishy = False
if total_found == 0:
info.append("Error: the template could not be found.")
seems_fishy = True
elif total_found > 1:
info.append("Warning: multiple loaders returned a match for the template.")
seems_fishy = True
if blueprint is not None and seems_fishy:
info.append(
" The template was looked up from an endpoint that belongs"
f" to the blueprint {blueprint!r}."
)
info.append(" Maybe you did not place a template in the right folder?")
info.append(" See https://flask.palletsprojects.com/blueprints/#templates")
app.logger.info("\n".join(info))
server-temp/venv/lib/python3.10/site-packages/flask/globals.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import typing as t
from contextvars import ContextVar
from werkzeug.local import LocalProxy
if t.TYPE_CHECKING: # pragma: no cover
from .app import Flask
from .ctx import _AppCtxGlobals
from .ctx import AppContext
from .ctx import RequestContext
from .sessions import SessionMixin
from .wrappers import Request
_no_app_msg = """\
Working outside of application context.
This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.\
"""
_cv_app: ContextVar[AppContext] = ContextVar("flask.app_ctx")
app_ctx: AppContext = LocalProxy( # type: ignore[assignment]
_cv_app, unbound_message=_no_app_msg
)
current_app: Flask = LocalProxy( # type: ignore[assignment]
_cv_app, "app", unbound_message=_no_app_msg
)
g: _AppCtxGlobals = LocalProxy( # type: ignore[assignment]
_cv_app, "g", unbound_message=_no_app_msg
)
_no_req_msg = """\
Working outside of request context.
This typically means that you attempted to use functionality that needed
an active HTTP request. Consult the documentation on testing for
information about how to avoid this problem.\
"""
_cv_request: ContextVar[RequestContext] = ContextVar("flask.request_ctx")
request_ctx: RequestContext = LocalProxy( # type: ignore[assignment]
_cv_request, unbound_message=_no_req_msg
)
request: Request = LocalProxy( # type: ignore[assignment]
_cv_request, "request", unbound_message=_no_req_msg
)
session: SessionMixin = LocalProxy( # type: ignore[assignment]
_cv_request, "session", unbound_message=_no_req_msg
)
server-temp/venv/lib/python3.10/site-packages/flask/helpers.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import importlib.util
import os
import sys
import typing as t
from datetime import datetime
from functools import lru_cache
from functools import update_wrapper
import werkzeug.utils
from werkzeug.exceptions import abort as _wz_abort
from werkzeug.utils import redirect as _wz_redirect
from werkzeug.wrappers import Response as BaseResponse
from .globals import _cv_request
from .globals import current_app
from .globals import request
from .globals import request_ctx
from .globals import session
from .signals import message_flashed
if t.TYPE_CHECKING: # pragma: no cover
from .wrappers import Response
def get_debug_flag() -> bool:
"""Get whether debug mode should be enabled for the app, indicated by the
:envvar:`FLASK_DEBUG` environment variable. The default is ``False``.
"""
val = os.environ.get("FLASK_DEBUG")
return bool(val and val.lower() not in {"0", "false", "no"})
def get_load_dotenv(default: bool = True) -> bool:
"""Get whether the user has disabled loading default dotenv files by
setting :envvar:`FLASK_SKIP_DOTENV`. The default is ``True``, load
the files.
:param default: What to return if the env var isn't set.
"""
val = os.environ.get("FLASK_SKIP_DOTENV")
if not val:
return default
return val.lower() in ("0", "false", "no")
def stream_with_context(
generator_or_function: t.Iterator[t.AnyStr] | t.Callable[..., t.Iterator[t.AnyStr]],
) -> t.Iterator[t.AnyStr]:
"""Request contexts disappear when the response is started on the server.
This is done for efficiency reasons and to make it less likely to encounter
memory leaks with badly written WSGI middlewares. The downside is that if
you are using streamed responses, the generator cannot access request bound
information any more.
This function however can help you keep the context around for longer::
from flask import stream_with_context, request, Response
@app.route('/stream')
def streamed_response():
@stream_with_context
def generate():
yield 'Hello '
yield request.args['name']
yield '!'
return Response(generate())
Alternatively it can also be used around a specific generator::
from flask import stream_with_context, request, Response
@app.route('/stream')
def streamed_response():
def generate():
yield 'Hello '
yield request.args['name']
yield '!'
return Response(stream_with_context(generate()))
.. versionadded:: 0.9
"""
try:
gen = iter(generator_or_function) # type: ignore[arg-type]
except TypeError:
def decorator(*args: t.Any, **kwargs: t.Any) -> t.Any:
gen = generator_or_function(*args, **kwargs) # type: ignore[operator]
return stream_with_context(gen)
return update_wrapper(decorator, generator_or_function) # type: ignore[arg-type]
def generator() -> t.Iterator[t.AnyStr | None]:
ctx = _cv_request.get(None)
if ctx is None:
raise RuntimeError(
"'stream_with_context' can only be used when a request"
" context is active, such as in a view function."
)
with ctx:
# Dummy sentinel. Has to be inside the context block or we're
# not actually keeping the context around.
yield None
# The try/finally is here so that if someone passes a WSGI level
# iterator in we're still running the cleanup logic. Generators
# don't need that because they are closed on their destruction
# automatically.
try:
yield from gen
finally:
if hasattr(gen, "close"):
gen.close()
# The trick is to start the generator. Then the code execution runs until
# the first dummy None is yielded at which point the context was already
# pushed. This item is discarded. Then when the iteration continues the
# real generator is executed.
wrapped_g = generator()
next(wrapped_g)
return wrapped_g # type: ignore[return-value]
def make_response(*args: t.Any) -> Response:
"""Sometimes it is necessary to set additional headers in a view. Because
views do not have to return response objects but can return a value that
is converted into a response object by Flask itself, it becomes tricky to
add headers to it. This function can be called instead of using a return
and you will get a response object which you can use to attach headers.
If view looked like this and you want to add a new header::
def index():
return render_template('index.html', foo=42)
You can now do something like this::
def index():
response = make_response(render_template('index.html', foo=42))
response.headers['X-Parachutes'] = 'parachutes are cool'
return response
This function accepts the very same arguments you can return from a
view function. This for example creates a response with a 404 error
code::
response = make_response(render_template('not_found.html'), 404)
The other use case of this function is to force the return value of a
view function into a response which is helpful with view
decorators::
response = make_response(view_function())
response.headers['X-Parachutes'] = 'parachutes are cool'
Internally this function does the following things:
- if no arguments are passed, it creates a new response argument
- if one argument is passed, :meth:`flask.Flask.make_response`
is invoked with it.
- if more than one argument is passed, the arguments are passed
to the :meth:`flask.Flask.make_response` function as tuple.
.. versionadded:: 0.6
"""
if not args:
return current_app.response_class()
if len(args) == 1:
args = args[0]
return current_app.make_response(args)
def url_for(
endpoint: str,
*,
_anchor: str | None = None,
_method: str | None = None,
_scheme: str | None = None,
_external: bool | None = None,
**values: t.Any,
) -> str:
"""Generate a URL to the given endpoint with the given values.
This requires an active request or application context, and calls
:meth:`current_app.url_for() <flask.Flask.url_for>`. See that method
for full documentation.
:param endpoint: The endpoint name associated with the URL to
generate. If this starts with a ``.``, the current blueprint
name (if any) will be used.
:param _anchor: If given, append this as ``#anchor`` to the URL.
:param _method: If given, generate the URL associated with this
method for the endpoint.
:param _scheme: If given, the URL will have this scheme if it is
external.
:param _external: If given, prefer the URL to be internal (False) or
require it to be external (True). External URLs include the
scheme and domain. When not in an active request, URLs are
external by default.
:param values: Values to use for the variable parts of the URL rule.
Unknown keys are appended as query string arguments, like
``?a=b&c=d``.
.. versionchanged:: 2.2
Calls ``current_app.url_for``, allowing an app to override the
behavior.
.. versionchanged:: 0.10
The ``_scheme`` parameter was added.
.. versionchanged:: 0.9
The ``_anchor`` and ``_method`` parameters were added.
.. versionchanged:: 0.9
Calls ``app.handle_url_build_error`` on build errors.
"""
return current_app.url_for(
endpoint,
_anchor=_anchor,
_method=_method,
_scheme=_scheme,
_external=_external,
**values,
)
def redirect(
location: str, code: int = 302, Response: type[BaseResponse] | None = None
) -> BaseResponse:
"""Create a redirect response object.
If :data:`~flask.current_app` is available, it will use its
:meth:`~flask.Flask.redirect` method, otherwise it will use
:func:`werkzeug.utils.redirect`.
:param location: The URL to redirect to.
:param code: The status code for the redirect.
:param Response: The response class to use. Not used when
``current_app`` is active, which uses ``app.response_class``.
.. versionadded:: 2.2
Calls ``current_app.redirect`` if available instead of always
using Werkzeug's default ``redirect``.
"""
if current_app:
return current_app.redirect(location, code=code)
return _wz_redirect(location, code=code, Response=Response)
def abort(code: int | BaseResponse, *args: t.Any, **kwargs: t.Any) -> t.NoReturn:
"""Raise an :exc:`~werkzeug.exceptions.HTTPException` for the given
status code.
If :data:`~flask.current_app` is available, it will call its
:attr:`~flask.Flask.aborter` object, otherwise it will use
:func:`werkzeug.exceptions.abort`.
:param code: The status code for the exception, which must be
registered in ``app.aborter``.
:param args: Passed to the exception.
:param kwargs: Passed to the exception.
.. versionadded:: 2.2
Calls ``current_app.aborter`` if available instead of always
using Werkzeug's default ``abort``.
"""
if current_app:
current_app.aborter(code, *args, **kwargs)
_wz_abort(code, *args, **kwargs)
def get_template_attribute(template_name: str, attribute: str) -> t.Any:
"""Loads a macro (or variable) a template exports. This can be used to
invoke a macro from within Python code. If you for example have a
template named :file:`_cider.html` with the following contents:
.. sourcecode:: html+jinja
{% macro hello(name) %}Hello {{ name }}!{% endmacro %}
You can access this from Python code like this::
hello = get_template_attribute('_cider.html', 'hello')
return hello('World')
.. versionadded:: 0.2
:param template_name: the name of the template
:param attribute: the name of the variable of macro to access
"""
return getattr(current_app.jinja_env.get_template(template_name).module, attribute)
def flash(message: str, category: str = "message") -> None:
"""Flashes a message to the next request. In order to remove the
flashed message from the session and to display it to the user,
the template has to call :func:`get_flashed_messages`.
.. versionchanged:: 0.3
`category` parameter added.
:param message: the message to be flashed.
:param category: the category for the message. The following values
are recommended: ``'message'`` for any kind of message,
``'error'`` for errors, ``'info'`` for information
messages and ``'warning'`` for warnings. However any
kind of string can be used as category.
"""
# Original implementation:
#
# session.setdefault('_flashes', []).append((category, message))
#
# This assumed that changes made to mutable structures in the session are
# always in sync with the session object, which is not true for session
# implementations that use external storage for keeping their keys/values.
flashes = session.get("_flashes", [])
flashes.append((category, message))
session["_flashes"] = flashes
app = current_app._get_current_object() # type: ignore
message_flashed.send(
app,
_async_wrapper=app.ensure_sync,
message=message,
category=category,
)
def get_flashed_messages(
with_categories: bool = False, category_filter: t.Iterable[str] = ()
) -> list[str] | list[tuple[str, str]]:
"""Pulls all flashed messages from the session and returns them.
Further calls in the same request to the function will return
the same messages. By default just the messages are returned,
but when `with_categories` is set to ``True``, the return value will
be a list of tuples in the form ``(category, message)`` instead.
Filter the flashed messages to one or more categories by providing those
categories in `category_filter`. This allows rendering categories in
separate html blocks. The `with_categories` and `category_filter`
arguments are distinct:
* `with_categories` controls whether categories are returned with message
text (``True`` gives a tuple, where ``False`` gives just the message text).
* `category_filter` filters the messages down to only those matching the
provided categories.
See :doc:`/patterns/flashing` for examples.
.. versionchanged:: 0.3
`with_categories` parameter added.
.. versionchanged:: 0.9
`category_filter` parameter added.
:param with_categories: set to ``True`` to also receive categories.
:param category_filter: filter of categories to limit return values. Only
categories in the list will be returned.
"""
flashes = request_ctx.flashes
if flashes is None:
flashes = session.pop("_flashes") if "_flashes" in session else []
request_ctx.flashes = flashes
if category_filter:
flashes = list(filter(lambda f: f[0] in category_filter, flashes))
if not with_categories:
return [x[1] for x in flashes]
return flashes
def _prepare_send_file_kwargs(**kwargs: t.Any) -> dict[str, t.Any]:
if kwargs.get("max_age") is None:
kwargs["max_age"] = current_app.get_send_file_max_age
kwargs.update(
environ=request.environ,
use_x_sendfile=current_app.config["USE_X_SENDFILE"],
response_class=current_app.response_class,
_root_path=current_app.root_path, # type: ignore
)
return kwargs
def send_file(
path_or_file: os.PathLike[t.AnyStr] | str | t.BinaryIO,
mimetype: str | None = None,
as_attachment: bool = False,
download_name: str | None = None,
conditional: bool = True,
etag: bool | str = True,
last_modified: datetime | int | float | None = None,
max_age: None | (int | t.Callable[[str | None], int | None]) = None,
) -> Response:
"""Send the contents of a file to the client.
The first argument can be a file path or a file-like object. Paths
are preferred in most cases because Werkzeug can manage the file and
get extra information from the path. Passing a file-like object
requires that the file is opened in binary mode, and is mostly
useful when building a file in memory with :class:`io.BytesIO`.
Never pass file paths provided by a user. The path is assumed to be
trusted, so a user could craft a path to access a file you didn't
intend. Use :func:`send_from_directory` to safely serve
user-requested paths from within a directory.
If the WSGI server sets a ``file_wrapper`` in ``environ``, it is
used, otherwise Werkzeug's built-in wrapper is used. Alternatively,
if the HTTP server supports ``X-Sendfile``, configuring Flask with
``USE_X_SENDFILE = True`` will tell the server to send the given
path, which is much more efficient than reading it in Python.
:param path_or_file: The path to the file to send, relative to the
current working directory if a relative path is given.
Alternatively, a file-like object opened in binary mode. Make
sure the file pointer is seeked to the start of the data.
:param mimetype: The MIME type to send for the file. If not
provided, it will try to detect it from the file name.
:param as_attachment: Indicate to a browser that it should offer to
save the file instead of displaying it.
:param download_name: The default name browsers will use when saving
the file. Defaults to the passed file name.
:param conditional: Enable conditional and range responses based on
request headers. Requires passing a file path and ``environ``.
:param etag: Calculate an ETag for the file, which requires passing
a file path. Can also be a string to use instead.
:param last_modified: The last modified time to send for the file,
in seconds. If not provided, it will try to detect it from the
file path.
:param max_age: How long the client should cache the file, in
seconds. If set, ``Cache-Control`` will be ``public``, otherwise
it will be ``no-cache`` to prefer conditional caching.
.. versionchanged:: 2.0
``download_name`` replaces the ``attachment_filename``
parameter. If ``as_attachment=False``, it is passed with
``Content-Disposition: inline`` instead.
.. versionchanged:: 2.0
``max_age`` replaces the ``cache_timeout`` parameter.
``conditional`` is enabled and ``max_age`` is not set by
default.
.. versionchanged:: 2.0
``etag`` replaces the ``add_etags`` parameter. It can be a
string to use instead of generating one.
.. versionchanged:: 2.0
Passing a file-like object that inherits from
:class:`~io.TextIOBase` will raise a :exc:`ValueError` rather
than sending an empty file.
.. versionadded:: 2.0
Moved the implementation to Werkzeug. This is now a wrapper to
pass some Flask-specific arguments.
.. versionchanged:: 1.1
``filename`` may be a :class:`~os.PathLike` object.
.. versionchanged:: 1.1
Passing a :class:`~io.BytesIO` object supports range requests.
.. versionchanged:: 1.0.3
Filenames are encoded with ASCII instead of Latin-1 for broader
compatibility with WSGI servers.
.. versionchanged:: 1.0
UTF-8 filenames as specified in :rfc:`2231` are supported.
.. versionchanged:: 0.12
The filename is no longer automatically inferred from file
objects. If you want to use automatic MIME and etag support,
pass a filename via ``filename_or_fp`` or
``attachment_filename``.
.. versionchanged:: 0.12
``attachment_filename`` is preferred over ``filename`` for MIME
detection.
.. versionchanged:: 0.9
``cache_timeout`` defaults to
:meth:`Flask.get_send_file_max_age`.
.. versionchanged:: 0.7
MIME guessing and etag support for file-like objects was
removed because it was unreliable. Pass a filename if you are
able to, otherwise attach an etag yourself.
.. versionchanged:: 0.5
The ``add_etags``, ``cache_timeout`` and ``conditional``
parameters were added. The default behavior is to add etags.
.. versionadded:: 0.2
"""
return werkzeug.utils.send_file( # type: ignore[return-value]
**_prepare_send_file_kwargs(
path_or_file=path_or_file,
environ=request.environ,
mimetype=mimetype,
as_attachment=as_attachment,
download_name=download_name,
conditional=conditional,
etag=etag,
last_modified=last_modified,
max_age=max_age,
)
)
def send_from_directory(
directory: os.PathLike[str] | str,
path: os.PathLike[str] | str,
**kwargs: t.Any,
) -> Response:
"""Send a file from within a directory using :func:`send_file`.
.. code-block:: python
@app.route("/uploads/<path:name>")
def download_file(name):
return send_from_directory(
app.config['UPLOAD_FOLDER'], name, as_attachment=True
)
This is a secure way to serve files from a folder, such as static
files or uploads. Uses :func:`~werkzeug.security.safe_join` to
ensure the path coming from the client is not maliciously crafted to
point outside the specified directory.
If the final path does not point to an existing regular file,
raises a 404 :exc:`~werkzeug.exceptions.NotFound` error.
:param directory: The directory that ``path`` must be located under,
relative to the current application's root path.
:param path: The path to the file to send, relative to
``directory``.
:param kwargs: Arguments to pass to :func:`send_file`.
.. versionchanged:: 2.0
``path`` replaces the ``filename`` parameter.
.. versionadded:: 2.0
Moved the implementation to Werkzeug. This is now a wrapper to
pass some Flask-specific arguments.
.. versionadded:: 0.5
"""
return werkzeug.utils.send_from_directory( # type: ignore[return-value]
directory, path, **_prepare_send_file_kwargs(**kwargs)
)
def get_root_path(import_name: str) -> str:
"""Find the root path of a package, or the path that contains a
module. If it cannot be found, returns the current working
directory.
Not to be confused with the value returned by :func:`find_package`.
:meta private:
"""
# Module already imported and has a file attribute. Use that first.
mod = sys.modules.get(import_name)
if mod is not None and hasattr(mod, "__file__") and mod.__file__ is not None:
return os.path.dirname(os.path.abspath(mod.__file__))
# Next attempt: check the loader.
try:
spec = importlib.util.find_spec(import_name)
if spec is None:
raise ValueError
except (ImportError, ValueError):
loader = None
else:
loader = spec.loader
# Loader does not exist or we're referring to an unloaded main
# module or a main module without path (interactive sessions), go
# with the current working directory.
if loader is None:
return os.getcwd()
if hasattr(loader, "get_filename"):
filepath = loader.get_filename(import_name)
else:
# Fall back to imports.
__import__(import_name)
mod = sys.modules[import_name]
filepath = getattr(mod, "__file__", None)
# If we don't have a file path it might be because it is a
# namespace package. In this case pick the root path from the
# first module that is contained in the package.
if filepath is None:
raise RuntimeError(
"No root path can be found for the provided module"
f" {import_name!r}. This can happen because the module"
" came from an import hook that does not provide file"
" name information or because it's a namespace package."
" In this case the root path needs to be explicitly"
" provided."
)
# filepath is import_name.py for a module, or __init__.py for a package.
return os.path.dirname(os.path.abspath(filepath)) # type: ignore[no-any-return]
@lru_cache(maxsize=None)
def _split_blueprint_path(name: str) -> list[str]:
out: list[str] = [name]
if "." in name:
out.extend(_split_blueprint_path(name.rpartition(".")[0]))
return out
server-temp/venv/lib/python3.10/site-packages/flask/json/__init__.py deleted 100644 → 0
View file @ 03bf2ef9
from __future__ import annotations
import json as _json
import typing as t
from ..globals import current_app
from .provider import _default
if t.TYPE_CHECKING: # pragma: no cover
from ..wrappers import Response
def dumps(obj: t.Any, **kwargs: t.Any) -> str:
"""Serialize data as JSON.
If :data:`~flask.current_app` is available, it will use its
:meth:`app.json.dumps() <flask.json.provider.JSONProvider.dumps>`
method, otherwise it will use :func:`json.dumps`.
:param obj: The data to serialize.
:param kwargs: Arguments passed to the ``dumps`` implementation.
.. versionchanged:: 2.3
The ``app`` parameter was removed.
.. versionchanged:: 2.2
Calls ``current_app.json.dumps``, allowing an app to override
the behavior.
.. versionchanged:: 2.0.2
:class:`decimal.Decimal` is supported by converting to a string.
.. versionchanged:: 2.0
``encoding`` will be removed in Flask 2.1.
.. versionchanged:: 1.0.3
``app`` can be passed directly, rather than requiring an app
context for configuration.
"""
if current_app:
return current_app.json.dumps(obj, **kwargs)
kwargs.setdefault("default", _default)
return _json.dumps(obj, **kwargs)
def dump(obj: t.Any, fp: t.IO[str], **kwargs: t.Any) -> None:
"""Serialize data as JSON and write to a file.
If :data:`~flask.current_app` is available, it will use its
:meth:`app.json.dump() <flask.json.provider.JSONProvider.dump>`
method, otherwise it will use :func:`json.dump`.
: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: Arguments passed to the ``dump`` implementation.
.. versionchanged:: 2.3
The ``app`` parameter was removed.
.. versionchanged:: 2.2
Calls ``current_app.json.dump``, allowing an app to override
the behavior.
.. versionchanged:: 2.0
Writing to a binary file, and the ``encoding`` argument, will be
removed in Flask 2.1.
"""
if current_app:
current_app.json.dump(obj, fp, **kwargs)
else:
kwargs.setdefault("default", _default)
_json.dump(obj, fp, **kwargs)
def loads(s: str | bytes, **kwargs: t.Any) -> t.Any:
"""Deserialize data as JSON.
If :data:`~flask.current_app` is available, it will use its
:meth:`app.json.loads() <flask.json.provider.JSONProvider.loads>`
method, otherwise it will use :func:`json.loads`.
:param s: Text or UTF-8 bytes.
:param kwargs: Arguments passed to the ``loads`` implementation.
.. versionchanged:: 2.3
The ``app`` parameter was removed.
.. versionchanged:: 2.2
Calls ``current_app.json.loads``, allowing an app to override
the behavior.
.. versionchanged:: 2.0
``encoding`` will be removed in Flask 2.1. The data must be a
string or UTF-8 bytes.
.. versionchanged:: 1.0.3
``app`` can be passed directly, rather than requiring an app
context for configuration.
"""
if current_app:
return current_app.json.loads(s, **kwargs)
return _json.loads(s, **kwargs)
def load(fp: t.IO[t.AnyStr], **kwargs: t.Any) -> t.Any:
"""Deserialize data as JSON read from a file.
If :data:`~flask.current_app` is available, it will use its
:meth:`app.json.load() <flask.json.provider.JSONProvider.load>`
method, otherwise it will use :func:`json.load`.
:param fp: A file opened for reading text or UTF-8 bytes.
:param kwargs: Arguments passed to the ``load`` implementation.
.. versionchanged:: 2.3
The ``app`` parameter was removed.
.. versionchanged:: 2.2
Calls ``current_app.json.load``, allowing an app to override
the behavior.
.. versionchanged:: 2.2
The ``app`` parameter will be removed in Flask 2.3.
.. versionchanged:: 2.0
``encoding`` will be removed in Flask 2.1. The file must be text
mode, or binary mode with UTF-8 bytes.
"""
if current_app:
return current_app.json.load(fp, **kwargs)
return _json.load(fp, **kwargs)
def jsonify(*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. A dict or list returned from a view will be converted to a
JSON response automatically without needing to call this.
This requires an active request or application context, and calls
:meth:`app.json.response() <flask.json.provider.JSONProvider.response>`.
In debug mode, the output is formatted with indentation to make it
easier to read. This may also be controlled by the provider.
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.
.. versionchanged:: 2.2
Calls ``current_app.json.response``, allowing an app to override
the behavior.
.. versionchanged:: 2.0.2
:class:`decimal.Decimal` is supported by converting to a string.
.. versionchanged:: 0.11
Added support for serializing top-level arrays. This was a
security risk in ancient browsers. See :ref:`security-json`.
.. versionadded:: 0.2
"""
return current_app.json.response(*args, **kwargs) # type: ignore[return-value]