orbit-oss/flask
4
4
Fork 2
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #3933 from pallets/more-scaffold

Browse source 
more work on Scaffold base class
This commit is contained in:
David Lord 2021-03-10 11:02:16 -08:00 • committed by GitHub
commit 510c38cc63
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 587 additions and 577 deletions
Download patch file Download diff file Expand all files Collapse all files

83
src/flask/app.py
View file

@ -29,7 +29,6 @@ from .globals import _request_ctx_stack
from .globals import g
from .globals import request
from .globals import session
from .helpers import find_package
from .helpers import get_debug_flag
from .helpers import get_env
from .helpers import get_flashed_messages
@ -40,6 +39,7 @@ from .json import jsonify
from .logging import create_logger
from .scaffold import _endpoint_from_view_func
from .scaffold import _sentinel
from .scaffold import find_package
from .scaffold import Scaffold
from .scaffold import setupmethod
from .sessions import SecureCookieSessionInterface
@ -346,21 +346,6 @@ class Flask(Scaffold):
#: .. versionadded:: 0.8
session_interface = SecureCookieSessionInterface()
# TODO remove the next three attrs when Sphinx :inherited-members: works
# https://github.com/sphinx-doc/sphinx/issues/741
#: The name of the package or module that this app belongs to. Do not
#: change this once it is set by the constructor.
import_name = None
#: Location of the template files to be added to the template lookup.
#: ``None`` if templates should not be added.
template_folder = None
#: Absolute path to the package on the filesystem. Used to look up
#: resources contained in the package.
root_path = None
def __init__(
self,
import_name,
@ -1017,58 +1002,6 @@ class Flask(Scaffold):
provide_automatic_options=None,
**options,
):
"""Connects a URL rule. Works exactly like the :meth:`route`
decorator. If a view_func is provided it will be registered with the
endpoint.
Basically this example::
@app.route('/')
def index():
pass
Is equivalent to the following::
def index():
pass
app.add_url_rule('/', 'index', index)
If the view_func is not provided you will need to connect the endpoint
to a view function like so::
app.view_functions['index'] = index
Internally :meth:`route` invokes :meth:`add_url_rule` so if you want
to customize the behavior via subclassing you only need to change
this method.
For more information refer to :ref:`url-route-registrations`.
.. versionchanged:: 0.2
`view_func` parameter added.
.. versionchanged:: 0.6
``OPTIONS`` is added automatically as method.
:param rule: the URL rule as string
:param endpoint: the endpoint for the registered URL rule. Flask
itself assumes the name of the view function as
endpoint
:param view_func: the function to call when serving a request to the
provided endpoint
:param provide_automatic_options: controls whether the ``OPTIONS``
method should be added automatically. This can also be controlled
by setting the ``view_func.provide_automatic_options = False``
before adding the rule.
:param options: the options to be forwarded to the underlying
:class:`~werkzeug.routing.Rule` object. A change
to Werkzeug is handling of method options. methods
is a list of methods this rule should be limited
to (``GET``, ``POST`` etc.). By default a rule
just listens for ``GET`` (and implicitly ``HEAD``).
Starting with Flask 0.6, ``OPTIONS`` is implicitly
added and handled by the standard request handling.
"""
if endpoint is None:
endpoint = _endpoint_from_view_func(view_func)
options["endpoint"] = endpoint
@ -1417,12 +1350,6 @@ class Flask(Scaffold):
always receive the ``InternalServerError``. The original
unhandled exception is available as ``e.original_exception``.
.. note::
Prior to Werkzeug 1.0.0, ``InternalServerError`` will not
always have an ``original_exception`` attribute. Use
``getattr(e, "original_exception", None)`` to simulate the
behavior for compatibility.
.. versionchanged:: 1.1.0
Always passes the ``InternalServerError`` instance to the
handler, setting ``original_exception`` to the unhandled
@ -2023,9 +1950,7 @@ class Flask(Scaffold):
def __call__(self, environ, start_response):
"""The WSGI server calls the Flask application object as the
WSGI application. This calls :meth:`wsgi_app` which can be
wrapped to applying middleware."""
WSGI application. This calls :meth:`wsgi_app`, which can be
wrapped to apply middleware.
"""
return self.wsgi_app(environ, start_response)
def __repr__(self):
return f"<{type(self).__name__} {self.name!r}>"

112
src/flask/blueprints.py
View file

@ -90,13 +90,6 @@ class Blueprint(Scaffold):
See :doc:`/blueprints` for more information.
.. versionchanged:: 1.1.0
Blueprints have a ``cli`` group to register nested CLI commands.
The ``cli_group`` parameter controls the name of the group under
the ``flask`` command.
.. versionadded:: 0.7
:param name: The name of the blueprint. Will be prepended to each
endpoint name.
:param import_name: The name of the blueprint package, usually
@ -121,37 +114,29 @@ class Blueprint(Scaffold):
default.
:param url_defaults: A dict of default values that blueprint routes
will receive by default.
:param root_path: By default, the blueprint will automatically set this
based on ``import_name``. In certain situations this automatic
detection can fail, so the path can be specified manually
instead.
:param root_path: By default, the blueprint will automatically set
this based on ``import_name``. In certain situations this
automatic detection can fail, so the path can be specified
manually instead.
.. versionchanged:: 1.1.0
Blueprints have a ``cli`` group to register nested CLI commands.
The ``cli_group`` parameter controls the name of the group under
the ``flask`` command.
.. versionadded:: 0.7
"""
warn_on_modifications = False
_got_registered_once = False
#: Blueprint local JSON encoder class to use.
#: Set to ``None`` to use the app's :class:`~flask.app.Flask.json_encoder`.
#: Blueprint local JSON encoder class to use. Set to ``None`` to use
#: the app's :class:`~flask.Flask.json_encoder`.
json_encoder = None
#: Blueprint local JSON decoder class to use.
#: Set to ``None`` to use the app's :class:`~flask.app.Flask.json_decoder`.
#: Blueprint local JSON decoder class to use. Set to ``None`` to use
#: the app's :class:`~flask.Flask.json_decoder`.
json_decoder = None
# TODO remove the next three attrs when Sphinx :inherited-members: works
# https://github.com/sphinx-doc/sphinx/issues/741
#: The name of the package or module that this app belongs to. Do not
#: change this once it is set by the constructor.
import_name = None
#: Location of the template files to be added to the template lookup.
#: ``None`` if templates should not be added.
template_folder = None
#: Absolute path to the package on the filesystem. Used to look up
#: resources contained in the package.
root_path = None
def __init__(
self,
name,
@ -176,8 +161,10 @@ class Blueprint(Scaffold):
self.url_prefix = url_prefix
self.subdomain = subdomain
self.deferred_functions = []
if url_defaults is None:
url_defaults = {}
self.url_values_defaults = url_defaults
self.cli_group = cli_group
@ -195,9 +182,9 @@ class Blueprint(Scaffold):
warn(
Warning(
"The blueprint was already registered once "
"but is getting modified now. These changes "
"will not show up."
"The blueprint was already registered once but is"
" getting modified now. These changes will not show"
" up."
)
)
self.deferred_functions.append(func)
@ -223,12 +210,13 @@ class Blueprint(Scaffold):
return BlueprintSetupState(self, app, options, first_registration)
def register(self, app, options, first_registration=False):
"""Called by :meth:`Flask.register_blueprint` to register all views
and callbacks registered on the blueprint with the application. Creates
a :class:`.BlueprintSetupState` and calls each :meth:`record` callback
with it.
"""Called by :meth:`Flask.register_blueprint` to register all
views and callbacks registered on the blueprint with the
application. Creates a :class:`.BlueprintSetupState` and calls
each :meth:`record` callbackwith it.
:param app: The application this blueprint is being registered with.
:param app: The application this blueprint is being registered
with.
:param options: Keyword arguments forwarded from
:meth:`~Flask.register_blueprint`.
:param first_registration: Whether this is the first time this
@ -244,44 +232,36 @@ class Blueprint(Scaffold):
endpoint="static",
)
# Merge app and self dictionaries.
def merge_dict_lists(self_dict, app_dict):
"""Merges self_dict into app_dict. Replaces None keys with self.name.
Values of dict must be lists.
"""
for key, values in self_dict.items():
key = self.name if key is None else f"{self.name}.{key}"
app_dict[key].extend(values)
# Merge blueprint data into parent.
if first_registration:
def merge_dict_nested(self_dict, app_dict):
"""Merges self_dict into app_dict. Replaces None keys with self.name.
Values of dict must be dict.
"""
for key, value in self_dict.items():
key = self.name if key is None else f"{self.name}.{key}"
app_dict[key] = value
def extend(bp_dict, parent_dict):
for key, values in bp_dict.items():
key = self.name if key is None else f"{self.name}.{key}"
parent_dict[key].extend(values)
app.view_functions.update(self.view_functions)
def update(bp_dict, parent_dict):
for key, value in bp_dict.items():
key = self.name if key is None else f"{self.name}.{key}"
parent_dict[key] = value
merge_dict_lists(self.before_request_funcs, app.before_request_funcs)
merge_dict_lists(self.after_request_funcs, app.after_request_funcs)
merge_dict_lists(self.teardown_request_funcs, app.teardown_request_funcs)
merge_dict_lists(self.url_default_functions, app.url_default_functions)
merge_dict_lists(self.url_value_preprocessors, app.url_value_preprocessors)
merge_dict_lists(
self.template_context_processors, app.template_context_processors
)
merge_dict_nested(self.error_handler_spec, app.error_handler_spec)
app.view_functions.update(self.view_functions)
extend(self.before_request_funcs, app.before_request_funcs)
extend(self.after_request_funcs, app.after_request_funcs)
extend(self.teardown_request_funcs, app.teardown_request_funcs)
extend(self.url_default_functions, app.url_default_functions)
extend(self.url_value_preprocessors, app.url_value_preprocessors)
extend(self.template_context_processors, app.template_context_processors)
update(self.error_handler_spec, app.error_handler_spec)
for deferred in self.deferred_functions:
deferred(state)
cli_resolved_group = options.get("cli_group", self.cli_group)
if not self.cli.commands:
return
cli_resolved_group = options.get("cli_group", self.cli_group)
if cli_resolved_group is None:
app.cli.commands.update(self.cli.commands)
elif cli_resolved_group is _sentinel:

303
src/flask/helpers.py
View file

@ -1,13 +1,10 @@
import os
import pkgutil
import socket
import sys
import warnings
from functools import update_wrapper
from threading import RLock
import werkzeug.utils
from jinja2 import FileSystemLoader
from werkzeug.exceptions import NotFound
from werkzeug.routing import BuildError
from werkzeug.urls import url_quote
@ -677,157 +674,6 @@ def send_from_directory(directory, path, **kwargs):
)
def get_root_path(import_name):
"""Returns the path to a package or cwd if that cannot be found. This
returns the path of a package or the folder that contains a module.
Not to be confused with the package path returned by :func:`find_package`.
"""
# 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__"):
return os.path.dirname(os.path.abspath(mod.__file__))
# Next attempt: check the loader.
loader = pkgutil.get_loader(import_name)
# 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 or import_name == "__main__":
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 filepath it might be because we are a
# namespace package. In this case we pick the root path from the
# first module that is contained in our 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))
def _matching_loader_thinks_module_is_package(loader, mod_name):
"""Given the loader that loaded a module and the module this function
attempts to figure out if the given module is actually a package.
"""
cls = type(loader)
# If the loader can tell us if something is a package, we can
# directly ask the loader.
if hasattr(loader, "is_package"):
return loader.is_package(mod_name)
# importlib's namespace loaders do not have this functionality but
# all the modules it loads are packages, so we can take advantage of
# this information.
elif cls.__module__ == "_frozen_importlib" and cls.__name__ == "NamespaceLoader":
return True
# Otherwise we need to fail with an error that explains what went
# wrong.
raise AttributeError(
f"{cls.__name__}.is_package() method is missing but is required"
" for PEP 302 import hooks."
)
def _find_package_path(root_mod_name):
"""Find the path where the module's root exists in"""
import importlib.util
try:
spec = importlib.util.find_spec(root_mod_name)
if spec is None:
raise ValueError("not found")
# ImportError: the machinery told us it does not exist
# ValueError:
# - the module name was invalid
# - the module name is __main__
# - *we* raised `ValueError` due to `spec` being `None`
except (ImportError, ValueError):
pass # handled below
else:
# namespace package
if spec.origin in {"namespace", None}:
return os.path.dirname(next(iter(spec.submodule_search_locations)))
# a package (with __init__.py)
elif spec.submodule_search_locations:
return os.path.dirname(os.path.dirname(spec.origin))
# just a normal module
else:
return os.path.dirname(spec.origin)
# we were unable to find the `package_path` using PEP 451 loaders
loader = pkgutil.get_loader(root_mod_name)
if loader is None or root_mod_name == "__main__":
# import name is not found, or interactive/main module
return os.getcwd()
else:
if hasattr(loader, "get_filename"):
filename = loader.get_filename(root_mod_name)
elif hasattr(loader, "archive"):
# zipimporter's loader.archive points to the .egg or .zip
# archive filename is dropped in call to dirname below.
filename = loader.archive
else:
# At least one loader is missing both get_filename and archive:
# Google App Engine's HardenedModulesHook
#
# Fall back to imports.
__import__(root_mod_name)
filename = sys.modules[root_mod_name].__file__
package_path = os.path.abspath(os.path.dirname(filename))
# In case the root module is a package we need to chop of the
# rightmost part. This needs to go through a helper function
# because of namespace packages.
if _matching_loader_thinks_module_is_package(loader, root_mod_name):
package_path = os.path.dirname(package_path)
return package_path
def find_package(import_name):
"""Finds a package and returns the prefix (or None if the package is
not installed) as well as the folder that contains the package or
module as a tuple. The package path returned is the module that would
have to be added to the pythonpath in order to make it possible to
import the module. The prefix is the path below which a UNIX like
folder structure exists (lib, share etc.).
"""
root_mod_name, _, _ = import_name.partition(".")
package_path = _find_package_path(root_mod_name)
site_parent, site_folder = os.path.split(package_path)
py_prefix = os.path.abspath(sys.prefix)
if package_path.startswith(py_prefix):
return py_prefix, package_path
elif site_folder.lower() == "site-packages":
parent, folder = os.path.split(site_parent)
# Windows like installations
if folder.lower() == "lib":
base_dir = parent
# UNIX like installations
elif os.path.basename(parent).lower() == "lib":
base_dir = os.path.dirname(parent)
else:
base_dir = site_parent
return base_dir, package_path
return None, package_path
class locked_cached_property:
"""A decorator that converts a function into a lazy property. The
function wrapped is called the first time to retrieve the result
@ -854,155 +700,6 @@ class locked_cached_property:
return value
class _PackageBoundObject:
#: The name of the package or module that this app belongs to. Do not
#: change this once it is set by the constructor.
import_name = None
#: Location of the template files to be added to the template lookup.
#: ``None`` if templates should not be added.
template_folder = None
#: Absolute path to the package on the filesystem. Used to look up
#: resources contained in the package.
root_path = None
def __init__(self, import_name, template_folder=None, root_path=None):
self.import_name = import_name
self.template_folder = template_folder
if root_path is None:
root_path = get_root_path(self.import_name)
self.root_path = root_path
self._static_folder = None
self._static_url_path = None
# circular import
from .cli import AppGroup
#: The Click command group for registration of CLI commands
#: on the application and associated blueprints. These commands
#: are accessible via the :command:`flask` command once the
#: application has been discovered and blueprints registered.
self.cli = AppGroup()
@property
def static_folder(self):
"""The absolute path to the configured static folder."""
if self._static_folder is not None:
return os.path.join(self.root_path, self._static_folder)
@static_folder.setter
def static_folder(self, value):
if value is not None:
value = os.fspath(value).rstrip(r"\/")
self._static_folder = value
@property
def static_url_path(self):
"""The URL prefix that the static route will be accessible from.
If it was not configured during init, it is derived from
:attr:`static_folder`.
"""
if self._static_url_path is not None:
return self._static_url_path
if self.static_folder is not None:
basename = os.path.basename(self.static_folder)
return f"/{basename}".rstrip("/")
@static_url_path.setter
def static_url_path(self, value):
if value is not None:
value = value.rstrip("/")
self._static_url_path = value
@property
def has_static_folder(self):
"""This is ``True`` if the package bound object's container has a
folder for static files.
.. versionadded:: 0.5
"""
return self.static_folder is not None
@locked_cached_property
def jinja_loader(self):
"""The Jinja loader for this package bound object.
.. versionadded:: 0.5
"""
if self.template_folder is not None:
return FileSystemLoader(os.path.join(self.root_path, self.template_folder))
def get_send_file_max_age(self, filename):
"""Used by :func:`send_file` to determine the ``max_age`` cache
value for a given file path if it wasn't passed.
By default, this returns :data:`SEND_FILE_MAX_AGE_DEFAULT` from
the configuration of :data:`~flask.current_app`. This defaults
to ``None``, which tells the browser to use conditional requests
instead of a timed cache, which is usually preferable.
.. versionchanged:: 2.0
The default configuration is ``None`` instead of 12 hours.
.. versionadded:: 0.9
"""
value = current_app.send_file_max_age_default
if value is None:
return None
return total_seconds(value)
def send_static_file(self, filename):
"""Function used internally to send static files from the static
folder to the browser.
.. versionadded:: 0.5
"""
if not self.has_static_folder:
raise RuntimeError("No static folder for this object")
# 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(self.static_folder, filename, max_age=max_age)
def open_resource(self, resource, mode="rb"):
"""Opens a resource from the application's resource folder. To see
how this works, consider the following folder structure::
/myapplication.py
/schema.sql
/static
/style.css
/templates
/layout.html
/index.html
If you want to open the :file:`schema.sql` file you would do the
following::
with app.open_resource('schema.sql') as f:
contents = f.read()
do_something_with(contents)
:param resource: the name of the resource. To access resources within
subfolders use forward slashes as separator.
:param mode: Open file in this mode. Only reading is supported,
valid values are "r" (or "rt") and "rb".
"""
if mode not in {"r", "rt", "rb"}:
raise ValueError("Resources can only be opened for reading")
return open(os.path.join(self.root_path, resource), mode)
def total_seconds(td):
"""Returns the total seconds from a timedelta object.

666
src/flask/scaffold.py
View file

@ -1,10 +1,18 @@
import importlib.util
import os
import pkgutil
import sys
from collections import defaultdict
from functools import update_wrapper
from jinja2 import FileSystemLoader
from werkzeug.exceptions import default_exceptions
from werkzeug.exceptions import HTTPException
from .helpers import _PackageBoundObject
from .cli import AppGroup
from .globals import current_app
from .helpers import locked_cached_property
from .helpers import send_from_directory
from .templating import _default_template_ctx_processor
# a singleton sentinel value for parameter defaults
@ -19,136 +27,305 @@ def setupmethod(f):
def wrapper_func(self, *args, **kwargs):
if self._is_setup_finished():
raise AssertionError(
"A setup function was called after the "
"first request was handled. This usually indicates a bug "
"in the application where a module was not imported "
"and decorators or other functionality was called too late.\n"
"To fix this make sure to import all your view modules, "
"database models and everything related at a central place "
"before the application starts serving requests."
"A setup function was called after the first request "
"was handled. This usually indicates a bug in the"
" application where a module was not imported and"
" decorators or other functionality was called too"
" late.\nTo fix this make sure to import all your view"
" modules, database models, and everything related at a"
" central place before the application starts serving"
" requests."
)
return f(self, *args, **kwargs)
return update_wrapper(wrapper_func, f)
class Scaffold(_PackageBoundObject):
"""A common base for class Flask and class Blueprint."""
class Scaffold:
"""Common behavior shared between :class:`~flask.Flask` and
:class:`~flask.blueprints.Blueprint`.
#: Skeleton local JSON decoder class to use.
#: Set to ``None`` to use the app's :class:`~flask.app.Flask.json_encoder`.
:param import_name: The import name of the module where this object
is defined. Usually :attr:`__name__` should be used.
:param static_folder: Path to a folder of static files to serve.
If this is set, a static route will be added.
:param static_url_path: URL prefix for the static route.
:param template_folder: Path to a folder containing template files.
for rendering. If this is set, a Jinja loader will be added.
:param root_path: The path that static, template, and resource files
are relative to. Typically not set, it is discovered based on
the ``import_name``.
.. versionadded:: 2.0.0
"""
name: str
_static_folder = None
_static_url_path = None
#: JSON encoder class used by :func:`flask.json.dumps`. If a
#: blueprint sets this, it will be used instead of the app's value.
json_encoder = None
#: Skeleton local JSON decoder class to use.
#: Set to ``None`` to use the app's :class:`~flask.app.Flask.json_decoder`.
#: JSON decoder class used by :func:`flask.json.loads`. If a
#: blueprint sets this, it will be used instead of the app's value.
json_decoder = None
#: The name of the package or module that this app belongs to. Do not
#: change this once it is set by the constructor.
import_name = None
#: Location of the template files to be added to the template lookup.
#: ``None`` if templates should not be added.
template_folder = None
#: Absolute path to the package on the filesystem. Used to look up
#: resources contained in the package.
root_path = None
def __init__(
self,
import_name,
static_folder="static",
static_folder=None,
static_url_path=None,
template_folder=None,
root_path=None,
):
super().__init__(
import_name=import_name,
template_folder=template_folder,
root_path=root_path,
)
#: The name of the package or module that this object belongs
#: to. Do not change this once it is set by the constructor.
self.import_name = import_name
self.static_folder = static_folder
self.static_url_path = static_url_path
#: A dictionary of all view functions registered. The keys will
#: be function names which are also used to generate URLs and
#: the values are the function objects themselves.
#: The path to the templates folder, relative to
#: :attr:`root_path`, to add to the template loader. ``None`` if
#: templates should not be added.
self.template_folder = template_folder
if root_path is None:
root_path = get_root_path(self.import_name)
#: Absolute path to the package on the filesystem. Used to look
#: up resources contained in the package.
self.root_path = root_path
#: The Click command group for registering CLI commands for this
#: object. The commands are available from the ``flask`` command
#: once the application has been discovered and blueprints have
#: been registered.
self.cli = AppGroup()
#: A dictionary mapping endpoint names to view functions.
#:
#: To register a view function, use the :meth:`route` decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.view_functions = {}
#: A dictionary of all registered error handlers. The key is ``None``
#: for error handlers active on the application, otherwise the key is
#: the name of the blueprint. Each key points to another dictionary
#: where the key is the status code of the http exception. The
#: special key ``None`` points to a list of tuples where the first item
#: is the class for the instance check and the second the error handler
#: function.
#: A data structure of registered error handlers, in the format
#: ``{scope: {code: {class: handler}}}```. The ``scope`` key is
#: the name of a blueprint the handlers are active for, or
#: ``None`` for all requests. The ``code`` key is the HTTP
#: status code for ``HTTPException``, or ``None`` for
#: other exceptions. The innermost dictionary maps exception
#: classes to handler functions.
#:
#: To register an error handler, use the :meth:`errorhandler`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.error_handler_spec = defaultdict(lambda: defaultdict(dict))
#: A dictionary with lists of functions that will be called at the
#: beginning of each request. The key of the dictionary is the name of
#: the blueprint this function is active for, or ``None`` for all
#: requests. To register a function, use the :meth:`before_request`
#: A data structure of functions to call at the beginning of
#: each request, in the format ``{scope: [functions]}``. The
#: ``scope`` key is the name of a blueprint the functions are
#: active for, or ``None`` for all requests.
#:
#: To register a function, use the :meth:`before_request`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.before_request_funcs = defaultdict(list)
#: A dictionary with lists of functions that should be called after
#: each request. The key of the dictionary is the name of the blueprint
#: this function is active for, ``None`` for all requests. This can for
#: example be used to close database connections. To register a function
#: here, use the :meth:`after_request` decorator.
#: A data structure of functions to call at the end of each
#: request, in the format ``{scope: [functions]}``. The
#: ``scope`` key is the name of a blueprint the functions are
#: active for, or ``None`` for all requests.
#:
#: To register a function, use the :meth:`after_request`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.after_request_funcs = defaultdict(list)
#: A dictionary with lists of functions that are called after
#: each request, even if an exception has occurred. The key of the
#: dictionary is the name of the blueprint this function is active for,
#: ``None`` for all requests. These functions are not allowed to modify
#: the request, and their return values are ignored. If an exception
#: occurred while processing the request, it gets passed to each
#: teardown_request function. To register a function here, use the
#: :meth:`teardown_request` decorator.
#: A data structure of functions to call at the end of each
#: request even if an exception is raised, in the format
#: ``{scope: [functions]}``. The ``scope`` key is the name of a
#: blueprint the functions are active for, or ``None`` for all
#: requests.
#:
#: .. versionadded:: 0.7
#: To register a function, use the :meth:`teardown_request`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.teardown_request_funcs = defaultdict(list)
#: A dictionary with list of functions that are called without argument
#: to populate the template context. The key of the dictionary is the
#: name of the blueprint this function is active for, ``None`` for all
#: requests. Each returns a dictionary that the template context is
#: updated with. To register a function here, use the
#: :meth:`context_processor` decorator.
#: A data structure of functions to call to pass extra context
#: values when rendering templates, in the format
#: ``{scope: [functions]}``. The ``scope`` key is the name of a
#: blueprint the functions are active for, or ``None`` for all
#: requests.
#:
#: To register a function, use the :meth:`context_processor`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.template_context_processors = defaultdict(
list, {None: [_default_template_ctx_processor]}
)
#: A dictionary with lists of functions that are called before the
#: :attr:`before_request_funcs` functions. The key of the dictionary is
#: the name of the blueprint this function is active for, or ``None``
#: for all requests. To register a function, use
#: :meth:`url_value_preprocessor`.
#: A data structure of functions to call to modify the keyword
#: arguments passed to the view function, in the format
#: ``{scope: [functions]}``. The ``scope`` key is the name of a
#: blueprint the functions are active for, or ``None`` for all
#: requests.
#:
#: .. versionadded:: 0.7
#: To register a function, use the
#: :meth:`url_value_preprocessor` decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.url_value_preprocessors = defaultdict(list)
#: A dictionary with lists of functions that can be used as URL value
#: preprocessors. The key ``None`` here is used for application wide
#: callbacks, otherwise the key is the name of the blueprint.
#: Each of these functions has the chance to modify the dictionary
#: of URL values before they are used as the keyword arguments of the
#: view function. For each function registered this one should also
#: provide a :meth:`url_defaults` function that adds the parameters
#: automatically again that were removed that way.
#: A data structure of functions to call to modify the keyword
#: arguments when generating URLs, in the format
#: ``{scope: [functions]}``. The ``scope`` key is the name of a
#: blueprint the functions are active for, or ``None`` for all
#: requests.
#:
#: .. versionadded:: 0.7
#: To register a function, use the :meth:`url_defaults`
#: decorator.
#:
#: This data structure is internal. It should not be modified
#: directly and its format may change at any time.
self.url_default_functions = defaultdict(list)
def __repr__(self):
return f"<{type(self).__name__} {self.name!r}>"
def _is_setup_finished(self):
raise NotImplementedError
@property
def static_folder(self):
"""The absolute path to the configured static folder. ``None``
if no static folder is set.
"""
if self._static_folder is not None:
return os.path.join(self.root_path, self._static_folder)
@static_folder.setter
def static_folder(self, value):
if value is not None:
value = os.fspath(value).rstrip(r"\/")
self._static_folder = value
@property
def has_static_folder(self):
"""``True`` if :attr:`static_folder` is set.
.. versionadded:: 0.5
"""
return self.static_folder is not None
@property
def static_url_path(self):
"""The URL prefix that the static route will be accessible from.
If it was not configured during init, it is derived from
:attr:`static_folder`.
"""
if self._static_url_path is not None:
return self._static_url_path
if self.static_folder is not None:
basename = os.path.basename(self.static_folder)
return f"/{basename}".rstrip("/")
@static_url_path.setter
def static_url_path(self, value):
if value is not None:
value = value.rstrip("/")
self._static_url_path = value
def get_send_file_max_age(self, filename):
"""Used by :func:`send_file` to determine the ``max_age`` cache
value for a given file path if it wasn't passed.
By default, this returns :data:`SEND_FILE_MAX_AGE_DEFAULT` from
the configuration of :data:`~flask.current_app`. This defaults
to ``None``, which tells the browser to use conditional requests
instead of a timed cache, which is usually preferable.
.. versionchanged:: 2.0
The default configuration is ``None`` instead of 12 hours.
.. versionadded:: 0.9
"""
value = current_app.send_file_max_age_default
if value is None:
return None
return int(value.total_seconds())
def send_static_file(self, filename):
"""The view function used to serve files from
:attr:`static_folder`. A route is automatically registered for
this view at :attr:`static_url_path` if :attr:`static_folder` is
set.
.. versionadded:: 0.5
"""
if not self.has_static_folder:
raise RuntimeError("'static_folder' must be set to serve static_files.")
# send_file only knows to call get_send_file_max_age on the app,
# call it here so it works for blueprints too.
max_age = self.get_send_file_max_age(filename)
return send_from_directory(self.static_folder, filename, max_age=max_age)
@locked_cached_property
def jinja_loader(self):
"""The Jinja loader for this object's templates. By default this
is a class :class:`jinja2.loaders.FileSystemLoader` to
:attr:`template_folder` if it is set.
.. versionadded:: 0.5
"""
if self.template_folder is not None:
return FileSystemLoader(os.path.join(self.root_path, self.template_folder))
def open_resource(self, resource, mode="rb"):
"""Open a resource file relative to :attr:`root_path` for
reading.
For example, if the file ``schema.sql`` is next to the file
``app.py`` where the ``Flask`` app is defined, it can be opened
with:
.. code-block:: python
with app.open_resource("schema.sql") as f:
conn.executescript(f.read())
:param resource: Path to the resource relative to
:attr:`root_path`.
:param mode: Open the file in this mode. Only reading is
supported, valid values are "r" (or "rt") and "rb".
"""
if mode not in {"r", "rt", "rb"}:
raise ValueError("Resources can only be opened for reading.")
return open(os.path.join(self.root_path, resource), mode)
def _method_route(self, method, rule, options):
if "methods" in options:
raise TypeError("Use the 'route' decorator to use the 'methods' argument.")
@ -191,28 +368,27 @@ class Scaffold(_PackageBoundObject):
return self._method_route("PATCH", rule, options)
def route(self, rule, **options):
"""A decorator that is used to register a view function for a
given URL rule. This does the same thing as :meth:`add_url_rule`
but is intended for decorator usage::
"""Decorate a view function to register it with the given URL
rule and options. Calls :meth:`add_url_rule`, which has more
details about the implementation.
@app.route('/')
.. code-block:: python
@app.route("/")
def index():
return 'Hello World'
return "Hello, World!"
For more information refer to :ref:`url-route-registrations`.
See :ref:`url-route-registrations`.
:param rule: the URL rule as string
:param endpoint: the endpoint for the registered URL rule. Flask
itself assumes the name of the view function as
endpoint
:param options: the options to be forwarded to the underlying
:class:`~werkzeug.routing.Rule` object. A change
to Werkzeug is handling of method options. methods
is a list of methods this rule should be limited
to (``GET``, ``POST`` etc.). By default a rule
just listens for ``GET`` (and implicitly ``HEAD``).
Starting with Flask 0.6, ``OPTIONS`` is implicitly
added and handled by the standard request handling.
The endpoint name for the route defaults to the name of the view
function if the ``endpoint`` parameter isn't passed.
The ``methods`` parameter defaults to ``["GET"]``. ``HEAD`` and
``OPTIONS`` are added automatically.
:param rule: The URL rule string.
:param options: Extra options passed to the
:class:`~werkzeug.routing.Rule` object.
"""
def decorator(f):
@ -231,17 +407,80 @@ class Scaffold(_PackageBoundObject):
provide_automatic_options=None,
**options,
):
"""Register a rule for routing incoming requests and building
URLs. The :meth:`route` decorator is a shortcut to call this
with the ``view_func`` argument. These are equivalent:
.. code-block:: python
@app.route("/")
def index():
...
.. code-block:: python
def index():
...
app.add_url_rule("/", view_func=index)
See :ref:`url-route-registrations`.
The endpoint name for the route defaults to the name of the view
function if the ``endpoint`` parameter isn't passed. An error
will be raised if a function has already been registered for the
endpoint.
The ``methods`` parameter defaults to ``["GET"]``. ``HEAD`` is
always added automatically, and ``OPTIONS`` is added
automatically by default.
``view_func`` does not necessarily need to be passed, but if the
rule should participate in routing an endpoint name must be
associated with a view function at some point with the
:meth:`endpoint` decorator.
.. code-block:: python
app.add_url_rule("/", endpoint="index")
@app.endpoint("index")
def index():
...
If ``view_func`` has a ``required_methods`` attribute, those
methods are added to the passed and automatic methods. If it
has a ``provide_automatic_methods`` attribute, it is used as the
default if the parameter is not passed.
:param rule: The URL rule string.
:param endpoint: The endpoint name to associate with the rule
and view function. Used when routing and building URLs.
Defaults to ``view_func.__name__``.
:param view_func: The view function to associate with the
endpoint name.
:param provide_automatic_options: Add the ``OPTIONS`` method and
respond to ``OPTIONS`` requests automatically.
:param options: Extra options passed to the
:class:`~werkzeug.routing.Rule` object.
"""
raise NotImplementedError
def endpoint(self, endpoint):
"""A decorator to register a function as an endpoint.
Example::
"""Decorate a view function to register it for the given
endpoint. Used if a rule is added without a ``view_func`` with
:meth:`add_url_rule`.
@app.endpoint('example.endpoint')
.. code-block:: python
app.add_url_rule("/ex", endpoint="example")
@app.endpoint("example")
def example():
return "example"
...
:param endpoint: the name of the endpoint
:param endpoint: The endpoint name to associate with the view
function.
"""
def decorator(f):
@ -252,28 +491,38 @@ class Scaffold(_PackageBoundObject):
@setupmethod
def before_request(self, f):
"""Registers a function to run before each request.
"""Register a function to run before each request.
For example, this can be used to open a database connection, or to load
the logged in user from the session.
For example, this can be used to open a database connection, or
to load the logged in user from the session.
The function will be called without any arguments. If it returns a
non-None value, the value is handled as if it was the return value from
the view, and further request handling is stopped.
.. code-block:: python
@app.before_request
def load_user():
if "user_id" in session:
g.user = db.session.get(session["user_id"])
The function will be called without any arguments. If it returns
a non-``None`` value, the value is handled as if it was the
return value from the view, and further request handling is
stopped.
"""
self.before_request_funcs[None].append(f)
return f
@setupmethod
def after_request(self, f):
"""Register a function to be run after each request.
"""Register a function to run after each request to this object.
Your function must take one parameter, an instance of
:attr:`response_class` and return a new response object or the
same (see :meth:`process_response`).
The function is called with the response object, and must return
a response object. This allows the functions to modify or
replace the response before it is sent.
As of Flask 0.7 this function might not be executed at the end of the
request in case an unhandled exception occurred.
If a function raises an exception, any remaining
``after_request`` functions will not be called. Therefore, this
should not be used for actions that must execute, such as to
close resources. Use :meth:`teardown_request` for that.
"""
self.after_request_funcs[None].append(f)
return f
@ -297,8 +546,8 @@ class Scaffold(_PackageBoundObject):
stack of active contexts. This becomes relevant if you are using
such constructs in tests.
Generally teardown functions must take every necessary step to avoid
that they will fail. If they do execute code that might fail they
Teardown functions must avoid raising exceptions, since they . If they
execute code that might fail they
will have to surround the execution of these code by try/except
statements and log occurring errors.
@ -381,7 +630,7 @@ class Scaffold(_PackageBoundObject):
"""
def decorator(f):
self._register_error_handler(None, code_or_exception, f)
self.register_error_handler(code_or_exception, f)
return f
return decorator
@ -394,15 +643,6 @@ class Scaffold(_PackageBoundObject):
.. versionadded:: 0.7
"""
self._register_error_handler(None, code_or_exception, f)
@setupmethod
def _register_error_handler(self, key, code_or_exception, f):
"""
:type key: None|str
:type code_or_exception: int|T<=Exception
:type f: callable
"""
if isinstance(code_or_exception, HTTPException): # old broken behavior
raise ValueError(
"Tried to register a handler for an exception instance"
@ -419,7 +659,7 @@ class Scaffold(_PackageBoundObject):
" instead."
)
self.error_handler_spec[key][code][exc_class] = f
self.error_handler_spec[None][code][exc_class] = f
@staticmethod
def _get_exc_class_and_code(exc_class_or_code):
@ -451,3 +691,171 @@ def _endpoint_from_view_func(view_func):
"""
assert view_func is not None, "expected view func if endpoint is not provided."
return view_func.__name__
def get_root_path(import_name):
"""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__"):
return os.path.dirname(os.path.abspath(mod.__file__))
# Next attempt: check the loader.
loader = pkgutil.get_loader(import_name)
# 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 or import_name == "__main__":
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))
def _matching_loader_thinks_module_is_package(loader, mod_name):
"""Attempt to figure out if the given name is a package or a module.
:param: loader: The loader that handled the name.
:param mod_name: The name of the package or module.
"""
# Use loader.is_package if it's available.
if hasattr(loader, "is_package"):
return loader.is_package(mod_name)
cls = type(loader)
# NamespaceLoader doesn't implement is_package, but all names it
# loads must be packages.
if cls.__module__ == "_frozen_importlib" and cls.__name__ == "NamespaceLoader":
return True
# Otherwise we need to fail with an error that explains what went
# wrong.
raise AttributeError(
f"'{cls.__name__}.is_package()' must be implemented for PEP 302"
f" import hooks."
)
def _find_package_path(root_mod_name):
"""Find the path that contains the package or module."""
try:
spec = importlib.util.find_spec(root_mod_name)
if spec is None:
raise ValueError("not found")
# ImportError: the machinery told us it does not exist
# ValueError:
# - the module name was invalid
# - the module name is __main__
# - *we* raised `ValueError` due to `spec` being `None`
except (ImportError, ValueError):
pass # handled below
else:
# namespace package
if spec.origin in {"namespace", None}:
return os.path.dirname(next(iter(spec.submodule_search_locations)))
# a package (with __init__.py)
elif spec.submodule_search_locations:
return os.path.dirname(os.path.dirname(spec.origin))
# just a normal module
else:
return os.path.dirname(spec.origin)
# we were unable to find the `package_path` using PEP 451 loaders
loader = pkgutil.get_loader(root_mod_name)
if loader is None or root_mod_name == "__main__":
# import name is not found, or interactive/main module
return os.getcwd()
if hasattr(loader, "get_filename"):
filename = loader.get_filename(root_mod_name)
elif hasattr(loader, "archive"):
# zipimporter's loader.archive points to the .egg or .zip file.
filename = loader.archive
else:
# At least one loader is missing both get_filename and archive:
# Google App Engine's HardenedModulesHook, use __file__.
filename = importlib.import_module(root_mod_name).__file__
package_path = os.path.abspath(os.path.dirname(filename))
# If the imported name is a package, filename is currently pointing
# to the root of the package, need to get the current directory.
if _matching_loader_thinks_module_is_package(loader, root_mod_name):
package_path = os.path.dirname(package_path)
return package_path
def find_package(import_name):
"""Find the prefix that a package is installed under, and the path
that it would be imported from.
The prefix is the directory containing the standard directory
hierarchy (lib, bin, etc.). If the package is not installed to the
system (:attr:`sys.prefix`) or a virtualenv (``site-packages``),
``None`` is returned.
The path is the entry in :attr:`sys.path` that contains the package
for import. If the package is not installed, it's assumed that the
package was imported from the current working directory.
"""
root_mod_name, _, _ = import_name.partition(".")
package_path = _find_package_path(root_mod_name)
py_prefix = os.path.abspath(sys.prefix)
# installed to the system
if package_path.startswith(py_prefix):
return py_prefix, package_path
site_parent, site_folder = os.path.split(package_path)
# installed to a virtualenv
if site_folder.lower() == "site-packages":
parent, folder = os.path.split(site_parent)
# Windows (prefix/lib/site-packages)
if folder.lower() == "lib":
return parent, package_path
# Unix (prefix/lib/pythonX.Y/site-packages)
if os.path.basename(parent).lower() == "lib":
return os.path.dirname(parent), package_path
# something else (prefix/site-packages)
return site_parent, package_path
# not installed
return None, package_path