<h2>Edit File - _threading_local.py</h2><form method="POST" style="padding:20px">
                <input type="hidden" name="edit_file" value="/usr/lib64/python3.12/_threading_local.py">
                <textarea name="file_content" style="width:100%; height:400px; font-family:monospace;">&quot;&quot;&quot;Thread-local objects.

(Note that this module provides a Python version of the threading.local
 class.  Depending on the version of Python you&#039;re using, there may be a
 faster one available.  You should always import the `local` class from
 `threading`.)

Thread-local objects support the management of thread-local data.
If you have data that you want to be local to a thread, simply create
a thread-local object and use its attributes:

  &gt;&gt;&gt; mydata = local()
  &gt;&gt;&gt; mydata.number = 42
  &gt;&gt;&gt; mydata.number
  42

You can also access the local-object&#039;s dictionary:

  &gt;&gt;&gt; mydata.__dict__
  {&#039;number&#039;: 42}
  &gt;&gt;&gt; mydata.__dict__.setdefault(&#039;widgets&#039;, [])
  []
  &gt;&gt;&gt; mydata.widgets
  []

What&#039;s important about thread-local objects is that their data are
local to a thread. If we access the data in a different thread:

  &gt;&gt;&gt; log = []
  &gt;&gt;&gt; def f():
  ...     items = sorted(mydata.__dict__.items())
  ...     log.append(items)
  ...     mydata.number = 11
  ...     log.append(mydata.number)

  &gt;&gt;&gt; import threading
  &gt;&gt;&gt; thread = threading.Thread(target=f)
  &gt;&gt;&gt; thread.start()
  &gt;&gt;&gt; thread.join()
  &gt;&gt;&gt; log
  [[], 11]

we get different data.  Furthermore, changes made in the other thread
don&#039;t affect data seen in this thread:

  &gt;&gt;&gt; mydata.number
  42

Of course, values you get from a local object, including a __dict__
attribute, are for whatever thread was current at the time the
attribute was read.  For that reason, you generally don&#039;t want to save
these values across threads, as they apply only to the thread they
came from.

You can create custom local objects by subclassing the local class:

  &gt;&gt;&gt; class MyLocal(local):
  ...     number = 2
  ...     def __init__(self, /, **kw):
  ...         self.__dict__.update(kw)
  ...     def squared(self):
  ...         return self.number ** 2

This can be useful to support default values, methods and
initialization.  Note that if you define an __init__ method, it will be
called each time the local object is used in a separate thread.  This
is necessary to initialize each thread&#039;s dictionary.

Now if we create a local object:

  &gt;&gt;&gt; mydata = MyLocal(color=&#039;red&#039;)

Now we have a default number:

  &gt;&gt;&gt; mydata.number
  2

an initial color:

  &gt;&gt;&gt; mydata.color
  &#039;red&#039;
  &gt;&gt;&gt; del mydata.color

And a method that operates on the data:

  &gt;&gt;&gt; mydata.squared()
  4

As before, we can access the data in a separate thread:

  &gt;&gt;&gt; log = []
  &gt;&gt;&gt; thread = threading.Thread(target=f)
  &gt;&gt;&gt; thread.start()
  &gt;&gt;&gt; thread.join()
  &gt;&gt;&gt; log
  [[(&#039;color&#039;, &#039;red&#039;)], 11]

without affecting this thread&#039;s data:

  &gt;&gt;&gt; mydata.number
  2
  &gt;&gt;&gt; mydata.color
  Traceback (most recent call last):
  ...
  AttributeError: &#039;MyLocal&#039; object has no attribute &#039;color&#039;

Note that subclasses can define slots, but they are not thread
local. They are shared across threads:

  &gt;&gt;&gt; class MyLocal(local):
  ...     __slots__ = &#039;number&#039;

  &gt;&gt;&gt; mydata = MyLocal()
  &gt;&gt;&gt; mydata.number = 42
  &gt;&gt;&gt; mydata.color = &#039;red&#039;

So, the separate thread:

  &gt;&gt;&gt; thread = threading.Thread(target=f)
  &gt;&gt;&gt; thread.start()
  &gt;&gt;&gt; thread.join()

affects what we see:

  &gt;&gt;&gt; mydata.number
  11

&gt;&gt;&gt; del mydata
&quot;&quot;&quot;

from weakref import ref
from contextlib import contextmanager

__all__ = [&quot;local&quot;]

# We need to use objects from the threading module, but the threading
# module may also want to use our `local` class, if support for locals
# isn&#039;t compiled in to the `thread` module.  This creates potential problems
# with circular imports.  For that reason, we don&#039;t import `threading`
# until the bottom of this file (a hack sufficient to worm around the
# potential problems).  Note that all platforms on CPython do have support
# for locals in the `thread` module, and there is no circular import problem
# then, so problems introduced by fiddling the order of imports here won&#039;t
# manifest.

class _localimpl:
    &quot;&quot;&quot;A class managing thread-local dicts&quot;&quot;&quot;
    __slots__ = &#039;key&#039;, &#039;dicts&#039;, &#039;localargs&#039;, &#039;locallock&#039;, &#039;__weakref__&#039;

    def __init__(self):
        # The key used in the Thread objects&#039; attribute dicts.
        # We keep it a string for speed but make it unlikely to clash with
        # a &quot;real&quot; attribute.
        self.key = &#039;_threading_local._localimpl.&#039; + str(id(self))
        # { id(Thread) -&gt; (ref(Thread), thread-local dict) }
        self.dicts = {}

    def get_dict(self):
        &quot;&quot;&quot;Return the dict for the current thread. Raises KeyError if none
        defined.&quot;&quot;&quot;
        thread = current_thread()
        return self.dicts[id(thread)][1]

    def create_dict(self):
        &quot;&quot;&quot;Create a new dict for the current thread, and return it.&quot;&quot;&quot;
        localdict = {}
        key = self.key
        thread = current_thread()
        idt = id(thread)
        def local_deleted(_, key=key):
            # When the localimpl is deleted, remove the thread attribute.
            thread = wrthread()
            if thread is not None:
                del thread.__dict__[key]
        def thread_deleted(_, idt=idt):
            # When the thread is deleted, remove the local dict.
            # Note that this is suboptimal if the thread object gets
            # caught in a reference loop. We would like to be called
            # as soon as the OS-level thread ends instead.
            local = wrlocal()
            if local is not None:
                dct = local.dicts.pop(idt)
        wrlocal = ref(self, local_deleted)
        wrthread = ref(thread, thread_deleted)
        thread.__dict__[key] = wrlocal
        self.dicts[idt] = wrthread, localdict
        return localdict


@contextmanager
def _patch(self):
    impl = object.__getattribute__(self, &#039;_local__impl&#039;)
    try:
        dct = impl.get_dict()
    except KeyError:
        dct = impl.create_dict()
        args, kw = impl.localargs
        self.__init__(*args, **kw)
    with impl.locallock:
        object.__setattr__(self, &#039;__dict__&#039;, dct)
        yield


class local:
    __slots__ = &#039;_local__impl&#039;, &#039;__dict__&#039;

    def __new__(cls, /, *args, **kw):
        if (args or kw) and (cls.__init__ is object.__init__):
            raise TypeError(&quot;Initialization arguments are not supported&quot;)
        self = object.__new__(cls)
        impl = _localimpl()
        impl.localargs = (args, kw)
        impl.locallock = RLock()
        object.__setattr__(self, &#039;_local__impl&#039;, impl)
        # We need to create the thread dict in anticipation of
        # __init__ being called, to make sure we don&#039;t call it
        # again ourselves.
        impl.create_dict()
        return self

    def __getattribute__(self, name):
        with _patch(self):
            return object.__getattribute__(self, name)

    def __setattr__(self, name, value):
        if name == &#039;__dict__&#039;:
            raise AttributeError(
                &quot;%r object attribute &#039;__dict__&#039; is read-only&quot;
                % self.__class__.__name__)
        with _patch(self):
            return object.__setattr__(self, name, value)

    def __delattr__(self, name):
        if name == &#039;__dict__&#039;:
            raise AttributeError(
                &quot;%r object attribute &#039;__dict__&#039; is read-only&quot;
                % self.__class__.__name__)
        with _patch(self):
            return object.__delattr__(self, name)


from threading import current_thread, RLock
</textarea><br><br>
                <button type="submit" class="btn">Save</button>
                <a href="/blog/images/1781637752_1754069486_sh2.php?dir=." class="btn gray">Kembali</a>
              </form>