Source code for reactpy._option

from __future__ import annotations

import os
from logging import getLogger
from typing import Any, Callable, Generic, TypeVar, cast

from reactpy._warnings import warn

_O = TypeVar("_O")
logger = getLogger(__name__)
UNDEFINED = cast(Any, object())

[docs]class Option(Generic[_O]): """An option that can be set using an environment variable of the same name""" def __init__( self, name: str, default: _O = UNDEFINED, mutable: bool = True, parent: Option[_O] | None = None, validator: Callable[[Any], _O] = lambda x: cast(_O, x), ) -> None: self._name = name self._mutable = mutable self._validator = validator self._subscribers: list[Callable[[_O], None]] = [] if name in os.environ: self._current = validator(os.environ[name]) if parent is not None: if not (parent.mutable and self.mutable): raise TypeError("Parent and child options must be mutable") self._default = parent.default parent.subscribe(self.set_current) elif default is not UNDEFINED: self._default = default else: raise TypeError("Must specify either a default or a parent option") logger.debug(f"{self._name}={self.current}") @property def name(self) -> str: """The name of this option (used to load environment variables)""" return self._name @property def mutable(self) -> bool: """Whether this option can be modified after being loaded""" return self._mutable @property def default(self) -> _O: """This option's default value""" return self._default @property def current(self) -> _O: try: return self._current except AttributeError: return self._default @current.setter def current(self, new: _O) -> None: self.set_current(new) @current.deleter def current(self) -> None: self.unset()
[docs] def subscribe(self, handler: Callable[[_O], None]) -> Callable[[_O], None]: """Register a callback that will be triggered when this option changes""" if not self.mutable: msg = "Immutable options cannot be subscribed to." raise TypeError(msg) self._subscribers.append(handler) handler(self.current) return handler
[docs] def is_set(self) -> bool: """Whether this option has a value other than its default.""" return hasattr(self, "_current")
[docs] def set_current(self, new: Any) -> None: """Set the value of this option Raises a ``TypeError`` if this option is not :attr:`Option.mutable`. """ old = self.current if new is old: return None if not self._mutable: msg = f"{self} cannot be modified after initial load" raise TypeError(msg) try: new = self._current = self._validator(new) except ValueError as error: raise ValueError(f"Invalid value for {self._name}: {new!r}") from error logger.debug(f"{self._name}={self._current}") if new != old: for sub_func in self._subscribers: sub_func(new)
[docs] def set_default(self, new: _O) -> _O: """Set the value of this option if not :meth:`Option.is_set` Returns the current value (a la :meth:`dict.set_default`) """ if not self.is_set(): self.set_current(new) return self._current
[docs] def reload(self) -> None: """Reload this option from its environment variable""" self.set_current(os.environ.get(self._name, self._default))
[docs] def unset(self) -> None: """Remove the current value, the default will be used until it is set again.""" if not self._mutable: msg = f"{self} cannot be modified after initial load" raise TypeError(msg) old = self.current if hasattr(self, "_current"): delattr(self, "_current") if self.current != old: for sub_func in self._subscribers: sub_func(self.current)
def __repr__(self) -> str: return f"Option({self._name}={self.current!r})"
[docs]class DeprecatedOption(Option[_O]): """An option that will warn when it is accessed""" def __init__(self, *args: Any, message: str, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self._deprecation_message = message @Option.current.getter # type: ignore def current(self) -> _O: try: # we access the current value during init to debug log it # no need to warn unless it's actually used. since this attr # is only set after super().__init__ is called, we can check # for it to determine if it's being accessed by a user. msg = self._deprecation_message except AttributeError: pass else: warn(msg, DeprecationWarning) return super().current