mode.locals

Proxy objects.

Proxies

Proxy objects are lazy and pass all method calls and attribute accesses to an underlying object.

There are mixins/roles for many of the generic classes, and these can be combined to create proxies.

For example to create a proxy to a class that both implements the mutable mapping interface and is an async context manager:

def create_real():
    print('CREATING X')
    return X()

class XProxy(MutableMappingRole, AsyncContextManagerRole):
    ...

x = XProxy(create_real)

Evaluation

By default the callable passed to Proxy will be evaluated every time it is needed, so in the example above a new X will be created every time you access the underlying object:

>>> x['foo'] = 'value'
CREATING X

>>> x['foo']
CREATING X
'value'

>>> X['foo']
CREATING X
'value'

>>> # evaluates twice, once for async with then for __getitem__:
>>> async with x:
...    x['foo']
CREATING X
CREATING X
'value'

If you want the creation of the object to be lazy (created when first needed), you can pass the cache=True argument to Proxy:

>>> x = XProxy(create_real, cache=True)

>>> # Now only evaluates the first time it is needed.
>>> x['foo'] = 'value'
CREATING X

>>> x['foo']
'value'

>>> X['foo']
'value'

>>> async with x:
...    x['foo']
'value'

AsyncContextManagerProxy

Bases: Proxy[AbstractAsyncContextManager[T_co]], AsyncContextManagerRole[T_co]

Proxy to typing.AsyncContextManager object.

Source code in mode/locals.py

class AsyncContextManagerProxy(
    Proxy[AbstractAsyncContextManager[T_co]], AsyncContextManagerRole[T_co]
):
    """Proxy to `typing.AsyncContextManager` object."""

AsyncContextManagerRole

Bases: AbstractAsyncContextManager[T_co]

Role/Mixin for contextlib.AbstractAsyncContextManager proxy methods.

Source code in mode/locals.py

class AsyncContextManagerRole(AbstractAsyncContextManager[T_co]):
    """Role/Mixin for `contextlib.AbstractAsyncContextManager` proxy methods."""

    def __aenter__(self) -> Coroutine[Any, Any, T_co]:
        obj = self._get_current_object()  # type: ignore
        return cast(Coroutine[Any, Any, T_co], obj.__aenter__())

    def __aexit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> Coroutine[Any, Any, Optional[bool]]:
        obj = self._get_current_object()  # type: ignore
        val = obj.__aexit__(exc_type, exc_value, traceback)
        return cast(Coroutine[Any, Any, Optional[bool]], val)

AsyncGeneratorProxy

Bases: Proxy[AsyncGenerator[T_co, T_contra]], AsyncGeneratorRole[T_co, T_contra]

Proxy to typing.AsyncGenerator object.

Source code in mode/locals.py

class AsyncGeneratorProxy(
    Proxy[AsyncGenerator[T_co, T_contra]], AsyncGeneratorRole[T_co, T_contra]
):
    """Proxy to `typing.AsyncGenerator` object."""

AsyncGeneratorRole

Bases: AsyncGenerator[T_co, T_contra]

Role/Mixin for typing.AsyncGenerator proxy methods.

Source code in mode/locals.py

class AsyncGeneratorRole(AsyncGenerator[T_co, T_contra]):
    """Role/Mixin for `typing.AsyncGenerator` proxy methods."""

    def _get_generator(self) -> AsyncGenerator[T_co, T_contra]:
        obj = self._get_current_object()  # type: ignore
        return cast(AsyncGenerator[T_co, T_contra], obj)

    def __anext__(self) -> Coroutine[Any, Any, T_co]:
        return self._get_generator().__anext__()

    def asend(self, value: T_contra) -> Coroutine[Any, Any, T_co]:
        return self._get_generator().asend(value)

    def athrow(
        self,
        typ: Union[type[BaseException], BaseException],
        val: object = None,
        tb: Optional[TracebackType] = None,
    ) -> Coroutine[Any, Any, T_co]:
        # See `CoroutineRole.throw`: `AsyncGenerator.athrow` is overloaded
        # the same way.
        athrow = cast(
            Callable[..., Coroutine[Any, Any, T_co]],
            self._get_generator().athrow,
        )
        return athrow(typ, val, tb)

    def aclose(self) -> Coroutine[Any, Any, None]:
        return self._get_generator().aclose()

    def __aiter__(self) -> AsyncGenerator[T_co, T_contra]:
        return cast(
            AsyncGenerator[T_co, T_contra], self._get_generator().__aiter__()
        )

AsyncIterableProxy

Bases: Proxy[AsyncIterable[T_co]], AsyncIterableRole[T_co]

Proxy to typing.AsyncIterable object.

Source code in mode/locals.py

class AsyncIterableProxy(Proxy[AsyncIterable[T_co]], AsyncIterableRole[T_co]):
    """Proxy to `typing.AsyncIterable` object."""

AsyncIterableRole

Bases: AsyncIterable[T_co]

Role/Mixin for typing.AsyncIterable proxy methods.

Source code in mode/locals.py

class AsyncIterableRole(AsyncIterable[T_co]):
    """Role/Mixin for `typing.AsyncIterable` proxy methods."""

    def _get_iterable(self) -> AsyncIterable[T_co]:
        obj = self._get_current_object()  # type: ignore
        return cast(AsyncIterable[T_co], obj)

    def __aiter__(self) -> AsyncIterator[T_co]:
        return self._get_iterable().__aiter__()

AsyncIteratorProxy

Bases: Proxy[AsyncIterator[T_co]], AsyncIteratorRole[T_co]

Proxy to typing.AsyncIterator object.

Source code in mode/locals.py

class AsyncIteratorProxy(Proxy[AsyncIterator[T_co]], AsyncIteratorRole[T_co]):
    """Proxy to `typing.AsyncIterator` object."""

AsyncIteratorRole

Bases: AsyncIterator[T_co]

Role/Mixin for typing.AsyncIterator proxy methods.

Source code in mode/locals.py

class AsyncIteratorRole(AsyncIterator[T_co]):
    """Role/Mixin for `typing.AsyncIterator` proxy methods."""

    def _get_iterator(self) -> AsyncIterator[T_co]:
        obj = self._get_current_object()  # type: ignore
        return cast(AsyncIterator[T_co], obj)

    def __aiter__(self) -> AsyncIterator[T_co]:
        return self._get_iterator().__aiter__()

    def __anext__(self) -> Awaitable[T_co]:
        return self._get_iterator().__anext__()

AwaitableProxy

Bases: Proxy[T], AwaitableRole[T]

Proxy to typing.Awaitable object.

Source code in mode/locals.py

class AwaitableProxy(Proxy[T], AwaitableRole[T]):
    """Proxy to `typing.Awaitable` object."""

AwaitableRole

Bases: Awaitable[T]

Role/Mixin for typing.Awaitable proxy methods.

Source code in mode/locals.py

class AwaitableRole(Awaitable[T]):
    """Role/Mixin for `typing.Awaitable` proxy methods."""

    def _get_awaitable(self) -> Awaitable[T]:
        obj = self._get_current_object()  # type: ignore
        return cast(Awaitable[T], obj)

    def __await__(self) -> Generator[Any, None, T]:
        return self._get_awaitable().__await__()

CallableProxy

Bases: Proxy[Callable], CallableRole

Proxy to typing.Callable object.

Source code in mode/locals.py

class CallableProxy(Proxy[Callable], CallableRole):
    """Proxy to `typing.Callable` object."""

CallableRole

Role/Mixin for typing.Callable proxy methods.

Source code in mode/locals.py

class CallableRole:
    """Role/Mixin for `typing.Callable` proxy methods."""

    def _get_callable(self) -> Callable:
        obj = self._get_current_object()  # type: ignore
        return cast(Callable, obj)

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        return self._get_callable()(*args, **kwargs)

ContextManagerProxy

Bases: Proxy[AbstractContextManager[T]], ContextManagerRole[T]

Proxy to contextlib.AbstractContextManager object.

Source code in mode/locals.py

class ContextManagerProxy(
    Proxy[AbstractContextManager[T]], ContextManagerRole[T]
):
    """Proxy to `contextlib.AbstractContextManager` object."""

ContextManagerRole

Bases: AbstractContextManager[T]

Role/Mixin for typing.ContextManager proxy methods.

Source code in mode/locals.py

class ContextManagerRole(AbstractContextManager[T]):
    """Role/Mixin for `typing.ContextManager` proxy methods."""

    def _get_context(self) -> AbstractContextManager[T]:
        obj = self._get_current_object()  # type: ignore
        return cast(AbstractContextManager[T], obj)

    def __enter__(self) -> Any:
        return self._get_context().__enter__()

    def __exit__(self, *exc_info: Any) -> Any:
        return self._get_context().__exit__(*exc_info)

CoroutineProxy

Bases: Proxy[Coroutine[T_co, T_contra, V_co]], CoroutineRole[T_co, T_contra, V_co]

Proxy to typing.Coroutine object.

Source code in mode/locals.py

class CoroutineProxy(
    Proxy[Coroutine[T_co, T_contra, V_co]], CoroutineRole[T_co, T_contra, V_co]
):
    """Proxy to `typing.Coroutine` object."""

CoroutineRole

Bases: Coroutine[T_co, T_contra, V_co]

Role/Mixin for typing.Coroutine proxy methods.

Source code in mode/locals.py

class CoroutineRole(Coroutine[T_co, T_contra, V_co]):
    """Role/Mixin for `typing.Coroutine` proxy methods."""

    def _get_coroutine(self) -> Coroutine[T_co, T_contra, V_co]:
        obj = self._get_current_object()  # type: ignore
        return cast(Coroutine[T_co, T_contra, V_co], obj)

    def __await__(self) -> Generator[Any, None, V_co]:
        return self._get_coroutine().__await__()

    def send(self, value: T_contra) -> T_co:
        return self._get_coroutine().send(value)

    def throw(
        self,
        typ: Union[type[BaseException], BaseException],
        val: object = None,
        tb: Optional[TracebackType] = None,
    ) -> T_co:
        # `Coroutine.throw` is overloaded (exception type plus optional
        # value, or a bare exception instance), and a proxy that forwards
        # whatever it is given matches neither overload exactly.
        throw = cast(Callable[..., T_co], self._get_coroutine().throw)
        return throw(typ, val, tb)

    def close(self) -> None:
        return self._get_coroutine().close()

LocalStack

Bases: Generic[T]

LocalStack.

Manage state per coroutine (also thread safe).

Most famously used probably in Flask to keep track of the current request object.

Source code in mode/utils/locals.py

class LocalStack(Generic[T]):
    """LocalStack.

    Manage state per coroutine (also thread safe).

    Most famously used probably in Flask to keep track of the current
    request object.
    """

    _stack: ContextVar[Optional[list[T]]]

    def __init__(self) -> None:
        self._stack = ContextVar("_stack")

    # XXX mypy bug; when fixed type Generator, should be ContextManager.
    @contextmanager
    def push(self, obj: T) -> Generator[None, None, None]:
        """Push a new item to the stack."""
        self.push_without_automatic_cleanup(obj)
        try:
            yield
        finally:
            self.pop()

    def push_without_automatic_cleanup(self, obj: T) -> None:
        stack = self._stack.get(None)
        if stack is None:
            stack = []
            self._stack.set(stack)
        stack.append(obj)

    def pop(self) -> Optional[T]:
        """Remove the topmost item from the stack.

        Note:
            Will return the old value or `None` if the stack was already empty.
        """
        stack = self._stack.get(None)
        if stack is None:
            return None
        else:
            size = len(stack)
            if not size:
                self._stack.set(None)
                return None
            elif size == 1:
                item = stack[-1]
                self._stack.set(None)
            else:
                item = stack.pop()
            return item

    def __len__(self) -> int:
        stack = self._stack.get(None)
        return len(stack) if stack else 0

    @property
    def stack(self) -> Sequence[T]:
        # read-only version, do not modify
        return self._stack.get(None) or []

    @property
    def top(self) -> Optional[T]:
        """Return the topmost item on the stack.

        Does not remove it from the stack.

        Note:
            If the stack is empty, :const:`None` is returned.
        """
        stack = self._stack.get(None)
        return stack[-1] if stack else None

top property

Return the topmost item on the stack.

Does not remove it from the stack.

Note

If the stack is empty, :const:None is returned.

pop()

Remove the topmost item from the stack.

Note

Will return the old value or None if the stack was already empty.

Source code in mode/utils/locals.py

def pop(self) -> Optional[T]:
    """Remove the topmost item from the stack.

    Note:
        Will return the old value or `None` if the stack was already empty.
    """
    stack = self._stack.get(None)
    if stack is None:
        return None
    else:
        size = len(stack)
        if not size:
            self._stack.set(None)
            return None
        elif size == 1:
            item = stack[-1]
            self._stack.set(None)
        else:
            item = stack.pop()
        return item

push(obj)

Push a new item to the stack.

Source code in mode/utils/locals.py

@contextmanager
def push(self, obj: T) -> Generator[None, None, None]:
    """Push a new item to the stack."""
    self.push_without_automatic_cleanup(obj)
    try:
        yield
    finally:
        self.pop()

MappingProxy

Bases: Proxy[Mapping[KT, VT_co]], MappingRole[KT, VT_co]

Proxy to typing.Mapping object.

Source code in mode/locals.py

class MappingProxy(Proxy[Mapping[KT, VT_co]], MappingRole[KT, VT_co]):
    """Proxy to `typing.Mapping` object."""

MappingRole

Bases: Mapping[KT, VT_co]

Role/Mixin for typing.Mapping proxy methods.

Source code in mode/locals.py

class MappingRole(Mapping[KT, VT_co]):
    """Role/Mixin for `typing.Mapping` proxy methods."""

    def _get_mapping(self) -> Mapping[KT, VT_co]:
        obj = self._get_current_object()  # type: ignore
        return cast(Mapping[KT, VT_co], obj)

    def __getitem__(self, key: KT) -> VT_co:
        return self._get_mapping().__getitem__(key)

    @overload
    def get(self, k: KT) -> Optional[VT_co]: ...

    @overload
    def get(self, k: KT, default: Union[VT_co, T]) -> Union[VT_co, T]: ...

    def get(self, *args: Any, **kwargs: Any) -> Any:
        return self._get_mapping().get(*args, **kwargs)

    def items(self) -> ItemsView[KT, VT_co]:
        return self._get_mapping().items()

    def keys(self) -> KeysView[KT]:
        return self._get_mapping().keys()

    def values(self) -> ValuesView[VT_co]:
        return self._get_mapping().values()

    def __contains__(self, o: Any) -> bool:
        return self._get_mapping().__contains__(o)

    def __iter__(self) -> Iterator[KT]:
        return self._get_mapping().__iter__()

    def __len__(self) -> int:
        return self._get_mapping().__len__()

MutableMappingProxy

Bases: Proxy[MutableMapping[KT, VT]], MutableMappingRole[KT, VT]

Proxy to typing.MutableMapping object.

Source code in mode/locals.py

class MutableMappingProxy(
    Proxy[MutableMapping[KT, VT]], MutableMappingRole[KT, VT]
):
    """Proxy to `typing.MutableMapping` object."""

MutableMappingRole

Bases: MappingRole[KT, VT], MutableMapping[KT, VT]

Role/Mixin for typing.MutableMapping proxy methods.

Source code in mode/locals.py

class MutableMappingRole(MappingRole[KT, VT], MutableMapping[KT, VT]):
    """Role/Mixin for `typing.MutableMapping` proxy methods."""

    def _get_mapping(self) -> MutableMapping[KT, VT]:
        obj = self._get_current_object()  # type: ignore
        return cast(MutableMapping[KT, VT], obj)

    def __setitem__(self, key: KT, value: VT) -> None:
        self._get_mapping().__setitem__(key, value)

    def __delitem__(self, key: KT) -> None:
        self._get_mapping().__delitem__(key)

    def clear(self) -> None:
        self._get_mapping().clear()

    @overload
    def pop(self, k: KT) -> VT: ...

    @overload
    def pop(self, k: KT, default: Union[VT, T] = ...) -> Union[VT, T]: ...

    def pop(self, *args: Any, **kwargs: Any) -> Any:
        return self._get_mapping().pop(*args, **kwargs)

    def popitem(self) -> tuple[KT, VT]:
        return self._get_mapping().popitem()

    def setdefault(self, k: KT, *args: Any) -> VT:
        return self._get_mapping().setdefault(k, *args)

    # Mirrors `MutableMapping.update` in typeshed, minus the overloads
    # whose `self:` annotation restricts `**kwargs` to str-keyed mappings
    # -- an overload implementation cannot satisfy those.
    @overload
    def update(self, m: "SupportsKeysAndGetItem[KT, VT]", /) -> None: ...

    @overload
    def update(self, m: Iterable[tuple[KT, VT]], /) -> None: ...

    @overload
    def update(self, **kwargs: VT) -> None: ...

    def update(self, *args: Any, **kwargs: Any) -> None:
        self._get_mapping().update(*args, **kwargs)

MutableSequenceProxy

Bases: Proxy[MutableSequence[T]], MutableSequenceRole[T]

Proxy to typing.MutableSequence object.

Source code in mode/locals.py

class MutableSequenceProxy(Proxy[MutableSequence[T]], MutableSequenceRole[T]):
    """Proxy to `typing.MutableSequence` object."""

MutableSequenceRole

Bases: SequenceRole[T], MutableSequence[T]

Role/Mixin for typing.MutableSequence proxy methods.

Source code in mode/locals.py

class MutableSequenceRole(SequenceRole[T], MutableSequence[T]):
    """Role/Mixin for `typing.MutableSequence` proxy methods."""

    def _get_sequence(self) -> MutableSequence[T]:
        obj = self._get_current_object()  # type: ignore
        return cast(MutableSequence[T], obj)

    def insert(self, index: int, object: T) -> None:
        self._get_sequence().insert(index, object)

    @overload
    def __setitem__(self, s: int, o: T) -> None: ...

    @overload
    def __setitem__(self, s: slice, o: Iterable[T]) -> None: ...

    def __setitem__(self, s: Any, o: Any) -> None:
        self._get_sequence().__setitem__(s, o)

    @overload
    def __delitem__(self, i: int) -> None: ...

    @overload
    def __delitem__(self, i: slice) -> None: ...

    def __delitem__(self, i: Any) -> None:
        self._get_sequence().__delitem__(i)

    def append(self, obj: T) -> None:
        self._get_sequence().append(obj)

    def extend(self, iterable: Iterable[T]) -> None:
        self._get_sequence().extend(iterable)

    def reverse(self) -> None:
        self._get_sequence().reverse()

    def pop(self, *args: Any) -> Any:
        return self._get_sequence().pop(*args)

    def remove(self, object: T) -> None:
        self._get_sequence().remove(object)

    def __iadd__(self, x: Iterable[T]) -> "MutableSequenceRole[T]":
        return cast("MutableSequenceRole[T]", self._get_sequence().__iadd__(x))

MutableSetProxy

Bases: Proxy[MutableSet[T]], MutableSetRole[T]

Proxy to typing.MutableSet object.

Source code in mode/locals.py

class MutableSetProxy(Proxy[MutableSet[T]], MutableSetRole[T]):
    """Proxy to `typing.MutableSet` object."""

MutableSetRole

Bases: SetRole[T], MutableSet[T]

Role/Mixin for typing.MutableSet proxy methods.

Source code in mode/locals.py

class MutableSetRole(SetRole[T], MutableSet[T]):
    """Role/Mixin for `typing.MutableSet` proxy methods."""

    def _get_set(self) -> MutableSet[T]:
        obj = self._get_current_object()  # type: ignore
        return cast(MutableSet[T], obj)

    def add(self, x: T) -> None:
        self._get_set().add(x)

    def discard(self, x: T) -> None:
        self._get_set().discard(x)

    def clear(self) -> None:
        self._get_set().clear()

    def pop(self) -> T:
        return self._get_set().pop()

    def remove(self, element: T) -> None:
        self._get_set().remove(element)

    def __ior__(self, s: Set[S]) -> "MutableSetRole[Union[T, S]]":
        data = cast(MutableSet[Union[T, S]], self._get_set())
        return cast("MutableSetRole[Union[T, S]]", data.__ior__(s))

    def __iand__(self, s: Set[Any]) -> "MutableSetRole[T]":
        return cast("MutableSetRole[T]", self._get_set().__iand__(s))

    def __ixor__(self, s: Set[S]) -> "MutableSetRole[Union[T, S]]":
        data = cast(MutableSet[Union[T, S]], self._get_set())
        return cast("MutableSetRole[Union[T, S]]", data.__ixor__(s))

    def __isub__(self, s: Set[Any]) -> "MutableSetRole[T]":
        return cast("MutableSetRole[T]", self._get_set().__isub__(s))

Proxy

Bases: Generic[T]

Proxy to another object.

Source code in mode/locals.py

class Proxy(Generic[T]):
    """Proxy to another object."""

    __proxy_source__: ClassVar[Optional[type[T]]] = None

    # Code initially stolen from werkzeug.local.Proxy.
    if not SLOTS_ISSUE_PRESENT and not PYPY:  # pragma: no cover
        __slots__ = (
            "__args",
            "__dict__",
            "__finalizers",
            "__kwargs",
            "__local",
        )

    def __init_subclass__(self, source: Optional[type[T]] = None) -> None:
        # NOTE: Delegated to a module-level helper on purpose -- do not
        # inline this back to `super().__init_subclass__()`.  Naming `super`
        # anywhere in this class body makes the compiler add an implicit
        # `__class__` closure cell, which breaks the `__class__` property
        # below on PyPy.  See `_cooperative_init_subclass`.
        _cooperative_init_subclass(self)
        if source is not None:
            self._init_from_source(source)
        elif self.__proxy_source__ is not None:
            # __proxy_source__ must be used on Python < 3.7
            # to work around https://bugs.python.org/issue29581
            self._init_from_source(self.__proxy_source__)

    @classmethod
    def _init_from_source(cls, source: type[T]) -> None:
        # source must have metaclass ABCMeta
        abstractmethods = getattr(source, "__abstractmethods__", None)
        if abstractmethods is None:
            raise TypeError("class is not using metaclass ABCMeta")
        for method_name in abstractmethods:
            setattr(
                cls,
                method_name,
                cls._generate_proxy_method(source, method_name),
            )

    @classmethod
    def _generate_proxy_method(
        cls, source: type[T], method_name: str
    ) -> Callable:
        @wraps(getattr(source, method_name))
        def _classmethod(self: Proxy[T], *args: Any, **kwargs: Any) -> Any:
            obj = self._get_current_object()
            return getattr(obj, method_name)(*args, **kwargs)

        _classmethod.__isabstractmethod__ = False  # type: ignore

        return _classmethod

    def __init__(
        self,
        local: Callable[..., T],
        args: Optional[tuple] = None,
        kwargs: Optional[dict] = None,
        name: Optional[str] = None,
        cache: bool = False,
        __doc__: Optional[str] = None,
    ) -> None:
        object.__setattr__(self, "_Proxy__local", local)
        object.__setattr__(self, "_Proxy__args", args or ())
        object.__setattr__(self, "_Proxy__kwargs", kwargs or {})
        object.__setattr__(self, "_Proxy__cached", cache)
        object.__setattr__(self, "_Proxy__finalizers", deque())
        if name is not None:
            object.__setattr__(self, "__custom_name__", name)
        if __doc__ is not None:
            object.__setattr__(self, "__doc__", __doc__)

    def _add_proxy_finalizer(self, fun: "Proxy") -> None:
        finalizers = object.__getattribute__(self, "_Proxy__finalizers")
        finalizers.append(fun)

    def _call_proxy_finalizers(self) -> None:
        finalizers = object.__getattribute__(self, "_Proxy__finalizers")
        while finalizers:
            finalizer = finalizers.popleft()
            finalizer._get_current_object()  # evaluate

    @_default_cls_attr("name", str, __name__)
    @no_type_check
    def __name__(self) -> str:
        try:
            return self.__custom_name__
        except AttributeError:
            return self._get_current_object().__name__

    @_default_cls_attr("module", str, __name__)
    @no_type_check
    def __module__(self) -> str:
        return self._get_current_object().__module__

    @_default_cls_attr("doc", str, __doc__)
    @no_type_check
    def __doc__(self) -> Optional[str]:
        return cast(str, self._get_current_object().__doc__)

    def _get_class(self) -> type[T]:
        return self._get_current_object().__class__

    # NOTE: This ordinary property spelling is only safe while the class
    # body has no implicit `__class__` closure cell -- see
    # `_cooperative_init_subclass` before adding any use of `super` here.
    @property
    def __class__(self) -> Any:
        return self._get_class()

    @__class__.setter
    def __class__(self, t: type) -> None:
        raise NotImplementedError()

    def _get_current_object(self) -> T:
        """Get current object.

        This is useful if you want the real
        object behind the proxy at a time for performance reasons or because
        you want to pass the object into a different context.
        """
        try:
            return cast(T, object.__getattribute__(self, "__cache"))
        except AttributeError:
            return self.__evaluate__()

    def __evaluate__(
        self,
        _clean: tuple[str, ...] = (
            "_Proxy__local",
            "_Proxy__args",
            "_Proxy__kwargs",
        ),
    ) -> T:
        thing = self._evaluate_proxy()
        cached = object.__getattribute__(self, "_Proxy__cached")
        if cached:
            object.__setattr__(self, "__cache", thing)
            for attr in _clean:
                try:
                    object.__delattr__(self, attr)
                except AttributeError:  # pragma: no cover
                    # May mask errors so ignore
                    pass
        return thing

    def _evaluate_proxy(self) -> T:
        self._call_proxy_finalizers()
        loc = object.__getattribute__(self, "_Proxy__local")
        if not hasattr(loc, "__release_local__"):
            return cast(T, loc(*self.__args, **self.__kwargs))
        try:  # pragma: no cover
            # not sure what this is about
            return cast(T, getattr(loc, self.__name__))
        except AttributeError as err:  # pragma: no cover
            raise RuntimeError(f"no object bound to {self.__name__}") from err

    def __evaluated__(self) -> bool:
        try:
            object.__getattribute__(self, "__cache")
        except AttributeError:
            return False
        return True

    def __maybe_evaluate__(self) -> T:
        return self._get_current_object()

    @property
    def __dict__(self) -> dict[str, Any]:  # type: ignore
        try:
            return self._get_current_object().__dict__
        except RuntimeError as err:  # pragma: no cover
            raise AttributeError("__dict__") from err

    def __repr__(self) -> str:
        try:
            obj = self._get_current_object()
        except RuntimeError:  # pragma: no cover
            return f"<{self.__class__.__name__} unbound>"
        return repr(obj)

    def __bool__(self) -> bool:
        try:
            return bool(self._get_current_object())
        except RuntimeError:  # pragma: no cover
            return False

    __nonzero__ = __bool__  # Py2

    def __dir__(self) -> list[str]:
        try:
            return dir(self._get_current_object())
        except RuntimeError:  # pragma: no cover
            return []

    def __getattr__(self, name: str) -> Any:
        if name == "__members__":
            return dir(self._get_current_object())
        return getattr(self._get_current_object(), name)

    def __eq__(self, other: Any) -> Any:
        return self._get_current_object() == other

    def __ne__(self, other: Any) -> Any:
        return self._get_current_object() != other

    def __setattr__(self, name: str, value: Any) -> None:
        setattr(self._get_current_object(), name, value)

    def __delattr__(self, name: str) -> None:
        delattr(self._get_current_object(), name)

    def __str__(self) -> str:
        return str(self._get_current_object())

    def __hash__(self) -> int:
        return hash(self._get_current_object())

    def __reduce__(self) -> tuple:
        return cast(tuple, self._get_current_object().__reduce__())

SequenceProxy

Bases: Proxy[Sequence[T_co]], SequenceRole[T_co]

Proxy to typing.Sequence object.

Source code in mode/locals.py

class SequenceProxy(Proxy[Sequence[T_co]], SequenceRole[T_co]):
    """Proxy to `typing.Sequence` object."""

SequenceRole

Bases: Sequence[T_co]

Role/Mixin for typing.Sequence proxy methods.

Source code in mode/locals.py

class SequenceRole(Sequence[T_co]):
    """Role/Mixin for `typing.Sequence` proxy methods."""

    def _get_sequence(self) -> Sequence[T_co]:
        obj = self._get_current_object()  # type: ignore
        return cast(Sequence[T_co], obj)

    @overload
    def __getitem__(self, s: int) -> T_co: ...

    @overload
    def __getitem__(self, s: slice) -> MutableSequence[T_co]: ...

    def __getitem__(self, s: Any) -> Any:
        return self._get_sequence().__getitem__(s)

    def index(self, x: Any, *args: Any, **kwargs: Any) -> int:
        return self._get_sequence().index(x, *args, **kwargs)

    def count(self, x: Any) -> int:
        return self._get_sequence().count(x)

    def __contains__(self, x: Any) -> bool:
        return self._get_sequence().__contains__(x)

    def __iter__(self) -> Iterator[T_co]:
        return self._get_sequence().__iter__()

    def __reversed__(self) -> Iterator[T_co]:
        return self._get_sequence().__reversed__()

    def __len__(self) -> int:
        return self._get_sequence().__len__()

SetProxy

Bases: Proxy[Set[T_co]], SetRole[T_co]

Proxy to collections.abc.Set object.

Source code in mode/locals.py

class SetProxy(Proxy[Set[T_co]], SetRole[T_co]):
    """Proxy to `collections.abc.Set` object."""

SetRole

Bases: Set[T_co]

Role/Mixin for collections.abc.Set proxy methods.

Source code in mode/locals.py

class SetRole(Set[T_co]):
    """Role/Mixin for `collections.abc.Set` proxy methods."""

    def _get_set(self) -> Set[T_co]:
        obj = self._get_current_object()  # type: ignore
        return cast(Set[T_co], obj)

    def __le__(self, s: Set[Any]) -> bool:
        return self._get_set().__le__(s)

    def __lt__(self, s: Set[Any]) -> bool:
        return self._get_set().__lt__(s)

    def __gt__(self, s: Set[Any]) -> bool:
        return self._get_set().__gt__(s)

    def __ge__(self, s: Set[Any]) -> bool:
        return self._get_set().__ge__(s)

    def __and__(self, s: Set[Any]) -> Set[T_co]:
        return self._get_set().__and__(s)

    def __or__(self, s: Set[T]) -> Set[Union[T_co, T]]:
        return self._get_set().__or__(s)

    def __sub__(self, s: Set[Any]) -> Set[T_co]:
        return self._get_set().__sub__(s)

    def __xor__(self, s: Set[T]) -> Set[Union[T_co, T]]:
        return self._get_set().__xor__(s)

    def isdisjoint(self, s: Iterable[Any]) -> bool:
        return self._get_set().isdisjoint(s)

    def __contains__(self, x: Any) -> bool:
        return self._get_set().__contains__(x)

    def __iter__(self) -> Iterator[T_co]:
        return self._get_set().__iter__()

    def __len__(self) -> int:
        return self._get_set().__len__()

maybe_evaluate(obj)

Attempt to evaluate promise, even if obj is not a promise.

Source code in mode/locals.py

def maybe_evaluate(obj: Any) -> Any:
    """Attempt to evaluate promise, even if obj is not a promise."""
    try:
        return obj.__maybe_evaluate__()
    except AttributeError:
        return obj