Class ThreadLocal

hypothesis/src/hypothesis/utils/threading.py:16–52

Manages thread-local state. ThreadLocal forwards getattr and setattr to a threading.local() instance. The passed kwargs defines the available attributes on the threadlocal and their default values. The only supported names to geattr and setattr are the keys of the passed kwargs.

class ThreadLocal:
    """
    Manages thread-local state. ThreadLocal forwards getattr and setattr to a
    threading.local() instance. The passed kwargs defines the available attributes
    on the threadlocal and their default values.

    The only supported names to geattr and setattr are the keys of the passed kwargs.
    """

    def __init__(self, **kwargs: Callable) -> None:
        for name, value in kwargs.items():
            if not callable(value):
                raise TypeError(f"Attribute {name} must be a callable. Got {value}")

        self.__initialized = False
        self.__kwargs = kwargs
        self.__threadlocal = threading.local()
        self.__initialized = True

    def __getattr__(self, name: str) -> Any:
        if name not in self.__kwargs:
            raise AttributeError(f"No attribute {name}")

        if not hasattr(self.__threadlocal, name):
            default = self.__kwargs[name]()
            setattr(self.__threadlocal, name, default)

        return getattr(self.__threadlocal, name)

    def __setattr__(self, name: str, value: Any) -> None:
        # disable attribute-forwarding while initializing
        if "_ThreadLocal__initialized" not in self.__dict__ or not self.__initialized:
            super().__setattr__(name, value)
        else:
            if name not in self.__kwargs:
                raise AttributeError(f"No attribute {name}")
            setattr(self.__threadlocal, name, value)