Bugfix blueprint naming

Following discussions for Flask we've decided to name blueprints based
on how they are registered. This allows for two different blueprints
to have the same self-name as long as they are registered in different
nested positions. This helps users choose better blueprint names.
This commit is contained in:
pgjones 2021-05-18 13:33:45 +01:00
parent 99afbb277d
commit 141fde1d8e
3 changed files with 50 additions and 33 deletions

View file

@ -747,7 +747,7 @@ class Flask(Scaffold):
] = self.template_context_processors[None]
reqctx = _request_ctx_stack.top
if reqctx is not None:
for bp in self._request_blueprints():
for bp in request.blueprints:
if bp in self.template_context_processors:
funcs = chain(funcs, self.template_context_processors[bp])
orig_ctx = context.copy()
@ -1267,7 +1267,7 @@ class Flask(Scaffold):
exc_class, code = self._get_exc_class_and_code(type(e))
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]
if not handler_map:
@ -1788,9 +1788,16 @@ class Flask(Scaffold):
.. versionadded:: 0.7
"""
funcs: t.Iterable[URLDefaultCallable] = self.url_default_functions[None]
if "." in endpoint:
bp = endpoint.rsplit(".", 1)[0]
funcs = chain(funcs, self.url_default_functions[bp])
bps: t.List[str] = [endpoint.rsplit(".", 1)[0]]
while "." in bps[-1]:
bps.append(bps[-1].rpartition(".")[0])
for bp in bps:
funcs = chain(funcs, self.url_default_functions[bp])
for func in funcs:
func(endpoint, values)
@ -1831,14 +1838,14 @@ class Flask(Scaffold):
funcs: t.Iterable[URLValuePreprocessorCallable] = self.url_value_preprocessors[
None
]
for bp in self._request_blueprints():
for bp in request.blueprints:
if bp in self.url_value_preprocessors:
funcs = chain(funcs, self.url_value_preprocessors[bp])
for func in funcs:
func(request.endpoint, request.view_args)
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:
funcs = chain(funcs, self.before_request_funcs[bp])
for func in funcs:
@ -1863,7 +1870,7 @@ class Flask(Scaffold):
"""
ctx = _request_ctx_stack.top
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:
funcs = chain(funcs, reversed(self.after_request_funcs[bp]))
if None in self.after_request_funcs:
@ -1902,7 +1909,7 @@ class Flask(Scaffold):
funcs: t.Iterable[TeardownCallable] = reversed(
self.teardown_request_funcs[None]
)
for bp in self._request_blueprints():
for bp in request.blueprints:
if bp in self.teardown_request_funcs:
funcs = chain(funcs, reversed(self.teardown_request_funcs[bp]))
for func in funcs:
@ -2074,9 +2081,3 @@ class Flask(Scaffold):
wrapped to apply middleware.
"""
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

@ -98,7 +98,7 @@ class BlueprintSetupState:
defaults = dict(defaults, **options.pop("defaults"))
self.app.add_url_rule(
rule,
f"{self.name_prefix}{self.blueprint.name}.{endpoint}",
f"{self.name_prefix}.{self.blueprint.name}.{endpoint}".lstrip("."),
view_func,
defaults=defaults,
**options,
@ -266,23 +266,24 @@ class Blueprint(Scaffold):
with.
:param options: Keyword arguments forwarded from
:meth:`~Flask.register_blueprint`.
:param first_registration: Whether this is the first time this
blueprint has been registered on the application.
"""
first_registration = False
first_registration = True
if self.name in app.blueprints:
assert app.blueprints[self.name] is self, (
"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."
for blueprint in app.blueprints.values():
if blueprint is self:
first_registration = False
name_prefix = options.get("name_prefix", "")
name = f"{name_prefix}.{self.name}".lstrip(".")
if name in app.blueprints and app.blueprints[name] is not self:
raise ValueError(
f"Blueprint name '{self.name}' "
f"is already registered by {app.blueprints[self.name]}. "
"Blueprints must have unique names."
)
else:
app.blueprints[self.name] = self
first_registration = True
app.blueprints[name] = self
self._got_registered_once = True
state = self.make_setup_state(app, options, first_registration)
@ -298,12 +299,11 @@ class Blueprint(Scaffold):
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}"
key = name if key is None else f"{name}.{key}"
parent_dict[key].extend(values)
for key, value in self.error_handler_spec.items():
key = self.name if key is None else f"{self.name}.{key}"
key = name if key is None else f"{name}.{key}"
value = defaultdict(
dict,
{
@ -337,7 +337,7 @@ class Blueprint(Scaffold):
if cli_resolved_group is None:
app.cli.commands.update(self.cli.commands)
elif cli_resolved_group is _sentinel:
self.cli.name = self.name
self.cli.name = name
app.cli.add_command(self.cli)
else:
self.cli.name = cli_resolved_group
@ -359,7 +359,7 @@ class Blueprint(Scaffold):
elif state.url_prefix is not None:
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)
def add_url_rule(

View file

@ -1,6 +1,7 @@
import typing as t
from werkzeug.exceptions import BadRequest
from werkzeug.utils import cached_property
from werkzeug.wrappers import Request as RequestBase
from werkzeug.wrappers import Response as ResponseBase
@ -77,6 +78,21 @@ class Request(RequestBase):
else:
return None
@cached_property
def blueprints(self) -> t.List[str]:
"""The names of the current blueprint upwards through parent
blueprints.
"""
if self.blueprint is None:
return []
bps: t.List[str] = [self.blueprint]
while "." in bps[-1]:
bps.append(bps[-1].rpartition(".")[0])
return bps
def _load_form_data(self) -> None:
RequestBase._load_form_data(self)