A naive implementation of thread-local space in pure Python

thread_local.py

from threading import RLock, get_native_id


class ThreadSpace:
    """
    A naive implementation of thread-local
    space in pure Python. All operations on
    this object are thread-safe as they internally
    acquire and release a thread lock when needed

    How to use this class (if you really want to, but please don't):
    >>> space = ThreadSpace()
    >>> space.number = 1

    Dict-like usage is supported!
    >>> space["fruits"] = ["banana", "apple"]

    Entries can also be deleted!
    >>> del space["fruits"]
    Or...
    >>> del space.fruits

    We can check that the entries got deleted with the exists() method
    >>> space.exists("fruits")
    False
    >>> space.exists("number")
    True

    This will error out because it's in a different thread
    >>> Thread(target=lambda: print(space.number)).start()
    AttributeError: object of type ThreadSpace has no attribute 'number'

    P.S.: This is a slower, simpler (and probably buggier) version of the native
    implementation already baked into most CPython versions as the threading.local
    class. Furthermore, when a faster C alternative isn't available Python will
    fallback to a pure Python one (located at cpython/_threading_local.py),
    effectively making this class useless. But hey, at least it looks cool!

    P.P.S.: This class doesn't take into account the possibility of thread identifiers
    being recycled by the OS. If a thread exits, another one might be assigned
    its identifier, causing some weird bugs. You've been warned!

    P.P.P.S.: Don't overwrite self._storage and/or self._lock! They're what makes
    this whole machinery work
    """

    def __init__(self):
        """
        Object constructor
        """

        self._storage = {}
        self._lock = RLock()

    def __setattr__(self, name: object, val: object):
        """
        Implements self.name = val (with some hacks)
        """

        if name not in ("_storage", "_lock"):
            with self._lock:
                if (ident := get_native_id()) not in self._storage:
                    self._storage[ident] = {}
                self._storage[ident][name] = val
        else:
          super().__setattr__(name, val)

    def __getattr__(self, name: object):
        """
        Implements dot accessing with self.name
        (with some hacks to work as intended even
        internally)
        """

        with super().__getattribute__("_lock"):
            if (ident := get_native_id()) not in self._storage:
                self._storage[ident] = {}
            try:
                return super().__getattribute__("_storage")[ident][name]
            except KeyError:
                raise AttributeError(f"object of type {self.__class__.__name__} has no attribute '{name}'") from None
    
    def __getitem__(self, name: object):
        """
        Implements getitem expressions
        such as self[name]
        """

        return self.__getattr__(name)
    

    def __setitem__(self, name: object, val: object):
      """
      Implements setitem expressions such as
      self[name] = val
      """

      self.__setattr__(name, val)

    def __delitem__(self, name: object):
      """
      Implements the del destructor syntax
      for getitem expressions
      """

      self.__getattr__(name)
      del self._storage[get_native_id()][name]

    def __delattr__(self, name: object):
      """
      Implements the del destructor syntax
      for getattr expressions 
      """

      self.__getattr__(name)
      del self._storage[get_native_id()][name]
    
    def exists(self, key: object) -> bool:
      """
      Returns True if the given key exists
      in the current thread's local space
      """

      with super().__getattribute__("_lock"):
          if (ident := get_native_id()) not in self._storage:
              self._storage[ident] = {}
              return False
          return key in super().__getattribute__("_storage")[ident]