codekingpro/portable-devtools

1from __future__ import annotations2 3import contextvars4import sys5import typing as t6from functools import update_wrapper7from types import TracebackType8 9from werkzeug.exceptions import HTTPException10 11from . import typing as ft12from .globals import _cv_app13from .globals import _cv_request14from .signals import appcontext_popped15from .signals import appcontext_pushed16 17if t.TYPE_CHECKING:  # pragma: no cover18    from _typeshed.wsgi import WSGIEnvironment19 20    from .app import Flask21    from .sessions import SessionMixin22    from .wrappers import Request23 24 25# a singleton sentinel value for parameter defaults26_sentinel = object()27 28 29class _AppCtxGlobals:30    """A plain object. Used as a namespace for storing data during an31    application context.32 33    Creating an app context automatically creates this object, which is34    made available as the :data:`g` proxy.35 36    .. describe:: 'key' in g37 38        Check whether an attribute is present.39 40        .. versionadded:: 0.1041 42    .. describe:: iter(g)43 44        Return an iterator over the attribute names.45 46        .. versionadded:: 0.1047    """48 49    # Define attr methods to let mypy know this is a namespace object50    # that has arbitrary attributes.51 52    def __getattr__(self, name: str) -> t.Any:53        try:54            return self.__dict__[name]55        except KeyError:56            raise AttributeError(name) from None57 58    def __setattr__(self, name: str, value: t.Any) -> None:59        self.__dict__[name] = value60 61    def __delattr__(self, name: str) -> None:62        try:63            del self.__dict__[name]64        except KeyError:65            raise AttributeError(name) from None66 67    def get(self, name: str, default: t.Any | None = None) -> t.Any:68        """Get an attribute by name, or a default value. Like69        :meth:`dict.get`.70 71        :param name: Name of attribute to get.72        :param default: Value to return if the attribute is not present.73 74        .. versionadded:: 0.1075        """76        return self.__dict__.get(name, default)77 78    def pop(self, name: str, default: t.Any = _sentinel) -> t.Any:79        """Get and remove an attribute by name. Like :meth:`dict.pop`.80 81        :param name: Name of attribute to pop.82        :param default: Value to return if the attribute is not present,83            instead of raising a ``KeyError``.84 85        .. versionadded:: 0.1186        """87        if default is _sentinel:88            return self.__dict__.pop(name)89        else:90            return self.__dict__.pop(name, default)91 92    def setdefault(self, name: str, default: t.Any = None) -> t.Any:93        """Get the value of an attribute if it is present, otherwise94        set and return a default value. Like :meth:`dict.setdefault`.95 96        :param name: Name of attribute to get.97        :param default: Value to set and return if the attribute is not98            present.99 100        .. versionadded:: 0.11101        """102        return self.__dict__.setdefault(name, default)103 104    def __contains__(self, item: str) -> bool:105        return item in self.__dict__106 107    def __iter__(self) -> t.Iterator[str]:108        return iter(self.__dict__)109 110    def __repr__(self) -> str:111        ctx = _cv_app.get(None)112        if ctx is not None:113            return f"<flask.g of '{ctx.app.name}'>"114        return object.__repr__(self)115 116 117def after_this_request(118    f: ft.AfterRequestCallable[t.Any],119) -> ft.AfterRequestCallable[t.Any]:120    """Executes a function after this request.  This is useful to modify121    response objects.  The function is passed the response object and has122    to return the same or a new one.123 124    Example::125 126        @app.route('/')127        def index():128            @after_this_request129            def add_header(response):130                response.headers['X-Foo'] = 'Parachute'131                return response132            return 'Hello World!'133 134    This is more useful if a function other than the view function wants to135    modify a response.  For instance think of a decorator that wants to add136    some headers without converting the return value into a response object.137 138    .. versionadded:: 0.9139    """140    ctx = _cv_request.get(None)141 142    if ctx is None:143        raise RuntimeError(144            "'after_this_request' can only be used when a request"145            " context is active, such as in a view function."146        )147 148    ctx._after_request_functions.append(f)149    return f150 151 152F = t.TypeVar("F", bound=t.Callable[..., t.Any])153 154 155def copy_current_request_context(f: F) -> F:156    """A helper function that decorates a function to retain the current157    request context.  This is useful when working with greenlets.  The moment158    the function is decorated a copy of the request context is created and159    then pushed when the function is called.  The current session is also160    included in the copied request context.161 162    Example::163 164        import gevent165        from flask import copy_current_request_context166 167        @app.route('/')168        def index():169            @copy_current_request_context170            def do_some_work():171                # do some work here, it can access flask.request or172                # flask.session like you would otherwise in the view function.173                ...174            gevent.spawn(do_some_work)175            return 'Regular response'176 177    .. versionadded:: 0.10178    """179    ctx = _cv_request.get(None)180 181    if ctx is None:182        raise RuntimeError(183            "'copy_current_request_context' can only be used when a"184            " request context is active, such as in a view function."185        )186 187    ctx = ctx.copy()188 189    def wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any:190        with ctx:  # type: ignore[union-attr]191            return ctx.app.ensure_sync(f)(*args, **kwargs)  # type: ignore[union-attr]192 193    return update_wrapper(wrapper, f)  # type: ignore[return-value]194 195 196def has_request_context() -> bool:197    """If you have code that wants to test if a request context is there or198    not this function can be used.  For instance, you may want to take advantage199    of request information if the request object is available, but fail200    silently if it is unavailable.201 202    ::203 204        class User(db.Model):205 206            def __init__(self, username, remote_addr=None):207                self.username = username208                if remote_addr is None and has_request_context():209                    remote_addr = request.remote_addr210                self.remote_addr = remote_addr211 212    Alternatively you can also just test any of the context bound objects213    (such as :class:`request` or :class:`g`) for truthness::214 215        class User(db.Model):216 217            def __init__(self, username, remote_addr=None):218                self.username = username219                if remote_addr is None and request:220                    remote_addr = request.remote_addr221                self.remote_addr = remote_addr222 223    .. versionadded:: 0.7224    """225    return _cv_request.get(None) is not None226 227 228def has_app_context() -> bool:229    """Works like :func:`has_request_context` but for the application230    context.  You can also just do a boolean check on the231    :data:`current_app` object instead.232 233    .. versionadded:: 0.9234    """235    return _cv_app.get(None) is not None236 237 238class AppContext:239    """The app context contains application-specific information. An app240    context is created and pushed at the beginning of each request if241    one is not already active. An app context is also pushed when242    running CLI commands.243    """244 245    def __init__(self, app: Flask) -> None:246        self.app = app247        self.url_adapter = app.create_url_adapter(None)248        self.g: _AppCtxGlobals = app.app_ctx_globals_class()249        self._cv_tokens: list[contextvars.Token[AppContext]] = []250 251    def push(self) -> None:252        """Binds the app context to the current context."""253        self._cv_tokens.append(_cv_app.set(self))254        appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync)255 256    def pop(self, exc: BaseException | None = _sentinel) -> None:  # type: ignore257        """Pops the app context."""258        try:259            if len(self._cv_tokens) == 1:260                if exc is _sentinel:261                    exc = sys.exc_info()[1]262                self.app.do_teardown_appcontext(exc)263        finally:264            ctx = _cv_app.get()265            _cv_app.reset(self._cv_tokens.pop())266 267        if ctx is not self:268            raise AssertionError(269                f"Popped wrong app context. ({ctx!r} instead of {self!r})"270            )271 272        appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync)273 274    def __enter__(self) -> AppContext:275        self.push()276        return self277 278    def __exit__(279        self,280        exc_type: type | None,281        exc_value: BaseException | None,282        tb: TracebackType | None,283    ) -> None:284        self.pop(exc_value)285 286 287class RequestContext:288    """The request context contains per-request information. The Flask289    app creates and pushes it at the beginning of the request, then pops290    it at the end of the request. It will create the URL adapter and291    request object for the WSGI environment provided.292 293    Do not attempt to use this class directly, instead use294    :meth:`~flask.Flask.test_request_context` and295    :meth:`~flask.Flask.request_context` to create this object.296 297    When the request context is popped, it will evaluate all the298    functions registered on the application for teardown execution299    (:meth:`~flask.Flask.teardown_request`).300 301    The request context is automatically popped at the end of the302    request. When using the interactive debugger, the context will be303    restored so ``request`` is still accessible. Similarly, the test304    client can preserve the context after the request ends. However,305    teardown functions may already have closed some resources such as306    database connections.307    """308 309    def __init__(310        self,311        app: Flask,312        environ: WSGIEnvironment,313        request: Request | None = None,314        session: SessionMixin | None = None,315    ) -> None:316        self.app = app317        if request is None:318            request = app.request_class(environ)319            request.json_module = app.json320        self.request: Request = request321        self.url_adapter = None322        try:323            self.url_adapter = app.create_url_adapter(self.request)324        except HTTPException as e:325            self.request.routing_exception = e326        self.flashes: list[tuple[str, str]] | None = None327        self.session: SessionMixin | None = session328        # Functions that should be executed after the request on the response329        # object.  These will be called before the regular "after_request"330        # functions.331        self._after_request_functions: list[ft.AfterRequestCallable[t.Any]] = []332 333        self._cv_tokens: list[334            tuple[contextvars.Token[RequestContext], AppContext | None]335        ] = []336 337    def copy(self) -> RequestContext:338        """Creates a copy of this request context with the same request object.339        This can be used to move a request context to a different greenlet.340        Because the actual request object is the same this cannot be used to341        move a request context to a different thread unless access to the342        request object is locked.343 344        .. versionadded:: 0.10345 346        .. versionchanged:: 1.1347           The current session object is used instead of reloading the original348           data. This prevents `flask.session` pointing to an out-of-date object.349        """350        return self.__class__(351            self.app,352            environ=self.request.environ,353            request=self.request,354            session=self.session,355        )356 357    def match_request(self) -> None:358        """Can be overridden by a subclass to hook into the matching359        of the request.360        """361        try:362            result = self.url_adapter.match(return_rule=True)  # type: ignore363            self.request.url_rule, self.request.view_args = result  # type: ignore364        except HTTPException as e:365            self.request.routing_exception = e366 367    def push(self) -> None:368        # Before we push the request context we have to ensure that there369        # is an application context.370        app_ctx = _cv_app.get(None)371 372        if app_ctx is None or app_ctx.app is not self.app:373            app_ctx = self.app.app_context()374            app_ctx.push()375        else:376            app_ctx = None377 378        self._cv_tokens.append((_cv_request.set(self), app_ctx))379 380        # Open the session at the moment that the request context is available.381        # This allows a custom open_session method to use the request context.382        # Only open a new session if this is the first time the request was383        # pushed, otherwise stream_with_context loses the session.384        if self.session is None:385            session_interface = self.app.session_interface386            self.session = session_interface.open_session(self.app, self.request)387 388            if self.session is None:389                self.session = session_interface.make_null_session(self.app)390 391        # Match the request URL after loading the session, so that the392        # session is available in custom URL converters.393        if self.url_adapter is not None:394            self.match_request()395 396    def pop(self, exc: BaseException | None = _sentinel) -> None:  # type: ignore397        """Pops the request context and unbinds it by doing that.  This will398        also trigger the execution of functions registered by the399        :meth:`~flask.Flask.teardown_request` decorator.400 401        .. versionchanged:: 0.9402           Added the `exc` argument.403        """404        clear_request = len(self._cv_tokens) == 1405 406        try:407            if clear_request:408                if exc is _sentinel:409                    exc = sys.exc_info()[1]410                self.app.do_teardown_request(exc)411 412                request_close = getattr(self.request, "close", None)413                if request_close is not None:414                    request_close()415        finally:416            ctx = _cv_request.get()417            token, app_ctx = self._cv_tokens.pop()418            _cv_request.reset(token)419 420            # get rid of circular dependencies at the end of the request421            # so that we don't require the GC to be active.422            if clear_request:423                ctx.request.environ["werkzeug.request"] = None424 425            if app_ctx is not None:426                app_ctx.pop(exc)427 428            if ctx is not self:429                raise AssertionError(430                    f"Popped wrong request context. ({ctx!r} instead of {self!r})"431                )432 433    def __enter__(self) -> RequestContext:434        self.push()435        return self436 437    def __exit__(438        self,439        exc_type: type | None,440        exc_value: BaseException | None,441        tb: TracebackType | None,442    ) -> None:443        self.pop(exc_value)444 445    def __repr__(self) -> str:446        return (447            f"<{type(self).__name__} {self.request.url!r}"448            f" [{self.request.method}] of {self.app.name}>"449        )450