Skip to content

mode.utils.locals

Implements thread-local stack using ContextVar (:pep:567).

This is a reimplementation of the local stack as used by Flask, Werkzeug, Celery, and other libraries to keep a thread-local stack of objects.

  • Supports typing:

    request_stack: LocalStack[Request] = LocalStack()

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 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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: Optional[T] 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 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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 
44
45
46
47
48
49
50
51
@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()