Skip to content

deep.thread_local

Provide type to store data based on the calling thread.

ThreadLocal

Bases: Generic[T]

This type offers the ability to store a value based on the thread that accessed the value.

Source code in deep/thread_local.py 
25
26
27
28
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
class ThreadLocal(Generic[T]):
    """This type offers the ability to store a value based on the thread that accessed the value."""

    __store = {}

    def __init__(self, default_provider: Callable[[], T] = lambda: None):
        """
        Create a new ThreadLocal value.

        :param default_provider: a provider that will produce a default value
        """
        self.__default_provider = default_provider

    def get(self) -> T:
        """
        Get the value stored for the calling thread.

        :return: the stored value, or the value from the default_provider
        """
        current_thread = threading.current_thread()
        get = self.__store.get(current_thread.ident, None)
        if get is None:
            get = self.__default_provider()
            self.__store[current_thread.ident] = get
        return get

    def set(self, val: T):
        """
        Set the value to get stored.

        :param val: the value to store
        """
        current_thread = threading.current_thread()
        self.__store[current_thread.ident] = val

    def clear(self):
        """Remove the value for this thread."""
        current_thread = threading.current_thread()
        if current_thread.ident in self.__store:
            del self.__store[current_thread.ident]

    @property
    def is_set(self):
        """
        Check if the value is set for this thread.

        :return: True if there is a value for this thread
        """
        current_thread = threading.current_thread()
        return current_thread.ident in self.__store

    @property
    def value(self):
        """
        Get the value stored for the calling thread.

        :return: the stored value, or the value from the default_provider
        """
        return self.get()

    @value.setter
    def value(self, value):
        """
        Set the value to get stored.

        :param value: the value to store
        """
        self.set(value)

is_set property

Check if the value is set for this thread.

:return: True if there is a value for this thread

value property writable

Get the value stored for the calling thread.

:return: the stored value, or the value from the default_provider

__init__(default_provider=lambda: None)

Create a new ThreadLocal value.

:param default_provider: a provider that will produce a default value

Source code in deep/thread_local.py 
30
31
32
33
34
35
36
def __init__(self, default_provider: Callable[[], T] = lambda: None):
    """
    Create a new ThreadLocal value.

    :param default_provider: a provider that will produce a default value
    """
    self.__default_provider = default_provider

clear()

Remove the value for this thread.

Source code in deep/thread_local.py 
60
61
62
63
64
def clear(self):
    """Remove the value for this thread."""
    current_thread = threading.current_thread()
    if current_thread.ident in self.__store:
        del self.__store[current_thread.ident]

get()

Get the value stored for the calling thread.

:return: the stored value, or the value from the default_provider

Source code in deep/thread_local.py 
38
39
40
41
42
43
44
45
46
47
48
49
def get(self) -> T:
    """
    Get the value stored for the calling thread.

    :return: the stored value, or the value from the default_provider
    """
    current_thread = threading.current_thread()
    get = self.__store.get(current_thread.ident, None)
    if get is None:
        get = self.__default_provider()
        self.__store[current_thread.ident] = get
    return get

set(val)

Set the value to get stored.

:param val: the value to store

Source code in deep/thread_local.py 
51
52
53
54
55
56
57
58
def set(self, val: T):
    """
    Set the value to get stored.

    :param val: the value to store
    """
    current_thread = threading.current_thread()
    self.__store[current_thread.ident] = val