codekingpro/portable-devtools

1from __future__ import annotations2 3import copy4import math5import operator6import typing as t7from contextvars import ContextVar8from functools import partial9from functools import update_wrapper10from operator import attrgetter11 12from .wsgi import ClosingIterator13 14if t.TYPE_CHECKING:15    from _typeshed.wsgi import StartResponse16    from _typeshed.wsgi import WSGIApplication17    from _typeshed.wsgi import WSGIEnvironment18 19T = t.TypeVar("T")20F = t.TypeVar("F", bound=t.Callable[..., t.Any])21 22 23def release_local(local: Local | LocalStack[t.Any]) -> None:24    """Release the data for the current context in a :class:`Local` or25    :class:`LocalStack` without using a :class:`LocalManager`.26 27    This should not be needed for modern use cases, and may be removed28    in the future.29 30    .. versionadded:: 0.6.131    """32    local.__release_local__()33 34 35class Local:36    """Create a namespace of context-local data. This wraps a37    :class:`ContextVar` containing a :class:`dict` value.38 39    This may incur a performance penalty compared to using individual40    context vars, as it has to copy data to avoid mutating the dict41    between nested contexts.42 43    :param context_var: The :class:`~contextvars.ContextVar` to use as44        storage for this local. If not given, one will be created.45        Context vars not created at the global scope may interfere with46        garbage collection.47 48    .. versionchanged:: 2.049        Uses ``ContextVar`` instead of a custom storage implementation.50    """51 52    __slots__ = ("__storage",)53 54    def __init__(self, context_var: ContextVar[dict[str, t.Any]] | None = None) -> None:55        if context_var is None:56            # A ContextVar not created at global scope interferes with57            # Python's garbage collection. However, a local only makes58            # sense defined at the global scope as well, in which case59            # the GC issue doesn't seem relevant.60            context_var = ContextVar(f"werkzeug.Local<{id(self)}>.storage")61 62        object.__setattr__(self, "_Local__storage", context_var)63 64    def __iter__(self) -> t.Iterator[tuple[str, t.Any]]:65        return iter(self.__storage.get({}).items())66 67    def __call__(68        self, name: str, *, unbound_message: str | None = None69    ) -> LocalProxy[t.Any]:70        """Create a :class:`LocalProxy` that access an attribute on this71        local namespace.72 73        :param name: Proxy this attribute.74        :param unbound_message: The error message that the proxy will75            show if the attribute isn't set.76        """77        return LocalProxy(self, name, unbound_message=unbound_message)78 79    def __release_local__(self) -> None:80        self.__storage.set({})81 82    def __getattr__(self, name: str) -> t.Any:83        values = self.__storage.get({})84 85        if name in values:86            return values[name]87 88        raise AttributeError(name)89 90    def __setattr__(self, name: str, value: t.Any) -> None:91        values = self.__storage.get({}).copy()92        values[name] = value93        self.__storage.set(values)94 95    def __delattr__(self, name: str) -> None:96        values = self.__storage.get({})97 98        if name in values:99            values = values.copy()100            del values[name]101            self.__storage.set(values)102        else:103            raise AttributeError(name)104 105 106class LocalStack(t.Generic[T]):107    """Create a stack of context-local data. This wraps a108    :class:`ContextVar` containing a :class:`list` value.109 110    This may incur a performance penalty compared to using individual111    context vars, as it has to copy data to avoid mutating the list112    between nested contexts.113 114    :param context_var: The :class:`~contextvars.ContextVar` to use as115        storage for this local. If not given, one will be created.116        Context vars not created at the global scope may interfere with117        garbage collection.118 119    .. versionchanged:: 2.0120        Uses ``ContextVar`` instead of a custom storage implementation.121 122    .. versionadded:: 0.6.1123    """124 125    __slots__ = ("_storage",)126 127    def __init__(self, context_var: ContextVar[list[T]] | None = None) -> None:128        if context_var is None:129            # A ContextVar not created at global scope interferes with130            # Python's garbage collection. However, a local only makes131            # sense defined at the global scope as well, in which case132            # the GC issue doesn't seem relevant.133            context_var = ContextVar(f"werkzeug.LocalStack<{id(self)}>.storage")134 135        self._storage = context_var136 137    def __release_local__(self) -> None:138        self._storage.set([])139 140    def push(self, obj: T) -> list[T]:141        """Add a new item to the top of the stack."""142        stack = self._storage.get([]).copy()143        stack.append(obj)144        self._storage.set(stack)145        return stack146 147    def pop(self) -> T | None:148        """Remove the top item from the stack and return it. If the149        stack is empty, return ``None``.150        """151        stack = self._storage.get([])152 153        if len(stack) == 0:154            return None155 156        rv = stack[-1]157        self._storage.set(stack[:-1])158        return rv159 160    @property161    def top(self) -> T | None:162        """The topmost item on the stack.  If the stack is empty,163        `None` is returned.164        """165        stack = self._storage.get([])166 167        if len(stack) == 0:168            return None169 170        return stack[-1]171 172    def __call__(173        self, name: str | None = None, *, unbound_message: str | None = None174    ) -> LocalProxy[t.Any]:175        """Create a :class:`LocalProxy` that accesses the top of this176        local stack.177 178        :param name: If given, the proxy access this attribute of the179            top item, rather than the item itself.180        :param unbound_message: The error message that the proxy will181            show if the stack is empty.182        """183        return LocalProxy(self, name, unbound_message=unbound_message)184 185 186class LocalManager:187    """Manage releasing the data for the current context in one or more188    :class:`Local` and :class:`LocalStack` objects.189 190    This should not be needed for modern use cases, and may be removed191    in the future.192 193    :param locals: A local or list of locals to manage.194 195    .. versionchanged:: 2.1196        The ``ident_func`` was removed.197 198    .. versionchanged:: 0.7199        The ``ident_func`` parameter was added.200 201    .. versionchanged:: 0.6.1202        The :func:`release_local` function can be used instead of a203        manager.204    """205 206    __slots__ = ("locals",)207 208    def __init__(209        self,210        locals: None211        | (Local | LocalStack[t.Any] | t.Iterable[Local | LocalStack[t.Any]]) = None,212    ) -> None:213        if locals is None:214            self.locals = []215        elif isinstance(locals, Local):216            self.locals = [locals]217        else:218            self.locals = list(locals)  # type: ignore[arg-type]219 220    def cleanup(self) -> None:221        """Release the data in the locals for this context. Call this at222        the end of each request or use :meth:`make_middleware`.223        """224        for local in self.locals:225            release_local(local)226 227    def make_middleware(self, app: WSGIApplication) -> WSGIApplication:228        """Wrap a WSGI application so that local data is released229        automatically after the response has been sent for a request.230        """231 232        def application(233            environ: WSGIEnvironment, start_response: StartResponse234        ) -> t.Iterable[bytes]:235            return ClosingIterator(app(environ, start_response), self.cleanup)236 237        return application238 239    def middleware(self, func: WSGIApplication) -> WSGIApplication:240        """Like :meth:`make_middleware` but used as a decorator on the241        WSGI application function.242 243        .. code-block:: python244 245            @manager.middleware246            def application(environ, start_response):247                ...248        """249        return update_wrapper(self.make_middleware(func), func)250 251    def __repr__(self) -> str:252        return f"<{type(self).__name__} storages: {len(self.locals)}>"253 254 255class _ProxyLookup:256    """Descriptor that handles proxied attribute lookup for257    :class:`LocalProxy`.258 259    :param f: The built-in function this attribute is accessed through.260        Instead of looking up the special method, the function call261        is redone on the object.262    :param fallback: Return this function if the proxy is unbound263        instead of raising a :exc:`RuntimeError`.264    :param is_attr: This proxied name is an attribute, not a function.265        Call the fallback immediately to get the value.266    :param class_value: Value to return when accessed from the267        ``LocalProxy`` class directly. Used for ``__doc__`` so building268        docs still works.269    """270 271    __slots__ = ("bind_f", "fallback", "is_attr", "class_value", "name")272 273    def __init__(274        self,275        f: t.Callable[..., t.Any] | None = None,276        fallback: t.Callable[[LocalProxy[t.Any]], t.Any] | None = None,277        class_value: t.Any | None = None,278        is_attr: bool = False,279    ) -> None:280        bind_f: t.Callable[[LocalProxy[t.Any], t.Any], t.Callable[..., t.Any]] | None281 282        if hasattr(f, "__get__"):283            # A Python function, can be turned into a bound method.284 285            def bind_f(286                instance: LocalProxy[t.Any], obj: t.Any287            ) -> t.Callable[..., t.Any]:288                return f.__get__(obj, type(obj))  # type: ignore289 290        elif f is not None:291            # A C function, use partial to bind the first argument.292 293            def bind_f(294                instance: LocalProxy[t.Any], obj: t.Any295            ) -> t.Callable[..., t.Any]:296                return partial(f, obj)297 298        else:299            # Use getattr, which will produce a bound method.300            bind_f = None301 302        self.bind_f = bind_f303        self.fallback = fallback304        self.class_value = class_value305        self.is_attr = is_attr306 307    def __set_name__(self, owner: LocalProxy[t.Any], name: str) -> None:308        self.name = name309 310    def __get__(self, instance: LocalProxy[t.Any], owner: type | None = None) -> t.Any:311        if instance is None:312            if self.class_value is not None:313                return self.class_value314 315            return self316 317        try:318            obj = instance._get_current_object()319        except RuntimeError:320            if self.fallback is None:321                raise322 323            fallback = self.fallback.__get__(instance, owner)324 325            if self.is_attr:326                # __class__ and __doc__ are attributes, not methods.327                # Call the fallback to get the value.328                return fallback()329 330            return fallback331 332        if self.bind_f is not None:333            return self.bind_f(instance, obj)334 335        return getattr(obj, self.name)336 337    def __repr__(self) -> str:338        return f"proxy {self.name}"339 340    def __call__(341        self, instance: LocalProxy[t.Any], *args: t.Any, **kwargs: t.Any342    ) -> t.Any:343        """Support calling unbound methods from the class. For example,344        this happens with ``copy.copy``, which does345        ``type(x).__copy__(x)``. ``type(x)`` can't be proxied, so it346        returns the proxy type and descriptor.347        """348        return self.__get__(instance, type(instance))(*args, **kwargs)349 350 351class _ProxyIOp(_ProxyLookup):352    """Look up an augmented assignment method on a proxied object. The353    method is wrapped to return the proxy instead of the object.354    """355 356    __slots__ = ()357 358    def __init__(359        self,360        f: t.Callable[..., t.Any] | None = None,361        fallback: t.Callable[[LocalProxy[t.Any]], t.Any] | None = None,362    ) -> None:363        super().__init__(f, fallback)364 365        def bind_f(instance: LocalProxy[t.Any], obj: t.Any) -> t.Callable[..., t.Any]:366            def i_op(self: t.Any, other: t.Any) -> LocalProxy[t.Any]:367                f(self, other)  # type: ignore368                return instance369 370            return i_op.__get__(obj, type(obj))  # type: ignore371 372        self.bind_f = bind_f373 374 375def _l_to_r_op(op: F) -> F:376    """Swap the argument order to turn an l-op into an r-op."""377 378    def r_op(obj: t.Any, other: t.Any) -> t.Any:379        return op(other, obj)380 381    return t.cast(F, r_op)382 383 384def _identity(o: T) -> T:385    return o386 387 388class LocalProxy(t.Generic[T]):389    """A proxy to the object bound to a context-local object. All390    operations on the proxy are forwarded to the bound object. If no391    object is bound, a ``RuntimeError`` is raised.392 393    :param local: The context-local object that provides the proxied394        object.395    :param name: Proxy this attribute from the proxied object.396    :param unbound_message: The error message to show if the397        context-local object is unbound.398 399    Proxy a :class:`~contextvars.ContextVar` to make it easier to400    access. Pass a name to proxy that attribute.401 402    .. code-block:: python403 404        _request_var = ContextVar("request")405        request = LocalProxy(_request_var)406        session = LocalProxy(_request_var, "session")407 408    Proxy an attribute on a :class:`Local` namespace by calling the409    local with the attribute name:410 411    .. code-block:: python412 413        data = Local()414        user = data("user")415 416    Proxy the top item on a :class:`LocalStack` by calling the local.417    Pass a name to proxy that attribute.418 419    .. code-block::420 421        app_stack = LocalStack()422        current_app = app_stack()423        g = app_stack("g")424 425    Pass a function to proxy the return value from that function. This426    was previously used to access attributes of local objects before427    that was supported directly.428 429    .. code-block:: python430 431        session = LocalProxy(lambda: request.session)432 433    ``__repr__`` and ``__class__`` are proxied, so ``repr(x)`` and434    ``isinstance(x, cls)`` will look like the proxied object. Use435    ``issubclass(type(x), LocalProxy)`` to check if an object is a436    proxy.437 438    .. code-block:: python439 440        repr(user)  # <User admin>441        isinstance(user, User)  # True442        issubclass(type(user), LocalProxy)  # True443 444    .. versionchanged:: 2.2.2445        ``__wrapped__`` is set when wrapping an object, not only when446        wrapping a function, to prevent doctest from failing.447 448    .. versionchanged:: 2.2449        Can proxy a ``ContextVar`` or ``LocalStack`` directly.450 451    .. versionchanged:: 2.2452        The ``name`` parameter can be used with any proxied object, not453        only ``Local``.454 455    .. versionchanged:: 2.2456        Added the ``unbound_message`` parameter.457 458    .. versionchanged:: 2.0459        Updated proxied attributes and methods to reflect the current460        data model.461 462    .. versionchanged:: 0.6.1463        The class can be instantiated with a callable.464    """465 466    __slots__ = ("__wrapped", "_get_current_object")467 468    _get_current_object: t.Callable[[], T]469    """Return the current object this proxy is bound to. If the proxy is470    unbound, this raises a ``RuntimeError``.471 472    This should be used if you need to pass the object to something that473    doesn't understand the proxy. It can also be useful for performance474    if you are accessing the object multiple times in a function, rather475    than going through the proxy multiple times.476    """477 478    def __init__(479        self,480        local: ContextVar[T] | Local | LocalStack[T] | t.Callable[[], T],481        name: str | None = None,482        *,483        unbound_message: str | None = None,484    ) -> None:485        if name is None:486            get_name = _identity487        else:488            get_name = attrgetter(name)  # type: ignore[assignment]489 490        if unbound_message is None:491            unbound_message = "object is not bound"492 493        if isinstance(local, Local):494            if name is None:495                raise TypeError("'name' is required when proxying a 'Local' object.")496 497            def _get_current_object() -> T:498                try:499                    return get_name(local)  # type: ignore[return-value]500                except AttributeError:501                    raise RuntimeError(unbound_message) from None502 503        elif isinstance(local, LocalStack):504 505            def _get_current_object() -> T:506                obj = local.top507 508                if obj is None:509                    raise RuntimeError(unbound_message)510 511                return get_name(obj)512 513        elif isinstance(local, ContextVar):514 515            def _get_current_object() -> T:516                try:517                    obj = local.get()518                except LookupError:519                    raise RuntimeError(unbound_message) from None520 521                return get_name(obj)522 523        elif callable(local):524 525            def _get_current_object() -> T:526                return get_name(local())527 528        else:529            raise TypeError(f"Don't know how to proxy '{type(local)}'.")530 531        object.__setattr__(self, "_LocalProxy__wrapped", local)532        object.__setattr__(self, "_get_current_object", _get_current_object)533 534    __doc__ = _ProxyLookup(535        class_value=__doc__, fallback=lambda self: type(self).__doc__, is_attr=True536    )537    __wrapped__ = _ProxyLookup(538        fallback=lambda self: self._LocalProxy__wrapped,  # type: ignore[attr-defined]539        is_attr=True,540    )541    # __del__ should only delete the proxy542    __repr__ = _ProxyLookup(543        repr, fallback=lambda self: f"<{type(self).__name__} unbound>"544    )545    __str__ = _ProxyLookup(str)546    __bytes__ = _ProxyLookup(bytes)547    __format__ = _ProxyLookup()548    __lt__ = _ProxyLookup(operator.lt)549    __le__ = _ProxyLookup(operator.le)550    __eq__ = _ProxyLookup(operator.eq)551    __ne__ = _ProxyLookup(operator.ne)552    __gt__ = _ProxyLookup(operator.gt)553    __ge__ = _ProxyLookup(operator.ge)554    __hash__ = _ProxyLookup(hash)555    __bool__ = _ProxyLookup(bool, fallback=lambda self: False)556    __getattr__ = _ProxyLookup(getattr)557    # __getattribute__ triggered through __getattr__558    __setattr__ = _ProxyLookup(setattr)559    __delattr__ = _ProxyLookup(delattr)560    __dir__ = _ProxyLookup(dir, fallback=lambda self: [])561    # __get__ (proxying descriptor not supported)562    # __set__ (descriptor)563    # __delete__ (descriptor)564    # __set_name__ (descriptor)565    # __objclass__ (descriptor)566    # __slots__ used by proxy itself567    # __dict__ (__getattr__)568    # __weakref__ (__getattr__)569    # __init_subclass__ (proxying metaclass not supported)570    # __prepare__ (metaclass)571    __class__ = _ProxyLookup(fallback=lambda self: type(self), is_attr=True)572    __instancecheck__ = _ProxyLookup(lambda self, other: isinstance(other, self))573    __subclasscheck__ = _ProxyLookup(lambda self, other: issubclass(other, self))574    # __class_getitem__ triggered through __getitem__575    __call__ = _ProxyLookup(lambda self, *args, **kwargs: self(*args, **kwargs))576    __len__ = _ProxyLookup(len)577    __length_hint__ = _ProxyLookup(operator.length_hint)578    __getitem__ = _ProxyLookup(operator.getitem)579    __setitem__ = _ProxyLookup(operator.setitem)580    __delitem__ = _ProxyLookup(operator.delitem)581    # __missing__ triggered through __getitem__582    __iter__ = _ProxyLookup(iter)583    __next__ = _ProxyLookup(next)584    __reversed__ = _ProxyLookup(reversed)585    __contains__ = _ProxyLookup(operator.contains)586    __add__ = _ProxyLookup(operator.add)587    __sub__ = _ProxyLookup(operator.sub)588    __mul__ = _ProxyLookup(operator.mul)589    __matmul__ = _ProxyLookup(operator.matmul)590    __truediv__ = _ProxyLookup(operator.truediv)591    __floordiv__ = _ProxyLookup(operator.floordiv)592    __mod__ = _ProxyLookup(operator.mod)593    __divmod__ = _ProxyLookup(divmod)594    __pow__ = _ProxyLookup(pow)595    __lshift__ = _ProxyLookup(operator.lshift)596    __rshift__ = _ProxyLookup(operator.rshift)597    __and__ = _ProxyLookup(operator.and_)598    __xor__ = _ProxyLookup(operator.xor)599    __or__ = _ProxyLookup(operator.or_)600    __radd__ = _ProxyLookup(_l_to_r_op(operator.add))601    __rsub__ = _ProxyLookup(_l_to_r_op(operator.sub))602    __rmul__ = _ProxyLookup(_l_to_r_op(operator.mul))603    __rmatmul__ = _ProxyLookup(_l_to_r_op(operator.matmul))604    __rtruediv__ = _ProxyLookup(_l_to_r_op(operator.truediv))605    __rfloordiv__ = _ProxyLookup(_l_to_r_op(operator.floordiv))606    __rmod__ = _ProxyLookup(_l_to_r_op(operator.mod))607    __rdivmod__ = _ProxyLookup(_l_to_r_op(divmod))608    __rpow__ = _ProxyLookup(_l_to_r_op(pow))609    __rlshift__ = _ProxyLookup(_l_to_r_op(operator.lshift))610    __rrshift__ = _ProxyLookup(_l_to_r_op(operator.rshift))611    __rand__ = _ProxyLookup(_l_to_r_op(operator.and_))612    __rxor__ = _ProxyLookup(_l_to_r_op(operator.xor))613    __ror__ = _ProxyLookup(_l_to_r_op(operator.or_))614    __iadd__ = _ProxyIOp(operator.iadd)615    __isub__ = _ProxyIOp(operator.isub)616    __imul__ = _ProxyIOp(operator.imul)617    __imatmul__ = _ProxyIOp(operator.imatmul)618    __itruediv__ = _ProxyIOp(operator.itruediv)619    __ifloordiv__ = _ProxyIOp(operator.ifloordiv)620    __imod__ = _ProxyIOp(operator.imod)621    __ipow__ = _ProxyIOp(operator.ipow)622    __ilshift__ = _ProxyIOp(operator.ilshift)623    __irshift__ = _ProxyIOp(operator.irshift)624    __iand__ = _ProxyIOp(operator.iand)625    __ixor__ = _ProxyIOp(operator.ixor)626    __ior__ = _ProxyIOp(operator.ior)627    __neg__ = _ProxyLookup(operator.neg)628    __pos__ = _ProxyLookup(operator.pos)629    __abs__ = _ProxyLookup(abs)630    __invert__ = _ProxyLookup(operator.invert)631    __complex__ = _ProxyLookup(complex)632    __int__ = _ProxyLookup(int)633    __float__ = _ProxyLookup(float)634    __index__ = _ProxyLookup(operator.index)635    __round__ = _ProxyLookup(round)636    __trunc__ = _ProxyLookup(math.trunc)637    __floor__ = _ProxyLookup(math.floor)638    __ceil__ = _ProxyLookup(math.ceil)639    __enter__ = _ProxyLookup()640    __exit__ = _ProxyLookup()641    __await__ = _ProxyLookup()642    __aiter__ = _ProxyLookup()643    __anext__ = _ProxyLookup()644    __aenter__ = _ProxyLookup()645    __aexit__ = _ProxyLookup()646    __copy__ = _ProxyLookup(copy.copy)647    __deepcopy__ = _ProxyLookup(copy.deepcopy)648    # __getnewargs_ex__ (pickle through proxy not supported)649    # __getnewargs__ (pickle)650    # __getstate__ (pickle)651    # __setstate__ (pickle)652    # __reduce__ (pickle)653    # __reduce_ex__ (pickle)654