Merge pull request #4074 from pgjones/bp

blueprints are registered with nested names, can change registered name

Author: davidism

CHANGES.rst

@@ -27,6 +27,14 @@ Unreleased
     removed early. :issue:`4078`
 -   Improve typing for some functions using ``Callable`` in their type
     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

src/flask/app.py

@@ -36,6 +36,7 @@
 from .globals import g
 from .globals import request
 from .globals import session
+from .helpers import _split_blueprint_path
 from .helpers import get_debug_flag
 from .helpers import get_env
 from .helpers import get_flashed_messages
@@ -747,7 +748,7 @@ def update_template_context(self, context: dict) -> None:
         ] = 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()
@@ -1018,6 +1019,12 @@ def register_blueprint(self, blueprint: "Blueprint", **options: t.Any) -> None:
             :class:`~flask.blueprints.BlueprintSetupState`. They can be
             accessed in :meth:`~flask.Blueprint.record` callbacks.
 
+        .. versionchanged:: 2.0.1
+            The ``name`` option can be used to change the (pre-dotted)
+            name the blueprint is registered with. This allows the same
+            blueprint to be registered multiple times with unique names
+            for ``url_for``.
+
         .. versionadded:: 0.7
         """
         blueprint.register(self, options)
@@ -1267,7 +1274,7 @@ def _find_error_handler(self, e: Exception) -> t.Optional[ErrorHandlerCallable]:
         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 +1795,14 @@ def inject_url_defaults(self, endpoint: str, values: dict) -> None:
         .. 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])
+            # This is called by url_for, which can be called outside a
+            # 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:
             func(endpoint, values)
 
@@ -1831,14 +1843,14 @@ def preprocess_request(self) -> t.Optional[ResponseReturnValue]:
         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 +1875,7 @@ def process_response(self, response: Response) -> Response:
         """
         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 +1914,7 @@ def do_teardown_request(
         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 +2086,3 @@ def __call__(self, environ: dict, start_response: t.Callable) -> t.Any:
         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("."))

src/flask/blueprints.py

@@ -67,6 +67,7 @@ def __init__(
         #: blueprint.
         self.url_prefix = url_prefix
 
+        self.name = self.options.get("name", blueprint.name)
         self.name_prefix = self.options.get("name_prefix", "")
 
         #: A dictionary with URL defaults that is added to each and every
@@ -96,9 +97,10 @@ def add_url_rule(
         defaults = self.url_defaults
         if "defaults" in options:
             defaults = dict(defaults, **options.pop("defaults"))
+
         self.app.add_url_rule(
             rule,
-            f"{self.name_prefix}{self.blueprint.name}.{endpoint}",
+            f"{self.name_prefix}.{self.name}.{endpoint}".lstrip("."),
             view_func,
             defaults=defaults,
             **options,
@@ -252,8 +254,16 @@ def register_blueprint(self, blueprint: "Blueprint", **options: t.Any) -> None:
         arguments passed to this method will override the defaults set
         on the blueprint.
 
+        .. versionchanged:: 2.0.1
+            The ``name`` option can be used to change the (pre-dotted)
+            name the blueprint is registered with. This allows the same
+            blueprint to be registered multiple times with unique names
+            for ``url_for``.
+
         .. versionadded:: 2.0
         """
+        if blueprint is self:
+            raise ValueError("Cannot register a blueprint on itself")
         self._blueprints.append((blueprint, options))
 
     def register(self, app: "Flask", options: dict) -> None:
@@ -266,23 +276,48 @@ def register(self, app: "Flask", options: dict) -> None:
             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.
+
+        .. 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
-
-        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."
-            )
-        else:
-            app.blueprints[self.name] = self
-            first_registration = True
+        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 name in app.blueprints:
+            existing_at = f" '{name}'" if self_name != name else ""
+
+            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
         state = self.make_setup_state(app, options, first_registration)
 
@@ -298,12 +333,11 @@ def register(self, app: "Flask", options: dict) -> None:
 
             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 +371,7 @@ def extend(bp_dict, parent_dict):
             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
@@ -354,10 +388,12 @@ def extend(bp_dict, parent_dict):
                 bp_options["url_prefix"] = (
                     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["name_prefix"] = options.get("name_prefix", "") + self.name + "."
+            bp_options["name_prefix"] = name
             blueprint.register(app, bp_options)
 
     def add_url_rule(

src/flask/helpers.py

@@ -6,6 +6,7 @@
 import warnings
 from datetime import datetime
 from datetime import timedelta
+from functools import lru_cache
 from functools import update_wrapper
 from threading import RLock
 
@@ -821,3 +822,13 @@ def is_ip(value: str) -> bool:
             return True
 
     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

src/flask/wrappers.py

@@ -6,6 +6,7 @@
 
 from . import json
 from .globals import current_app
+from .helpers import _split_blueprint_path
 
 if t.TYPE_CHECKING:
     import typing_extensions as te
@@ -59,23 +60,54 @@ def max_content_length(self) -> t.Optional[int]:  # type: ignore
 
     @property
     def endpoint(self) -> t.Optional[str]:
-        """The endpoint that matched the request.  This in combination with
-        :attr:`view_args` can be used to reconstruct the same or a
-        modified URL.  If an exception happened when matching, this will
-        be ``None``.
+        """The endpoint that matched the request URL.
+
+        This will be ``None`` if matching failed or has not been
+        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:
             return self.url_rule.endpoint
-        else:
-            return None
+
+        return None
 
     @property
     def blueprint(self) -> t.Optional[str]:
-        """The name of the current blueprint"""
-        if self.url_rule and "." in self.url_rule.endpoint:
-            return self.url_rule.endpoint.rsplit(".", 1)[0]
-        else:
-            return None
+        """The registered name of the current blueprint.
+
+        This will be ``None`` if the endpoint is not part of a
+        blueprint, or if URL matching failed or has not been performed
+        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:
         RequestBase._load_form_data(self)