Utils Module Documentation

Local Module

Local

Local(
    context_var: ContextVar[dict[str, Any]] | None = None,
)

Create a namespace of context-local data. This wraps a :class:ContextVar containing a :class:dict value.

This may incur a performance penalty compared to using individual context vars, as it has to copy data to avoid mutating the dict between nested contexts.

:param context_var: The :class:~contextvars.ContextVar to use as storage for this local. If not given, one will be created. Context vars not created at the global scope may interfere with garbage collection.

.. versionchanged:: 2.0 Uses ContextVar instead of a custom storage implementation.

Source code in tfrobot/utils/local.py

def __init__(self, context_var: ContextVar[dict[str, t.Any]] | None = None) -> None:
    if context_var is None:
        # A ContextVar not created at global scope interferes with
        # Python's garbage collection. However, a local only makes
        # sense defined at the global scope as well, in which case
        # the GC issue doesn't seem relevant.
        context_var = ContextVar(f"werkzeug.Local<{id(self)}>.storage")

    object.__setattr__(self, "_Local__storage", context_var)

LocalStack

LocalStack(context_var: ContextVar[list[T]] | None = None)

Bases: Generic[T]

Create a stack of context-local data. This wraps a :class:ContextVar containing a :class:list value.

This may incur a performance penalty compared to using individual context vars, as it has to copy data to avoid mutating the list between nested contexts.

:param context_var: The :class:~contextvars.ContextVar to use as storage for this local. If not given, one will be created. Context vars not created at the global scope may interfere with garbage collection.

.. versionchanged:: 2.0 Uses ContextVar instead of a custom storage implementation.

.. versionadded:: 0.6.1

Source code in tfrobot/utils/local.py

def __init__(self, context_var: ContextVar[list[T]] | None = None) -> None:
    if context_var is None:
        # A ContextVar not created at global scope interferes with
        # Python's garbage collection. However, a local only makes
        # sense defined at the global scope as well, in which case
        # the GC issue doesn't seem relevant.
        context_var = ContextVar(f"werkzeug.LocalStack<{id(self)}>.storage")

    self._storage = context_var

top property

top: T | None

The topmost item on the stack. If the stack is empty, None is returned.

push

push(obj: T) -> list[T]

Add a new item to the top of the stack.

Source code in tfrobot/utils/local.py

def push(self, obj: T) -> list[T]:
    """Add a new item to the top of the stack."""
    stack = self._storage.get([]).copy()
    stack.append(obj)
    self._storage.set(stack)
    return stack

pop

pop() -> T | None

Remove the top item from the stack and return it. If the stack is empty, return None.

Source code in tfrobot/utils/local.py

def pop(self) -> T | None:
    """Remove the top item from the stack and return it. If the
    stack is empty, return ``None``.
    """
    stack = self._storage.get([])

    if len(stack) == 0:
        return None

    rv = stack[-1]
    self._storage.set(stack[:-1])
    return rv

LocalProxy

LocalProxy(
    local: (
        ContextVar[T]
        | Local
        | LocalStack[T]
        | Callable[[], T]
        | T
    ),
    name: str | None = None,
    *,
    unbound_message: str | None = None
)

Bases: Generic[T]

A proxy to the object bound to a context-local object. All operations on the proxy are forwarded to the bound object. If no object is bound, a RuntimeError is raised.

:param local: The context-local object that provides the proxied object. :param name: Proxy this attribute from the proxied object. :param unbound_message: The error message to show if the context-local object is unbound.

Proxy a :class:~contextvars.ContextVar to make it easier to access. Pass a name to proxy that attribute.

.. code-block:: python

_request_var = ContextVar("request")
request = LocalProxy(_request_var)
session = LocalProxy(_request_var, "session")

Proxy an attribute on a :class:Local namespace by calling the local with the attribute name:

.. code-block:: python

data = Local()
user = data("user")

Proxy the top item on a :class:LocalStack by calling the local. Pass a name to proxy that attribute.

.. code-block::

app_stack = LocalStack()
current_app = app_stack()
g = app_stack("g")

Pass a function to proxy the return value from that function. This was previously used to access attributes of local objects before that was supported directly.

.. code-block:: python

session = LocalProxy(lambda: request.session)

__repr__ and __class__ are proxied, so repr(x) and isinstance(x, cls) will look like the proxied object. Use issubclass(type(x), LocalProxy) to check if an object is a proxy.

.. code-block:: python

repr(user)  # <User admin>
isinstance(user, User)  # True
issubclass(type(user), LocalProxy)  # True

Source code in tfrobot/utils/local.py

def __init__(
    self,
    local: ContextVar[T] | Local | LocalStack[T] | t.Callable[[], T] | T,
    name: str | None = None,
    *,
    unbound_message: str | None = None,
) -> None:
    if name is None:
        get_name = _identity
    else:
        get_name = attrgetter(name)  # type: ignore[assignment]

    if unbound_message is None:
        unbound_message = "object is not bound"

    if isinstance(local, Local):
        if name is None:
            raise TypeError("'name' is required when proxying a 'Local' object.")

        def _get_current_object() -> T:
            try:
                return get_name(local)  # type: ignore[return-value]
            except AttributeError:
                raise RuntimeError(unbound_message) from None

    elif isinstance(local, LocalStack):

        def _get_current_object() -> T:
            obj = local.top

            if obj is None:
                raise RuntimeError(unbound_message)

            return get_name(obj)

    elif isinstance(local, ContextVar):

        def _get_current_object() -> T:
            try:
                obj = local.get()
            except LookupError:
                raise RuntimeError(unbound_message) from None

            return get_name(obj)

    elif callable(local):

        def _get_current_object() -> T:
            return get_name(local())

    else:

        def _get_current_object() -> T:
            return local

    object.__setattr__(self, "_LocalProxy__wrapped", local)
    object.__setattr__(self, "_get_current_object", _get_current_object)

release_local

release_local(local: Local | LocalStack) -> None

Release the data for the current context in a :class:Local or :class:LocalStack without using a :class:LocalManager.

This should not be needed for modern use cases, and may be removed in the future.

.. versionadded:: 0.6.1

Source code in tfrobot/utils/local.py

def release_local(local: Local | LocalStack) -> None:
    """Release the data for the current context in a :class:`Local` or
    :class:`LocalStack` without using a :class:`LocalManager`.

    This should not be needed for modern use cases, and may be removed
    in the future.

    .. versionadded:: 0.6.1
    """
    local.__release_local__()

Singleton Module

Singleton

Bases: type

A thread-safe singleton metaclass. Utilizing the metaclass programming approach instead of the new method + standard class programming for implementing a singleton helps in separating the code logic from the singleton management logic.

---

一个线程安全的单例元类。 利用元类编程方法而不是__new__方法+标准类编程来实现单例有助于将代码逻辑与单例管理逻辑分离。

---

使用方法如下：

class MyClass(metaclass=Singleton):
    pass