Merge pull request #4074 from pgjones/bp

blueprints are registered with nested names, can change registered name
This commit is contained in:
David Lord 2021-05-21 07:09:02 -07:00 committed by GitHub
commit 255461d895
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 205 additions and 83 deletions

View file

@ -27,6 +27,14 @@ Unreleased
removed early. :issue:`4078` removed early. :issue:`4078`
- Improve typing for some functions using ``Callable`` in their type - Improve typing for some functions using ``Callable`` in their type
signatures, focusing on decorator factories. :issue:`4060` signatures, focusing on decorator factories. :issue:`4060`
- Nested blueprints are registered with their dotted name. This allows
different blueprints with the same name to be nested at different
locations. :issue:`4069`
- ``register_blueprint`` takes a ``name`` option to change the
(pre-dotted) name the blueprint is registered with. This allows the
same blueprint to be registered multiple times with unique names for
``url_for``. Registering the same blueprint with the same name
multiple times is deprecated. :issue:`1091`
Version 2.0.0 Version 2.0.0

View file

@ -36,6 +36,7 @@ from .globals import _request_ctx_stack
from .globals import g from .globals import g
from .globals import request from .globals import request
from .globals import session from .globals import session
from .helpers import _split_blueprint_path
from .helpers import get_debug_flag from .helpers import get_debug_flag
from .helpers import get_env from .helpers import get_env
from .helpers import get_flashed_messages from .helpers import get_flashed_messages
@ -747,7 +748,7 @@ class Flask(Scaffold):
] = self.template_context_processors[None] ] = self.template_context_processors[None]
reqctx = _request_ctx_stack.top reqctx = _request_ctx_stack.top
if reqctx is not None: if reqctx is not None:
for bp in self._request_blueprints(): for bp in request.blueprints:
if bp in self.template_context_processors: if bp in self.template_context_processors:
funcs = chain(funcs, self.template_context_processors[bp]) funcs = chain(funcs, self.template_context_processors[bp])
orig_ctx = context.copy() orig_ctx = context.copy()
@ -1018,6 +1019,12 @@ class Flask(Scaffold):
:class:`~flask.blueprints.BlueprintSetupState`. They can be :class:`~flask.blueprints.BlueprintSetupState`. They can be
accessed in :meth:`~flask.Blueprint.record` callbacks. accessed in :meth:`~flask.Blueprint.record` callbacks.
.. versionchanged:: 2.0.1
The ``name`` option can be used to change the (pre-dotted)
name the blueprint is registered with. This allows the same
blueprint to be registered multiple times with unique names
for ``url_for``.
.. versionadded:: 0.7 .. versionadded:: 0.7
""" """
blueprint.register(self, options) blueprint.register(self, options)
@ -1267,7 +1274,7 @@ class Flask(Scaffold):
exc_class, code = self._get_exc_class_and_code(type(e)) exc_class, code = self._get_exc_class_and_code(type(e))
for c in [code, None]: for c in [code, None]:
for name in chain(self._request_blueprints(), [None]): for name in chain(request.blueprints, [None]):
handler_map = self.error_handler_spec[name][c] handler_map = self.error_handler_spec[name][c]
if not handler_map: if not handler_map:
@ -1788,9 +1795,14 @@ class Flask(Scaffold):
.. versionadded:: 0.7 .. versionadded:: 0.7
""" """
funcs: t.Iterable[URLDefaultCallable] = self.url_default_functions[None] funcs: t.Iterable[URLDefaultCallable] = self.url_default_functions[None]
if "." in endpoint: if "." in endpoint:
bp = endpoint.rsplit(".", 1)[0] # This is called by url_for, which can be called outside a
funcs = chain(funcs, self.url_default_functions[bp]) # request, can't use request.blueprints.
bps = _split_blueprint_path(endpoint.rpartition(".")[0])
bp_funcs = chain.from_iterable(self.url_default_functions[bp] for bp in bps)
funcs = chain(funcs, bp_funcs)
for func in funcs: for func in funcs:
func(endpoint, values) func(endpoint, values)
@ -1831,14 +1843,14 @@ class Flask(Scaffold):
funcs: t.Iterable[URLValuePreprocessorCallable] = self.url_value_preprocessors[ funcs: t.Iterable[URLValuePreprocessorCallable] = self.url_value_preprocessors[
None None
] ]
for bp in self._request_blueprints(): for bp in request.blueprints:
if bp in self.url_value_preprocessors: if bp in self.url_value_preprocessors:
funcs = chain(funcs, self.url_value_preprocessors[bp]) funcs = chain(funcs, self.url_value_preprocessors[bp])
for func in funcs: for func in funcs:
func(request.endpoint, request.view_args) func(request.endpoint, request.view_args)
funcs: t.Iterable[BeforeRequestCallable] = self.before_request_funcs[None] funcs: t.Iterable[BeforeRequestCallable] = self.before_request_funcs[None]
for bp in self._request_blueprints(): for bp in request.blueprints:
if bp in self.before_request_funcs: if bp in self.before_request_funcs:
funcs = chain(funcs, self.before_request_funcs[bp]) funcs = chain(funcs, self.before_request_funcs[bp])
for func in funcs: for func in funcs:
@ -1863,7 +1875,7 @@ class Flask(Scaffold):
""" """
ctx = _request_ctx_stack.top ctx = _request_ctx_stack.top
funcs: t.Iterable[AfterRequestCallable] = ctx._after_request_functions funcs: t.Iterable[AfterRequestCallable] = ctx._after_request_functions
for bp in self._request_blueprints(): for bp in request.blueprints:
if bp in self.after_request_funcs: if bp in self.after_request_funcs:
funcs = chain(funcs, reversed(self.after_request_funcs[bp])) funcs = chain(funcs, reversed(self.after_request_funcs[bp]))
if None in self.after_request_funcs: if None in self.after_request_funcs:
@ -1902,7 +1914,7 @@ class Flask(Scaffold):
funcs: t.Iterable[TeardownCallable] = reversed( funcs: t.Iterable[TeardownCallable] = reversed(
self.teardown_request_funcs[None] self.teardown_request_funcs[None]
) )
for bp in self._request_blueprints(): for bp in request.blueprints:
if bp in self.teardown_request_funcs: if bp in self.teardown_request_funcs:
funcs = chain(funcs, reversed(self.teardown_request_funcs[bp])) funcs = chain(funcs, reversed(self.teardown_request_funcs[bp]))
for func in funcs: for func in funcs:
@ -2074,9 +2086,3 @@ class Flask(Scaffold):
wrapped to apply middleware. wrapped to apply middleware.
""" """
return self.wsgi_app(environ, start_response) return self.wsgi_app(environ, start_response)
def _request_blueprints(self) -> t.Iterable[str]:
if _request_ctx_stack.top.request.blueprint is None:
return []
else:
return reversed(_request_ctx_stack.top.request.blueprint.split("."))

View file

@ -67,6 +67,7 @@ class BlueprintSetupState:
#: blueprint. #: blueprint.
self.url_prefix = url_prefix self.url_prefix = url_prefix
self.name = self.options.get("name", blueprint.name)
self.name_prefix = self.options.get("name_prefix", "") self.name_prefix = self.options.get("name_prefix", "")
#: A dictionary with URL defaults that is added to each and every #: A dictionary with URL defaults that is added to each and every
@ -96,9 +97,10 @@ class BlueprintSetupState:
defaults = self.url_defaults defaults = self.url_defaults
if "defaults" in options: if "defaults" in options:
defaults = dict(defaults, **options.pop("defaults")) defaults = dict(defaults, **options.pop("defaults"))
self.app.add_url_rule( self.app.add_url_rule(
rule, rule,
f"{self.name_prefix}{self.blueprint.name}.{endpoint}", f"{self.name_prefix}.{self.name}.{endpoint}".lstrip("."),
view_func, view_func,
defaults=defaults, defaults=defaults,
**options, **options,
@ -252,8 +254,16 @@ class Blueprint(Scaffold):
arguments passed to this method will override the defaults set arguments passed to this method will override the defaults set
on the blueprint. on the blueprint.
.. versionchanged:: 2.0.1
The ``name`` option can be used to change the (pre-dotted)
name the blueprint is registered with. This allows the same
blueprint to be registered multiple times with unique names
for ``url_for``.
.. versionadded:: 2.0 .. versionadded:: 2.0
""" """
if blueprint is self:
raise ValueError("Cannot register a blueprint on itself")
self._blueprints.append((blueprint, options)) self._blueprints.append((blueprint, options))
def register(self, app: "Flask", options: dict) -> None: def register(self, app: "Flask", options: dict) -> None:
@ -266,23 +276,48 @@ class Blueprint(Scaffold):
with. with.
:param options: Keyword arguments forwarded from :param options: Keyword arguments forwarded from
:meth:`~Flask.register_blueprint`. :meth:`~Flask.register_blueprint`.
:param first_registration: Whether this is the first time this
blueprint has been registered on the application. .. versionchanged:: 2.0.1
Nested blueprints are registered with their dotted name.
This allows different blueprints with the same name to be
nested at different locations.
.. versionchanged:: 2.0.1
The ``name`` option can be used to change the (pre-dotted)
name the blueprint is registered with. This allows the same
blueprint to be registered multiple times with unique names
for ``url_for``.
.. versionchanged:: 2.0.1
Registering the same blueprint with the same name multiple
times is deprecated and will become an error in Flask 2.1.
""" """
first_registration = False first_registration = not any(bp is self for bp in app.blueprints.values())
name_prefix = options.get("name_prefix", "")
self_name = options.get("name", self.name)
name = f"{name_prefix}.{self_name}".lstrip(".")
if self.name in app.blueprints: if name in app.blueprints:
assert app.blueprints[self.name] is self, ( existing_at = f" '{name}'" if self_name != name else ""
"A name collision occurred between blueprints"
f" {self!r} and {app.blueprints[self.name]!r}."
f" Both share the same name {self.name!r}."
f" Blueprints that are created on the fly need unique"
f" names."
)
else:
app.blueprints[self.name] = self
first_registration = True
if app.blueprints[name] is not self:
raise ValueError(
f"The name '{self_name}' is already registered for"
f" a different blueprint{existing_at}. Use 'name='"
" to provide a unique name."
)
else:
import warnings
warnings.warn(
f"The name '{self_name}' is already registered for"
f" this blueprint{existing_at}. Use 'name=' to"
" provide a unique name. This will become an error"
" in Flask 2.1.",
stacklevel=4,
)
app.blueprints[name] = self
self._got_registered_once = True self._got_registered_once = True
state = self.make_setup_state(app, options, first_registration) state = self.make_setup_state(app, options, first_registration)
@ -298,12 +333,11 @@ class Blueprint(Scaffold):
def extend(bp_dict, parent_dict): def extend(bp_dict, parent_dict):
for key, values in bp_dict.items(): for key, values in bp_dict.items():
key = self.name if key is None else f"{self.name}.{key}" key = name if key is None else f"{name}.{key}"
parent_dict[key].extend(values) parent_dict[key].extend(values)
for key, value in self.error_handler_spec.items(): for key, value in self.error_handler_spec.items():
key = self.name if key is None else f"{self.name}.{key}" key = name if key is None else f"{name}.{key}"
value = defaultdict( value = defaultdict(
dict, dict,
{ {
@ -337,7 +371,7 @@ class Blueprint(Scaffold):
if cli_resolved_group is None: if cli_resolved_group is None:
app.cli.commands.update(self.cli.commands) app.cli.commands.update(self.cli.commands)
elif cli_resolved_group is _sentinel: elif cli_resolved_group is _sentinel:
self.cli.name = self.name self.cli.name = name
app.cli.add_command(self.cli) app.cli.add_command(self.cli)
else: else:
self.cli.name = cli_resolved_group self.cli.name = cli_resolved_group
@ -354,10 +388,12 @@ class Blueprint(Scaffold):
bp_options["url_prefix"] = ( bp_options["url_prefix"] = (
state.url_prefix.rstrip("/") + "/" + bp_url_prefix.lstrip("/") state.url_prefix.rstrip("/") + "/" + bp_url_prefix.lstrip("/")
) )
else: elif bp_url_prefix is not None:
bp_options["url_prefix"] = bp_url_prefix
elif state.url_prefix is not None:
bp_options["url_prefix"] = state.url_prefix bp_options["url_prefix"] = state.url_prefix
bp_options["name_prefix"] = options.get("name_prefix", "") + self.name + "." bp_options["name_prefix"] = name
blueprint.register(app, bp_options) blueprint.register(app, bp_options)
def add_url_rule( def add_url_rule(

View file

@ -6,6 +6,7 @@ import typing as t
import warnings import warnings
from datetime import datetime from datetime import datetime
from datetime import timedelta from datetime import timedelta
from functools import lru_cache
from functools import update_wrapper from functools import update_wrapper
from threading import RLock from threading import RLock
@ -821,3 +822,13 @@ def is_ip(value: str) -> bool:
return True return True
return False return False
@lru_cache(maxsize=None)
def _split_blueprint_path(name: str) -> t.List[str]:
out: t.List[str] = [name]
if "." in name:
out.extend(_split_blueprint_path(name.rpartition(".")[0]))
return out

View file

@ -6,6 +6,7 @@ from werkzeug.wrappers import Response as ResponseBase
from . import json from . import json
from .globals import current_app from .globals import current_app
from .helpers import _split_blueprint_path
if t.TYPE_CHECKING: if t.TYPE_CHECKING:
import typing_extensions as te import typing_extensions as te
@ -59,23 +60,54 @@ class Request(RequestBase):
@property @property
def endpoint(self) -> t.Optional[str]: def endpoint(self) -> t.Optional[str]:
"""The endpoint that matched the request. This in combination with """The endpoint that matched the request URL.
:attr:`view_args` can be used to reconstruct the same or a
modified URL. If an exception happened when matching, this will This will be ``None`` if matching failed or has not been
be ``None``. performed yet.
This in combination with :attr:`view_args` can be used to
reconstruct the same URL or a modified URL.
""" """
if self.url_rule is not None: if self.url_rule is not None:
return self.url_rule.endpoint return self.url_rule.endpoint
else:
return None return None
@property @property
def blueprint(self) -> t.Optional[str]: def blueprint(self) -> t.Optional[str]:
"""The name of the current blueprint""" """The registered name of the current blueprint.
if self.url_rule and "." in self.url_rule.endpoint:
return self.url_rule.endpoint.rsplit(".", 1)[0] This will be ``None`` if the endpoint is not part of a
else: blueprint, or if URL matching failed or has not been performed
return None yet.
This does not necessarily match the name the blueprint was
created with. It may have been nested, or registered with a
different name.
"""
endpoint = self.endpoint
if endpoint is not None and "." in endpoint:
return endpoint.rpartition(".")[0]
return None
@property
def blueprints(self) -> t.List[str]:
"""The registered names of the current blueprint upwards through
parent blueprints.
This will be an empty list if there is no current blueprint, or
if URL matching failed.
.. versionadded:: 2.0.1
"""
name = self.blueprint
if name is None:
return []
return _split_blueprint_path(name)
def _load_form_data(self) -> None: def _load_form_data(self) -> None:
RequestBase._load_form_data(self) RequestBase._load_form_data(self)

View file

@ -140,7 +140,7 @@ def test_blueprint_url_defaults(app, client):
return str(bar) return str(bar)
app.register_blueprint(bp, url_prefix="/1", url_defaults={"bar": 23}) app.register_blueprint(bp, url_prefix="/1", url_defaults={"bar": 23})
app.register_blueprint(bp, url_prefix="/2", url_defaults={"bar": 19}) app.register_blueprint(bp, name="test2", url_prefix="/2", url_defaults={"bar": 19})
assert client.get("/1/foo").data == b"23/42" assert client.get("/1/foo").data == b"23/42"
assert client.get("/2/foo").data == b"19/42" assert client.get("/2/foo").data == b"19/42"
@ -837,48 +837,77 @@ def test_nested_blueprint(app, client):
assert client.get("/parent/child/grandchild/no").data == b"Grandchild no" assert client.get("/parent/child/grandchild/no").data == b"Grandchild no"
def test_nested_blueprint_url_prefix(app, client): @pytest.mark.parametrize(
parent = flask.Blueprint("parent", __name__, url_prefix="/parent") "parent_init, child_init, parent_registration, child_registration",
child = flask.Blueprint("child", __name__, url_prefix="/child") [
grandchild = flask.Blueprint("grandchild", __name__, url_prefix="/grandchild") ("/parent", "/child", None, None),
apple = flask.Blueprint("apple", __name__, url_prefix="/apple") ("/parent", None, None, "/child"),
(None, None, "/parent", "/child"),
@parent.route("/") ("/other", "/something", "/parent", "/child"),
def parent_index(): ],
return "Parent" )
def test_nesting_url_prefixes(
parent_init,
child_init,
parent_registration,
child_registration,
app,
client,
) -> None:
parent = flask.Blueprint("parent", __name__, url_prefix=parent_init)
child = flask.Blueprint("child", __name__, url_prefix=child_init)
@child.route("/") @child.route("/")
def child_index(): def index():
return "Child" return "index"
@grandchild.route("/") parent.register_blueprint(child, url_prefix=child_registration)
def grandchild_index(): app.register_blueprint(parent, url_prefix=parent_registration)
return "Grandchild"
@apple.route("/") response = client.get("/parent/child/")
def apple_index(): assert response.status_code == 200
return "Apple"
child.register_blueprint(grandchild)
child.register_blueprint(apple, url_prefix="/orange") # test overwrite
parent.register_blueprint(child)
app.register_blueprint(parent)
assert client.get("/parent/").data == b"Parent"
assert client.get("/parent/child/").data == b"Child"
assert client.get("/parent/child/grandchild/").data == b"Grandchild"
assert client.get("/parent/child/orange/").data == b"Apple"
def test_nested_blueprint_url_prefix_only_parent_prefix(app, client): def test_unique_blueprint_names(app, client) -> None:
parent = flask.Blueprint("parent", __name__) bp = flask.Blueprint("bp", __name__)
child = flask.Blueprint("child", __name__) bp2 = flask.Blueprint("bp", __name__)
@child.route("/child-endpoint") app.register_blueprint(bp)
def child_index():
return "Child"
parent.register_blueprint(child) with pytest.warns(UserWarning):
app.register_blueprint(parent, url_prefix="/parent") app.register_blueprint(bp) # same bp, same name, warning
assert client.get("/parent/child-endpoint").data == b"Child" app.register_blueprint(bp, name="again") # same bp, different name, ok
with pytest.raises(ValueError):
app.register_blueprint(bp2) # different bp, same name, error
app.register_blueprint(bp2, name="alt") # different bp, different name, ok
def test_self_registration(app, client) -> None:
bp = flask.Blueprint("bp", __name__)
with pytest.raises(ValueError):
bp.register_blueprint(bp)
def test_blueprint_renaming(app, client) -> None:
bp = flask.Blueprint("bp", __name__)
bp2 = flask.Blueprint("bp2", __name__)
@bp.get("/")
def index():
return flask.request.endpoint
@bp2.get("/")
def index2():
return flask.request.endpoint
bp.register_blueprint(bp2, url_prefix="/a", name="sub")
app.register_blueprint(bp, url_prefix="/a")
app.register_blueprint(bp, url_prefix="/b", name="alt")
assert client.get("/a/").data == b"bp.index"
assert client.get("/b/").data == b"alt.index"
assert client.get("/a/a/").data == b"bp.sub.index2"
assert client.get("/b/a/").data == b"alt.sub.index2"