Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| import importlib.util | |
| import os | |
| import pathlib | |
| import pkgutil | |
| import sys | |
| import typing as t | |
| from collections import defaultdict | |
| from datetime import timedelta | |
| from functools import update_wrapper | |
| from jinja2 import FileSystemLoader | |
| from werkzeug.exceptions import default_exceptions | |
| from werkzeug.exceptions import HTTPException | |
| from werkzeug.utils import cached_property | |
| from . import typing as ft | |
| from .cli import AppGroup | |
| from .globals import current_app | |
| from .helpers import get_root_path | |
| from .helpers import send_from_directory | |
| from .templating import _default_template_ctx_processor | |
| if t.TYPE_CHECKING: # pragma: no cover | |
| from .wrappers import Response | |
| # a singleton sentinel value for parameter defaults | |
| _sentinel = object() | |
| F = t.TypeVar("F", bound=t.Callable[..., t.Any]) | |
| T_after_request = t.TypeVar("T_after_request", bound=ft.AfterRequestCallable) | |
| T_before_request = t.TypeVar("T_before_request", bound=ft.BeforeRequestCallable) | |
| T_error_handler = t.TypeVar("T_error_handler", bound=ft.ErrorHandlerCallable) | |
| T_teardown = t.TypeVar("T_teardown", bound=ft.TeardownCallable) | |
| T_template_context_processor = t.TypeVar( | |
| "T_template_context_processor", bound=ft.TemplateContextProcessorCallable | |
| ) | |
| T_url_defaults = t.TypeVar("T_url_defaults", bound=ft.URLDefaultCallable) | |
| T_url_value_preprocessor = t.TypeVar( | |
| "T_url_value_preprocessor", bound=ft.URLValuePreprocessorCallable | |
| ) | |
| T_route = t.TypeVar("T_route", bound=ft.RouteCallable) | |
| def setupmethod(f: F) -> F: | |
| f_name = f.__name__ | |
| def wrapper_func(self, *args: t.Any, **kwargs: t.Any) -> t.Any: | |
| self._check_setup_finished(f_name) | |
| return f(self, *args, **kwargs) | |
| return t.cast(F, update_wrapper(wrapper_func, f)) | |
| class Scaffold: | |
| """Common behavior shared between :class:`~flask.Flask` and | |
| :class:`~flask.blueprints.Blueprint`. | |
| :param import_name: The import name of the module where this object | |
| is defined. Usually :attr:`__name__` should be used. | |
| :param static_folder: Path to a folder of static files to serve. | |
| If this is set, a static route will be added. | |
| :param static_url_path: URL prefix for the static route. | |
| :param template_folder: Path to a folder containing template files. | |
| for rendering. If this is set, a Jinja loader will be added. | |
| :param root_path: The path that static, template, and resource files | |
| are relative to. Typically not set, it is discovered based on | |
| the ``import_name``. | |
| .. versionadded:: 2.0 | |
| """ | |
| name: str | |
| _static_folder: str | None = None | |
| _static_url_path: str | None = None | |
| def __init__( | |
| self, | |
| import_name: str, | |
| static_folder: str | os.PathLike | None = None, | |
| static_url_path: str | None = None, | |
| template_folder: str | os.PathLike | None = None, | |
| root_path: str | None = None, | |
| ): | |
| #: The name of the package or module that this object belongs | |
| #: to. Do not change this once it is set by the constructor. | |
| self.import_name = import_name | |
| self.static_folder = static_folder # type: ignore | |
| self.static_url_path = static_url_path | |
| #: The path to the templates folder, relative to | |
| #: :attr:`root_path`, to add to the template loader. ``None`` if | |
| #: templates should not be added. | |
| self.template_folder = template_folder | |
| if root_path is None: | |
| root_path = get_root_path(self.import_name) | |
| #: Absolute path to the package on the filesystem. Used to look | |
| #: up resources contained in the package. | |
| self.root_path = root_path | |
| #: The Click command group for registering CLI commands for this | |
| #: object. The commands are available from the ``flask`` command | |
| #: once the application has been discovered and blueprints have | |
| #: been registered. | |
| self.cli = AppGroup() | |
| #: A dictionary mapping endpoint names to view functions. | |
| #: | |
| #: To register a view function, use the :meth:`route` decorator. | |
| #: | |
| #: This data structure is internal. It should not be modified | |
| #: directly and its format may change at any time. | |
| self.view_functions: dict[str, t.Callable] = {} | |
| #: A data structure of registered error handlers, in the format | |
| #: ``{scope: {code: {class: handler}}}``. The ``scope`` key is | |
| #: the name of a blueprint the handlers are active for, or | |
| #: ``None`` for all requests. The ``code`` key is the HTTP | |
| #: status code for ``HTTPException``, or ``None`` for | |
| #: other exceptions. The innermost dictionary maps exception | |
| #: classes to handler functions. | |
| #: | |
| #: To register an error handler, use the :meth:`errorhandler` | |
| #: decorator. | |
| #: | |
| #: This data structure is internal. It should not be modified | |
| #: directly and its format may change at any time. | |
| self.error_handler_spec: dict[ | |
| ft.AppOrBlueprintKey, | |
| dict[int | None, dict[type[Exception], ft.ErrorHandlerCallable]], | |
| ] = defaultdict(lambda: defaultdict(dict)) | |
| #: A data structure of functions to call at the beginning of | |
| #: each request, in the format ``{scope: [functions]}``. The | |
| #: ``scope`` key is the name of a blueprint the functions are | |
| #: active for, or ``None`` for all requests. | |
| #: | |
| #: To register a function, use the :meth:`before_request` | |
| #: decorator. | |
| #: | |
| #: This data structure is internal. It should not be modified | |
| #: directly and its format may change at any time. | |
| self.before_request_funcs: dict[ | |
| ft.AppOrBlueprintKey, list[ft.BeforeRequestCallable] | |
| ] = defaultdict(list) | |
| #: A data structure of functions to call at the end of each | |
| #: request, in the format ``{scope: [functions]}``. The | |
| #: ``scope`` key is the name of a blueprint the functions are | |
| #: active for, or ``None`` for all requests. | |
| #: | |
| #: To register a function, use the :meth:`after_request` | |
| #: decorator. | |
| #: | |
| #: This data structure is internal. It should not be modified | |
| #: directly and its format may change at any time. | |
| self.after_request_funcs: dict[ | |
| ft.AppOrBlueprintKey, list[ft.AfterRequestCallable] | |
| ] = defaultdict(list) | |
| #: A data structure of functions to call at the end of each | |
| #: request even if an exception is raised, in the format | |
| #: ``{scope: [functions]}``. The ``scope`` key is the name of a | |
| #: blueprint the functions are active for, or ``None`` for all | |
| #: requests. | |
| #: | |
| #: To register a function, use the :meth:`teardown_request` | |
| #: decorator. | |
| #: | |
| #: This data structure is internal. It should not be modified | |
| #: directly and its format may change at any time. | |
| self.teardown_request_funcs: dict[ | |
| ft.AppOrBlueprintKey, list[ft.TeardownCallable] | |
| ] = defaultdict(list) | |
| #: A data structure of functions to call to pass extra context | |
| #: values when rendering templates, in the format | |
| #: ``{scope: [functions]}``. The ``scope`` key is the name of a | |
| #: blueprint the functions are active for, or ``None`` for all | |
| #: requests. | |
| #: | |
| #: To register a function, use the :meth:`context_processor` | |
| #: decorator. | |
| #: | |
| #: This data structure is internal. It should not be modified | |
| #: directly and its format may change at any time. | |
| self.template_context_processors: dict[ | |
| ft.AppOrBlueprintKey, list[ft.TemplateContextProcessorCallable] | |
| ] = defaultdict(list, {None: [_default_template_ctx_processor]}) | |
| #: A data structure of functions to call to modify the keyword | |
| #: arguments passed to the view function, in the format | |
| #: ``{scope: [functions]}``. The ``scope`` key is the name of a | |
| #: blueprint the functions are active for, or ``None`` for all | |
| #: requests. | |
| #: | |
| #: To register a function, use the | |
| #: :meth:`url_value_preprocessor` decorator. | |
| #: | |
| #: This data structure is internal. It should not be modified | |
| #: directly and its format may change at any time. | |
| self.url_value_preprocessors: dict[ | |
| ft.AppOrBlueprintKey, | |
| list[ft.URLValuePreprocessorCallable], | |
| ] = defaultdict(list) | |
| #: A data structure of functions to call to modify the keyword | |
| #: arguments when generating URLs, in the format | |
| #: ``{scope: [functions]}``. The ``scope`` key is the name of a | |
| #: blueprint the functions are active for, or ``None`` for all | |
| #: requests. | |
| #: | |
| #: To register a function, use the :meth:`url_defaults` | |
| #: decorator. | |
| #: | |
| #: This data structure is internal. It should not be modified | |
| #: directly and its format may change at any time. | |
| self.url_default_functions: dict[ | |
| ft.AppOrBlueprintKey, list[ft.URLDefaultCallable] | |
| ] = defaultdict(list) | |
| def __repr__(self) -> str: | |
| return f"<{type(self).__name__} {self.name!r}>" | |
| def _check_setup_finished(self, f_name: str) -> None: | |
| raise NotImplementedError | |
| def static_folder(self) -> str | None: | |
| """The absolute path to the configured static folder. ``None`` | |
| if no static folder is set. | |
| """ | |
| if self._static_folder is not None: | |
| return os.path.join(self.root_path, self._static_folder) | |
| else: | |
| return None | |
| def static_folder(self, value: str | os.PathLike | None) -> None: | |
| if value is not None: | |
| value = os.fspath(value).rstrip(r"\/") | |
| self._static_folder = value | |
| def has_static_folder(self) -> bool: | |
| """``True`` if :attr:`static_folder` is set. | |
| .. versionadded:: 0.5 | |
| """ | |
| return self.static_folder is not None | |
| def static_url_path(self) -> str | None: | |
| """The URL prefix that the static route will be accessible from. | |
| If it was not configured during init, it is derived from | |
| :attr:`static_folder`. | |
| """ | |
| if self._static_url_path is not None: | |
| return self._static_url_path | |
| if self.static_folder is not None: | |
| basename = os.path.basename(self.static_folder) | |
| return f"/{basename}".rstrip("/") | |
| return None | |
| def static_url_path(self, value: str | None) -> None: | |
| if value is not None: | |
| value = value.rstrip("/") | |
| self._static_url_path = value | |
| def get_send_file_max_age(self, filename: str | None) -> int | None: | |
| """Used by :func:`send_file` to determine the ``max_age`` cache | |
| value for a given file path if it wasn't passed. | |
| By default, this returns :data:`SEND_FILE_MAX_AGE_DEFAULT` from | |
| the configuration of :data:`~flask.current_app`. This defaults | |
| to ``None``, which tells the browser to use conditional requests | |
| instead of a timed cache, which is usually preferable. | |
| .. versionchanged:: 2.0 | |
| The default configuration is ``None`` instead of 12 hours. | |
| .. versionadded:: 0.9 | |
| """ | |
| value = current_app.config["SEND_FILE_MAX_AGE_DEFAULT"] | |
| if value is None: | |
| return None | |
| if isinstance(value, timedelta): | |
| return int(value.total_seconds()) | |
| return value | |
| def send_static_file(self, filename: str) -> Response: | |
| """The view function used to serve files from | |
| :attr:`static_folder`. A route is automatically registered for | |
| this view at :attr:`static_url_path` if :attr:`static_folder` is | |
| set. | |
| .. versionadded:: 0.5 | |
| """ | |
| if not self.has_static_folder: | |
| raise RuntimeError("'static_folder' must be set to serve static_files.") | |
| # send_file only knows to call get_send_file_max_age on the app, | |
| # call it here so it works for blueprints too. | |
| max_age = self.get_send_file_max_age(filename) | |
| return send_from_directory( | |
| t.cast(str, self.static_folder), filename, max_age=max_age | |
| ) | |
| def jinja_loader(self) -> FileSystemLoader | None: | |
| """The Jinja loader for this object's templates. By default this | |
| is a class :class:`jinja2.loaders.FileSystemLoader` to | |
| :attr:`template_folder` if it is set. | |
| .. versionadded:: 0.5 | |
| """ | |
| if self.template_folder is not None: | |
| return FileSystemLoader(os.path.join(self.root_path, self.template_folder)) | |
| else: | |
| return None | |
| def open_resource(self, resource: str, mode: str = "rb") -> t.IO[t.AnyStr]: | |
| """Open a resource file relative to :attr:`root_path` for | |
| reading. | |
| For example, if the file ``schema.sql`` is next to the file | |
| ``app.py`` where the ``Flask`` app is defined, it can be opened | |
| with: | |
| .. code-block:: python | |
| with app.open_resource("schema.sql") as f: | |
| conn.executescript(f.read()) | |
| :param resource: Path to the resource relative to | |
| :attr:`root_path`. | |
| :param mode: Open the file in this mode. Only reading is | |
| supported, valid values are "r" (or "rt") and "rb". | |
| """ | |
| if mode not in {"r", "rt", "rb"}: | |
| raise ValueError("Resources can only be opened for reading.") | |
| return open(os.path.join(self.root_path, resource), mode) | |
| def _method_route( | |
| self, | |
| method: str, | |
| rule: str, | |
| options: dict, | |
| ) -> t.Callable[[T_route], T_route]: | |
| if "methods" in options: | |
| raise TypeError("Use the 'route' decorator to use the 'methods' argument.") | |
| return self.route(rule, methods=[method], **options) | |
| def get(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]: | |
| """Shortcut for :meth:`route` with ``methods=["GET"]``. | |
| .. versionadded:: 2.0 | |
| """ | |
| return self._method_route("GET", rule, options) | |
| def post(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]: | |
| """Shortcut for :meth:`route` with ``methods=["POST"]``. | |
| .. versionadded:: 2.0 | |
| """ | |
| return self._method_route("POST", rule, options) | |
| def put(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]: | |
| """Shortcut for :meth:`route` with ``methods=["PUT"]``. | |
| .. versionadded:: 2.0 | |
| """ | |
| return self._method_route("PUT", rule, options) | |
| def delete(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]: | |
| """Shortcut for :meth:`route` with ``methods=["DELETE"]``. | |
| .. versionadded:: 2.0 | |
| """ | |
| return self._method_route("DELETE", rule, options) | |
| def patch(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]: | |
| """Shortcut for :meth:`route` with ``methods=["PATCH"]``. | |
| .. versionadded:: 2.0 | |
| """ | |
| return self._method_route("PATCH", rule, options) | |
| def route(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]: | |
| """Decorate a view function to register it with the given URL | |
| rule and options. Calls :meth:`add_url_rule`, which has more | |
| details about the implementation. | |
| .. code-block:: python | |
| @app.route("/") | |
| def index(): | |
| return "Hello, World!" | |
| See :ref:`url-route-registrations`. | |
| The endpoint name for the route defaults to the name of the view | |
| function if the ``endpoint`` parameter isn't passed. | |
| The ``methods`` parameter defaults to ``["GET"]``. ``HEAD`` and | |
| ``OPTIONS`` are added automatically. | |
| :param rule: The URL rule string. | |
| :param options: Extra options passed to the | |
| :class:`~werkzeug.routing.Rule` object. | |
| """ | |
| def decorator(f: T_route) -> T_route: | |
| endpoint = options.pop("endpoint", None) | |
| self.add_url_rule(rule, endpoint, f, **options) | |
| return f | |
| return decorator | |
| 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: | |
| """Register a rule for routing incoming requests and building | |
| URLs. The :meth:`route` decorator is a shortcut to call this | |
| with the ``view_func`` argument. These are equivalent: | |
| .. code-block:: python | |
| @app.route("/") | |
| def index(): | |
| ... | |
| .. code-block:: python | |
| def index(): | |
| ... | |
| app.add_url_rule("/", view_func=index) | |
| See :ref:`url-route-registrations`. | |
| The endpoint name for the route defaults to the name of the view | |
| function if the ``endpoint`` parameter isn't passed. An error | |
| will be raised if a function has already been registered for the | |
| endpoint. | |
| The ``methods`` parameter defaults to ``["GET"]``. ``HEAD`` is | |
| always added automatically, and ``OPTIONS`` is added | |
| automatically by default. | |
| ``view_func`` does not necessarily need to be passed, but if the | |
| rule should participate in routing an endpoint name must be | |
| associated with a view function at some point with the | |
| :meth:`endpoint` decorator. | |
| .. code-block:: python | |
| app.add_url_rule("/", endpoint="index") | |
| @app.endpoint("index") | |
| def index(): | |
| ... | |
| If ``view_func`` has a ``required_methods`` attribute, those | |
| methods are added to the passed and automatic methods. If it | |
| has a ``provide_automatic_methods`` attribute, it is used as the | |
| default if the parameter is not passed. | |
| :param rule: The URL rule string. | |
| :param endpoint: The endpoint name to associate with the rule | |
| and view function. Used when routing and building URLs. | |
| Defaults to ``view_func.__name__``. | |
| :param view_func: The view function to associate with the | |
| endpoint name. | |
| :param provide_automatic_options: Add the ``OPTIONS`` method and | |
| respond to ``OPTIONS`` requests automatically. | |
| :param options: Extra options passed to the | |
| :class:`~werkzeug.routing.Rule` object. | |
| """ | |
| raise NotImplementedError | |
| def endpoint(self, endpoint: str) -> t.Callable[[F], F]: | |
| """Decorate a view function to register it for the given | |
| endpoint. Used if a rule is added without a ``view_func`` with | |
| :meth:`add_url_rule`. | |
| .. code-block:: python | |
| app.add_url_rule("/ex", endpoint="example") | |
| @app.endpoint("example") | |
| def example(): | |
| ... | |
| :param endpoint: The endpoint name to associate with the view | |
| function. | |
| """ | |
| def decorator(f: F) -> F: | |
| self.view_functions[endpoint] = f | |
| return f | |
| return decorator | |
| def before_request(self, f: T_before_request) -> T_before_request: | |
| """Register a function to run before each request. | |
| For example, this can be used to open a database connection, or | |
| to load the logged in user from the session. | |
| .. code-block:: python | |
| @app.before_request | |
| def load_user(): | |
| if "user_id" in session: | |
| g.user = db.session.get(session["user_id"]) | |
| The function will be called without any arguments. If it returns | |
| a non-``None`` value, the value is handled as if it was the | |
| return value from the view, and further request handling is | |
| stopped. | |
| This is available on both app and blueprint objects. When used on an app, this | |
| executes before every request. When used on a blueprint, this executes before | |
| every request that the blueprint handles. To register with a blueprint and | |
| execute before every request, use :meth:`.Blueprint.before_app_request`. | |
| """ | |
| self.before_request_funcs.setdefault(None, []).append(f) | |
| return f | |
| def after_request(self, f: T_after_request) -> T_after_request: | |
| """Register a function to run after each request to this object. | |
| The function is called with the response object, and must return | |
| a response object. This allows the functions to modify or | |
| replace the response before it is sent. | |
| If a function raises an exception, any remaining | |
| ``after_request`` functions will not be called. Therefore, this | |
| should not be used for actions that must execute, such as to | |
| close resources. Use :meth:`teardown_request` for that. | |
| This is available on both app and blueprint objects. When used on an app, this | |
| executes after every request. When used on a blueprint, this executes after | |
| every request that the blueprint handles. To register with a blueprint and | |
| execute after every request, use :meth:`.Blueprint.after_app_request`. | |
| """ | |
| self.after_request_funcs.setdefault(None, []).append(f) | |
| return f | |
| def teardown_request(self, f: T_teardown) -> T_teardown: | |
| """Register a function to be called when the request context is | |
| popped. Typically this happens at the end of each request, but | |
| contexts may be pushed manually as well during testing. | |
| .. code-block:: python | |
| with app.test_request_context(): | |
| ... | |
| When the ``with`` block exits (or ``ctx.pop()`` is called), the | |
| teardown functions are called just before the request context is | |
| made inactive. | |
| When a teardown function was called because of an unhandled | |
| exception it will be passed an error object. If an | |
| :meth:`errorhandler` is registered, it will handle the exception | |
| and the teardown will not receive it. | |
| Teardown functions must avoid raising exceptions. If they | |
| execute code that might fail they must surround that code with a | |
| ``try``/``except`` block and log any errors. | |
| The return values of teardown functions are ignored. | |
| This is available on both app and blueprint objects. When used on an app, this | |
| executes after every request. When used on a blueprint, this executes after | |
| every request that the blueprint handles. To register with a blueprint and | |
| execute after every request, use :meth:`.Blueprint.teardown_app_request`. | |
| """ | |
| self.teardown_request_funcs.setdefault(None, []).append(f) | |
| return f | |
| def context_processor( | |
| self, | |
| f: T_template_context_processor, | |
| ) -> T_template_context_processor: | |
| """Registers a template context processor function. These functions run before | |
| rendering a template. The keys of the returned dict are added as variables | |
| available in the template. | |
| This is available on both app and blueprint objects. When used on an app, this | |
| is called for every rendered template. When used on a blueprint, this is called | |
| for templates rendered from the blueprint's views. To register with a blueprint | |
| and affect every template, use :meth:`.Blueprint.app_context_processor`. | |
| """ | |
| self.template_context_processors[None].append(f) | |
| return f | |
| def url_value_preprocessor( | |
| self, | |
| f: T_url_value_preprocessor, | |
| ) -> T_url_value_preprocessor: | |
| """Register a URL value preprocessor function for all view | |
| functions in the application. These functions will be called before the | |
| :meth:`before_request` functions. | |
| The function can modify the values captured from the matched url before | |
| they are passed to the view. For example, this can be used to pop a | |
| common language code value and place it in ``g`` rather than pass it to | |
| every view. | |
| The function is passed the endpoint name and values dict. The return | |
| value is ignored. | |
| This is available on both app and blueprint objects. When used on an app, this | |
| is called for every request. When used on a blueprint, this is called for | |
| requests that the blueprint handles. To register with a blueprint and affect | |
| every request, use :meth:`.Blueprint.app_url_value_preprocessor`. | |
| """ | |
| self.url_value_preprocessors[None].append(f) | |
| return f | |
| def url_defaults(self, f: T_url_defaults) -> T_url_defaults: | |
| """Callback function for URL defaults for all view functions of the | |
| application. It's called with the endpoint and values and should | |
| update the values passed in place. | |
| This is available on both app and blueprint objects. When used on an app, this | |
| is called for every request. When used on a blueprint, this is called for | |
| requests that the blueprint handles. To register with a blueprint and affect | |
| every request, use :meth:`.Blueprint.app_url_defaults`. | |
| """ | |
| self.url_default_functions[None].append(f) | |
| return f | |
| def errorhandler( | |
| self, code_or_exception: type[Exception] | int | |
| ) -> t.Callable[[T_error_handler], T_error_handler]: | |
| """Register a function to handle errors by code or exception class. | |
| A decorator that is used to register a function given an | |
| error code. Example:: | |
| @app.errorhandler(404) | |
| def page_not_found(error): | |
| return 'This page does not exist', 404 | |
| You can also register handlers for arbitrary exceptions:: | |
| @app.errorhandler(DatabaseError) | |
| def special_exception_handler(error): | |
| return 'Database connection failed', 500 | |
| This is available on both app and blueprint objects. When used on an app, this | |
| can handle errors from every request. When used on a blueprint, this can handle | |
| errors from requests that the blueprint handles. To register with a blueprint | |
| and affect every request, use :meth:`.Blueprint.app_errorhandler`. | |
| .. versionadded:: 0.7 | |
| Use :meth:`register_error_handler` instead of modifying | |
| :attr:`error_handler_spec` directly, for application wide error | |
| handlers. | |
| .. versionadded:: 0.7 | |
| One can now additionally also register custom exception types | |
| that do not necessarily have to be a subclass of the | |
| :class:`~werkzeug.exceptions.HTTPException` class. | |
| :param code_or_exception: the code as integer for the handler, or | |
| an arbitrary exception | |
| """ | |
| def decorator(f: T_error_handler) -> T_error_handler: | |
| self.register_error_handler(code_or_exception, f) | |
| return f | |
| return decorator | |
| def register_error_handler( | |
| self, | |
| code_or_exception: type[Exception] | int, | |
| f: ft.ErrorHandlerCallable, | |
| ) -> None: | |
| """Alternative error attach function to the :meth:`errorhandler` | |
| decorator that is more straightforward to use for non decorator | |
| usage. | |
| .. versionadded:: 0.7 | |
| """ | |
| exc_class, code = self._get_exc_class_and_code(code_or_exception) | |
| self.error_handler_spec[None][code][exc_class] = f | |
| def _get_exc_class_and_code( | |
| exc_class_or_code: type[Exception] | int, | |
| ) -> tuple[type[Exception], int | None]: | |
| """Get the exception class being handled. For HTTP status codes | |
| or ``HTTPException`` subclasses, return both the exception and | |
| status code. | |
| :param exc_class_or_code: Any exception class, or an HTTP status | |
| code as an integer. | |
| """ | |
| exc_class: type[Exception] | |
| if isinstance(exc_class_or_code, int): | |
| try: | |
| exc_class = default_exceptions[exc_class_or_code] | |
| except KeyError: | |
| raise ValueError( | |
| f"'{exc_class_or_code}' is not a recognized HTTP" | |
| " error code. Use a subclass of HTTPException with" | |
| " that code instead." | |
| ) from None | |
| else: | |
| exc_class = exc_class_or_code | |
| if isinstance(exc_class, Exception): | |
| raise TypeError( | |
| f"{exc_class!r} is an instance, not a class. Handlers" | |
| " can only be registered for Exception classes or HTTP" | |
| " error codes." | |
| ) | |
| if not issubclass(exc_class, Exception): | |
| raise ValueError( | |
| f"'{exc_class.__name__}' is not a subclass of Exception." | |
| " Handlers can only be registered for Exception classes" | |
| " or HTTP error codes." | |
| ) | |
| if issubclass(exc_class, HTTPException): | |
| return exc_class, exc_class.code | |
| else: | |
| return exc_class, None | |
| def _endpoint_from_view_func(view_func: t.Callable) -> str: | |
| """Internal helper that returns the default endpoint for a given | |
| function. This always is the function name. | |
| """ | |
| assert view_func is not None, "expected view func if endpoint is not provided." | |
| return view_func.__name__ | |
| def _matching_loader_thinks_module_is_package(loader, mod_name): | |
| """Attempt to figure out if the given name is a package or a module. | |
| :param: loader: The loader that handled the name. | |
| :param mod_name: The name of the package or module. | |
| """ | |
| # Use loader.is_package if it's available. | |
| if hasattr(loader, "is_package"): | |
| return loader.is_package(mod_name) | |
| cls = type(loader) | |
| # NamespaceLoader doesn't implement is_package, but all names it | |
| # loads must be packages. | |
| if cls.__module__ == "_frozen_importlib" and cls.__name__ == "NamespaceLoader": | |
| return True | |
| # Otherwise we need to fail with an error that explains what went | |
| # wrong. | |
| raise AttributeError( | |
| f"'{cls.__name__}.is_package()' must be implemented for PEP 302" | |
| f" import hooks." | |
| ) | |
| def _path_is_relative_to(path: pathlib.PurePath, base: str) -> bool: | |
| # Path.is_relative_to doesn't exist until Python 3.9 | |
| try: | |
| path.relative_to(base) | |
| return True | |
| except ValueError: | |
| return False | |
| def _find_package_path(import_name): | |
| """Find the path that contains the package or module.""" | |
| root_mod_name, _, _ = import_name.partition(".") | |
| try: | |
| root_spec = importlib.util.find_spec(root_mod_name) | |
| if root_spec is None: | |
| raise ValueError("not found") | |
| # ImportError: the machinery told us it does not exist | |
| # ValueError: | |
| # - the module name was invalid | |
| # - the module name is __main__ | |
| # - *we* raised `ValueError` due to `root_spec` being `None` | |
| except (ImportError, ValueError): | |
| pass # handled below | |
| else: | |
| # namespace package | |
| if root_spec.origin in {"namespace", None}: | |
| package_spec = importlib.util.find_spec(import_name) | |
| if package_spec is not None and package_spec.submodule_search_locations: | |
| # Pick the path in the namespace that contains the submodule. | |
| package_path = pathlib.Path( | |
| os.path.commonpath(package_spec.submodule_search_locations) | |
| ) | |
| search_locations = ( | |
| location | |
| for location in root_spec.submodule_search_locations | |
| if _path_is_relative_to(package_path, location) | |
| ) | |
| else: | |
| # Pick the first path. | |
| search_locations = iter(root_spec.submodule_search_locations) | |
| return os.path.dirname(next(search_locations)) | |
| # a package (with __init__.py) | |
| elif root_spec.submodule_search_locations: | |
| return os.path.dirname(os.path.dirname(root_spec.origin)) | |
| # just a normal module | |
| else: | |
| return os.path.dirname(root_spec.origin) | |
| # we were unable to find the `package_path` using PEP 451 loaders | |
| loader = pkgutil.get_loader(root_mod_name) | |
| if loader is None or root_mod_name == "__main__": | |
| # import name is not found, or interactive/main module | |
| return os.getcwd() | |
| if hasattr(loader, "get_filename"): | |
| filename = loader.get_filename(root_mod_name) | |
| elif hasattr(loader, "archive"): | |
| # zipimporter's loader.archive points to the .egg or .zip file. | |
| filename = loader.archive | |
| else: | |
| # At least one loader is missing both get_filename and archive: | |
| # Google App Engine's HardenedModulesHook, use __file__. | |
| filename = importlib.import_module(root_mod_name).__file__ | |
| package_path = os.path.abspath(os.path.dirname(filename)) | |
| # If the imported name is a package, filename is currently pointing | |
| # to the root of the package, need to get the current directory. | |
| if _matching_loader_thinks_module_is_package(loader, root_mod_name): | |
| package_path = os.path.dirname(package_path) | |
| return package_path | |
| def find_package(import_name: str): | |
| """Find the prefix that a package is installed under, and the path | |
| that it would be imported from. | |
| The prefix is the directory containing the standard directory | |
| hierarchy (lib, bin, etc.). If the package is not installed to the | |
| system (:attr:`sys.prefix`) or a virtualenv (``site-packages``), | |
| ``None`` is returned. | |
| The path is the entry in :attr:`sys.path` that contains the package | |
| for import. If the package is not installed, it's assumed that the | |
| package was imported from the current working directory. | |
| """ | |
| package_path = _find_package_path(import_name) | |
| py_prefix = os.path.abspath(sys.prefix) | |
| # installed to the system | |
| if _path_is_relative_to(pathlib.PurePath(package_path), py_prefix): | |
| return py_prefix, package_path | |
| site_parent, site_folder = os.path.split(package_path) | |
| # installed to a virtualenv | |
| if site_folder.lower() == "site-packages": | |
| parent, folder = os.path.split(site_parent) | |
| # Windows (prefix/lib/site-packages) | |
| if folder.lower() == "lib": | |
| return parent, package_path | |
| # Unix (prefix/lib/pythonX.Y/site-packages) | |
| if os.path.basename(parent).lower() == "lib": | |
| return os.path.dirname(parent), package_path | |
| # something else (prefix/site-packages) | |
| return site_parent, package_path | |
| # not installed | |
| return None, package_path | |