Refactor Blueprint.register to reduce cyclomatic complexity
This commit is contained in:
VaksHSE
parent
2579ce9f18
commit
e74c39d007
1 changed files with 96 additions and 71 deletions
|
|
@ -275,43 +275,13 @@ class Blueprint(Scaffold):
|
|||
views and callbacks registered on the blueprint with the
|
||||
application. Creates a :class:`.BlueprintSetupState` and calls
|
||||
each :meth:`record` callback with it.
|
||||
|
||||
:param app: The application this blueprint is being registered
|
||||
with.
|
||||
:param options: Keyword arguments forwarded from
|
||||
:meth:`~Flask.register_blueprint`.
|
||||
|
||||
.. versionchanged:: 2.3
|
||||
Nested blueprints now correctly apply subdomains.
|
||||
|
||||
.. versionchanged:: 2.1
|
||||
Registering the same blueprint with the same name multiple
|
||||
times is an error.
|
||||
|
||||
.. 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``.
|
||||
...
|
||||
"""
|
||||
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:
|
||||
bp_desc = "this" if app.blueprints[name] is self else "a different"
|
||||
existing_at = f" '{name}'" if self_name != name else ""
|
||||
|
||||
raise ValueError(
|
||||
f"The name '{self_name}' is already registered for"
|
||||
f" {bp_desc} blueprint{existing_at}. Use 'name=' to"
|
||||
f" provide a unique name."
|
||||
)
|
||||
self._ensure_unique_name(app, self_name, name)
|
||||
|
||||
first_bp_registration = not any(bp is self for bp in app.blueprints.values())
|
||||
first_name_registration = name not in app.blueprints
|
||||
|
|
@ -320,61 +290,116 @@ class Blueprint(Scaffold):
|
|||
self._got_registered_once = True
|
||||
state = self.make_setup_state(app, options, first_bp_registration)
|
||||
|
||||
if self.has_static_folder:
|
||||
state.add_url_rule(
|
||||
f"{self.static_url_path}/<path:filename>",
|
||||
view_func=self.send_static_file, # type: ignore[attr-defined]
|
||||
endpoint="static",
|
||||
)
|
||||
self._register_static(state)
|
||||
self._merge_blueprint_funcs_if_needed(app, name, first_bp_registration, first_name_registration)
|
||||
self._run_deferred_functions(state)
|
||||
self._register_cli(app, name, options)
|
||||
self._register_nested_blueprints(app, state, name)
|
||||
|
||||
# Merge blueprint data into parent.
|
||||
def _ensure_unique_name(self, app: App, self_name: str, name: str) -> None:
|
||||
if name not in app.blueprints:
|
||||
return
|
||||
|
||||
bp_desc = "this" if app.blueprints[name] is self else "a different"
|
||||
existing_at = f" '{name}'" if self_name != name else ""
|
||||
|
||||
raise ValueError(
|
||||
f"The name '{self_name}' is already registered for"
|
||||
f" {bp_desc} blueprint{existing_at}. Use 'name=' to"
|
||||
f" provide a unique name."
|
||||
)
|
||||
|
||||
def _register_static(self, state: BlueprintSetupState) -> None:
|
||||
if not self.has_static_folder:
|
||||
return
|
||||
|
||||
state.add_url_rule(
|
||||
f"{self.static_url_path}/<path:filename>",
|
||||
view_func=self.send_static_file, # type: ignore[attr-defined]
|
||||
endpoint="static",
|
||||
)
|
||||
|
||||
def _merge_blueprint_funcs_if_needed(
|
||||
self,
|
||||
app: App,
|
||||
name: str,
|
||||
first_bp_registration: bool,
|
||||
first_name_registration: bool,
|
||||
) -> None:
|
||||
if first_bp_registration or first_name_registration:
|
||||
self._merge_blueprint_funcs(app, name)
|
||||
|
||||
def _run_deferred_functions(self, state: BlueprintSetupState) -> None:
|
||||
for deferred in self.deferred_functions:
|
||||
deferred(state)
|
||||
|
||||
def _register_cli(self, app: App, name: str, options: dict[str, t.Any]) -> None:
|
||||
if not self.cli.commands:
|
||||
return
|
||||
|
||||
cli_resolved_group = options.get("cli_group", self.cli_group)
|
||||
|
||||
if self.cli.commands:
|
||||
if cli_resolved_group is None:
|
||||
app.cli.commands.update(self.cli.commands)
|
||||
elif cli_resolved_group is _sentinel:
|
||||
self.cli.name = name
|
||||
app.cli.add_command(self.cli)
|
||||
else:
|
||||
self.cli.name = cli_resolved_group
|
||||
app.cli.add_command(self.cli)
|
||||
if cli_resolved_group is None:
|
||||
app.cli.commands.update(self.cli.commands)
|
||||
elif cli_resolved_group is _sentinel:
|
||||
self.cli.name = name
|
||||
app.cli.add_command(self.cli)
|
||||
else:
|
||||
self.cli.name = cli_resolved_group
|
||||
app.cli.add_command(self.cli)
|
||||
|
||||
def _register_nested_blueprints(
|
||||
self,
|
||||
app: App,
|
||||
state: BlueprintSetupState,
|
||||
name: str,
|
||||
) -> None:
|
||||
for blueprint, bp_options in self._blueprints:
|
||||
bp_options = bp_options.copy()
|
||||
bp_url_prefix = bp_options.get("url_prefix")
|
||||
bp_subdomain = bp_options.get("subdomain")
|
||||
merged_options = self._compute_nested_blueprint_options(
|
||||
blueprint, state, bp_options, name
|
||||
)
|
||||
blueprint.register(app, merged_options)
|
||||
|
||||
if bp_subdomain is None:
|
||||
bp_subdomain = blueprint.subdomain
|
||||
def _compute_nested_blueprint_options(
|
||||
self,
|
||||
blueprint: "Blueprint",
|
||||
state: BlueprintSetupState,
|
||||
bp_options: dict[str, t.Any],
|
||||
name: str,
|
||||
) -> dict[str, t.Any]:
|
||||
bp_options = bp_options.copy()
|
||||
bp_options["name_prefix"] = name
|
||||
|
||||
if state.subdomain is not None and bp_subdomain is not None:
|
||||
bp_options["subdomain"] = bp_subdomain + "." + state.subdomain
|
||||
elif bp_subdomain is not None:
|
||||
bp_options["subdomain"] = bp_subdomain
|
||||
elif state.subdomain is not None:
|
||||
bp_options["subdomain"] = state.subdomain
|
||||
# --- subdomain ---
|
||||
bp_subdomain = bp_options.get("subdomain")
|
||||
|
||||
if bp_url_prefix is None:
|
||||
bp_url_prefix = blueprint.url_prefix
|
||||
if bp_subdomain is None:
|
||||
bp_subdomain = blueprint.subdomain
|
||||
|
||||
if state.url_prefix is not None and bp_url_prefix is not None:
|
||||
bp_options["url_prefix"] = (
|
||||
state.url_prefix.rstrip("/") + "/" + bp_url_prefix.lstrip("/")
|
||||
)
|
||||
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
|
||||
if state.subdomain is not None and bp_subdomain is not None:
|
||||
bp_options["subdomain"] = bp_subdomain + "." + state.subdomain
|
||||
elif bp_subdomain is not None:
|
||||
bp_options["subdomain"] = bp_subdomain
|
||||
elif state.subdomain is not None:
|
||||
bp_options["subdomain"] = state.subdomain
|
||||
|
||||
# --- url_prefix ---
|
||||
bp_url_prefix = bp_options.get("url_prefix")
|
||||
|
||||
if bp_url_prefix is None:
|
||||
bp_url_prefix = blueprint.url_prefix
|
||||
|
||||
if state.url_prefix is not None and bp_url_prefix is not None:
|
||||
bp_options["url_prefix"] = (
|
||||
state.url_prefix.rstrip("/") + "/" + bp_url_prefix.lstrip("/")
|
||||
)
|
||||
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
|
||||
|
||||
return bp_options
|
||||
|
||||
bp_options["name_prefix"] = name
|
||||
blueprint.register(app, bp_options)
|
||||
|
||||
def _merge_blueprint_funcs(self, app: App, name: str) -> None:
|
||||
def extend(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue