Source code
from __future__ import annotations
import os
import typing as t
from collections import defaultdict
from functools import update_wrapper
from .. import typing as ft
from .scaffold import _endpoint_from_view_func
from .scaffold import _sentinel
from .scaffold import Scaffold
from .scaffold import setupmethod
class Blueprint(Scaffold):
_got_registered_once = False
def __init__(
self,
name: str,
import_name: str,
static_folder: str | os.PathLike[str] | None = None,
static_url_path: str | None = None,
template_folder: str | os.PathLike[str] | None = None,
url_prefix: str | None = None,
subdomain: str | None = None,
url_defaults: dict[str, t.Any] | None = None,
root_path: str | None = None,
cli_group: str | None = _sentinel,
):
super().__init__(
import_name=import_name,
static_folder=static_folder,
static_url_path=static_url_path,
template_folder=template_folder,
root_path=root_path,
)
if not name:
raise ValueError("'name' may not be empty.")
if "." in name:
raise ValueError("'name' may not contain a dot '.' character.")
self.name = name
self.url_prefix = url_prefix
self.subdomain = subdomain
self.deferred_functions: list[DeferredSetupFunction] = []
if url_defaults is None:
url_defaults = {}
self.url_values_defaults = url_defaults
self.cli_group = cli_group
self._blueprints: list[tuple[Blueprint, dict[str, t.Any]]] = []
def _check_setup_finished(self, f_name: str) -> None:
if self._got_registered_once:
raise AssertionError(
f"The setup method '{f_name}' can no longer be called on the blueprint"
f" '{self.name}'. It has already been registered at least once, any"
" changes will not be applied consistently.\n"
"Make sure all imports, decorators, functions, etc. needed to set up"
" the blueprint are done before registering it."
)
@setupmethod
def record(self, func: DeferredSetupFunction) -> None:
self.deferred_functions.append(func)
@setupmethod
def record_once(self, func: DeferredSetupFunction) -> None:
def wrapper(state: BlueprintSetupState) -> None:
if state.first_registration:
func(state)
self.record(update_wrapper(wrapper, func))
def make_setup_state(
self, app: App, options: dict[str, t.Any], first_registration: bool = False
) -> BlueprintSetupState:
return BlueprintSetupState(self, app, options, first_registration)
@setupmethod
def register_blueprint(self, blueprint: Blueprint, **options: t.Any) -> None:
if blueprint is self:
raise ValueError("Cannot register a blueprint on itself")
self._blueprints.append((blueprint, options))
def register(self, app: App, options: dict[str, t.Any]) -> None:
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."
)
first_bp_registration = not any(bp is self for bp in app.blueprints.values())
first_name_registration = name not in app.blueprints
app.blueprints[name] = self
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,
endpoint="static",
)
if first_bp_registration or first_name_registration:
self._merge_blueprint_funcs(app, name)
for deferred in self.deferred_functions:
deferred(state)
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)
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")
if bp_subdomain is None:
bp_subdomain = blueprint.subdomain
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
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
bp_options["name_prefix"] = name
blueprint.register(app, bp_options)
def _merge_blueprint_funcs(self, app: App, name: str) -> None:
def extend(
bp_dict: dict[ft.AppOrBlueprintKey, list[t.Any]],
parent_dict: dict[ft.AppOrBlueprintKey, list[t.Any]],
) -> None:
for key, values in bp_dict.items():
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 = name if key is None else f"{name}.{key}"
value = defaultdict(
dict,
{
code: {exc_class: func for exc_class, func in code_values.items()}
for code, code_values in value.items()
},
)
app.error_handler_spec[key] = value
for endpoint, func in self.view_functions.items():
app.view_functions[endpoint] = func
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)
@setupmethod
def add_url_rule(
self,
rule: str,
endpoint: str | None = None,
view_func: ft.RouteCallable | None = None,
provide_automatic_options: bool | None = None,
**options: t.Any,
) -> None:
if endpoint and "." in endpoint:
raise ValueError("'endpoint' may not contain a dot '.' character.")
if view_func and hasattr(view_func, "__name__") and "." in view_func.__name__:
raise ValueError("'view_func' name may not contain a dot '.' character.")
self.record(
lambda s: s.add_url_rule(
rule,
endpoint,
view_func,
provide_automatic_options=provide_automatic_options,
**options,
)
)
@t.overload
def app_template_filter(self, name: T_template_filter) -> T_template_filter: ...
@t.overload
def app_template_filter(
self, name: str | None = None
) -> t.Callable[[T_template_filter], T_template_filter]: ...
@setupmethod
def app_template_filter(
self, name: T_template_filter | str | None = None
) -> T_template_filter | t.Callable[[T_template_filter], T_template_filter]:
if callable(name):
self.add_app_template_filter(name)
return name
def decorator(f: T_template_filter) -> T_template_filter:
self.add_app_template_filter(f, name=name)
return f
return decorator
@setupmethod
def add_app_template_filter(
self, f: ft.TemplateFilterCallable, name: str | None = None
) -> None:
def register_template_filter(state: BlueprintSetupState) -> None:
state.app.add_template_filter(f, name=name)
self.record_once(register_template_filter)
@t.overload
def app_template_test(self, name: T_template_test) -> T_template_test: ...
@t.overload
def app_template_test(
self, name: str | None = None
) -> t.Callable[[T_template_test], T_template_test]: ...
@setupmethod
def app_template_test(
self, name: T_template_test | str | None = None
) -> T_template_test | t.Callable[[T_template_test], T_template_test]:
if callable(name):
self.add_app_template_test(name)
return name
def decorator(f: T_template_test) -> T_template_test:
self.add_app_template_test(f, name=name)
return f
return decorator
@setupmethod
def add_app_template_test(
self, f: ft.TemplateTestCallable, name: str | None = None
) -> None:
def register_template_test(state: BlueprintSetupState) -> None:
state.app.add_template_test(f, name=name)
self.record_once(register_template_test)
@t.overload
def app_template_global(self, name: T_template_global) -> T_template_global: ...
@t.overload
def app_template_global(
self, name: str | None = None
) -> t.Callable[[T_template_global], T_template_global]: ...
@setupmethod
def app_template_global(
self, name: T_template_global | str | None = None
) -> T_template_global | t.Callable[[T_template_global], T_template_global]:
if callable(name):
self.add_app_template_global(name)
return name
def decorator(f: T_template_global) -> T_template_global:
self.add_app_template_global(f, name=name)
return f
return decorator
@setupmethod
def add_app_template_global(
self, f: ft.TemplateGlobalCallable, name: str | None = None
) -> None:
def register_template_global(state: BlueprintSetupState) -> None:
state.app.add_template_global(f, name=name)
self.record_once(register_template_global)
@setupmethod
def before_app_request(self, f: T_before_request) -> T_before_request:
self.record_once(
lambda s: s.app.before_request_funcs.setdefault(None, []).append(f)
)
return f
@setupmethod
def after_app_request(self, f: T_after_request) -> T_after_request:
self.record_once(
lambda s: s.app.after_request_funcs.setdefault(None, []).append(f)
)
return f
@setupmethod
def teardown_app_request(self, f: T_teardown) -> T_teardown:
self.record_once(
lambda s: s.app.teardown_request_funcs.setdefault(None, []).append(f)
)
return f
@setupmethod
def app_context_processor(
self, f: T_template_context_processor
) -> T_template_context_processor:
self.record_once(
lambda s: s.app.template_context_processors.setdefault(None, []).append(f)
)
return f
@setupmethod
def app_errorhandler(
self, code: type[Exception] | int
) -> t.Callable[[T_error_handler], T_error_handler]:
def decorator(f: T_error_handler) -> T_error_handler:
def from_blueprint(state: BlueprintSetupState) -> None:
state.app.errorhandler(code)(f)
self.record_once(from_blueprint)
return f
return decorator
@setupmethod
def app_url_value_preprocessor(
self, f: T_url_value_preprocessor
) -> T_url_value_preprocessor:
self.record_once(
lambda s: s.app.url_value_preprocessors.setdefault(None, []).append(f)
)
return f
@setupmethod
def app_url_defaults(self, f: T_url_defaults) -> T_url_defaults:
self.record_once(
lambda s: s.app.url_default_functions.setdefault(None, []).append(f)
)
return f