|
@@ -132,9 +132,9 @@ def _get_env_or_extra_truthy(env_var: str, extra_val: Optional[str | bool], defa
|
|
|
return utils.is_truthy_value(_get_env_or_extra_str(env_var, extra_val, str(default)), default)
|
|
return utils.is_truthy_value(_get_env_or_extra_str(env_var, extra_val, str(default)), default)
|
|
|
|
|
|
|
|
|
|
|
|
|
-def _split_str_to_list(mystring: str) -> list:
|
|
|
|
|
- logger.info("mystring: %s", mystring)
|
|
|
|
|
- return list(c for c in mystring.split(","))
|
|
|
|
|
|
|
+def _split_str_to_list(mystring: str) -> list[str]:
|
|
|
|
|
+ """Split a comma-separated string, trimming whitespace and dropping empties."""
|
|
|
|
|
+ return [part.strip() for part in mystring.split(",") if part.strip()]
|
|
|
|
|
|
|
|
|
|
|
|
|
def _get_env_or_extra_list(env_var: str, extra_val: Optional[list[str]]) -> list[str]:
|
|
def _get_env_or_extra_list(env_var: str, extra_val: Optional[list[str]]) -> list[str]:
|
|
@@ -144,106 +144,164 @@ def _get_env_or_extra_list(env_var: str, extra_val: Optional[list[str]]) -> list
|
|
|
return _split_str_to_list(env_value)
|
|
return _split_str_to_list(env_value)
|
|
|
|
|
|
|
|
if extra_val is not None:
|
|
if extra_val is not None:
|
|
|
- #logger.info("extra_val is1: " + str(type(extra_val)))
|
|
|
|
|
- #logger.info("extra_val is2: " + str(extra_val))
|
|
|
|
|
if isinstance(extra_val, list):
|
|
if isinstance(extra_val, list):
|
|
|
- return list(
|
|
|
|
|
- c for c in extra_val
|
|
|
|
|
- )
|
|
|
|
|
|
|
+ return [str(c).strip() for c in extra_val if str(c).strip()]
|
|
|
if isinstance(extra_val, str):
|
|
if isinstance(extra_val, str):
|
|
|
return _split_str_to_list(extra_val)
|
|
return _split_str_to_list(extra_val)
|
|
|
|
|
|
|
|
return []
|
|
return []
|
|
|
|
|
|
|
|
-from typing import TypeVar, Generic, Any
|
|
|
|
|
|
|
|
|
|
-# 1. Define a generic Type Variable
|
|
|
|
|
-T = TypeVar('T')
|
|
|
|
|
|
|
+T = TypeVar("T")
|
|
|
|
|
|
|
|
-# 2. Inherit from Generic[T]
|
|
|
|
|
-class ConfigField(Generic[T]):
|
|
|
|
|
- """Repräsentiert ein einzelnes Konfigurationsfeld mit IDE-Support."""
|
|
|
|
|
-# 3. Type 'value' as T (or T | None since it starts as None)
|
|
|
|
|
- value: T
|
|
|
|
|
- field_name: str = ""
|
|
|
|
|
- env_name: str = ""
|
|
|
|
|
|
|
|
|
|
- # 4. Hint that the init argument should match type T
|
|
|
|
|
- # We add `| Any` as a fallback because complex types like `list[str]`
|
|
|
|
|
- # can sometimes confuse older type checkers when used with `type[T]`
|
|
|
|
|
- def __init__(self, type_: type[T] | Any):
|
|
|
|
|
- self._type = type_
|
|
|
|
|
|
|
+class ConfigValue(Generic[T]):
|
|
|
|
|
+ """The resolved value of a single config field, bound to one configuration
|
|
|
|
|
+ instance.
|
|
|
|
|
+
|
|
|
|
|
+ Access the payload via ``.value``. ``__bool__``/``__eq__`` delegate to it,
|
|
|
|
|
+ so a forgotten ``.value`` (``if config.allow_all_users:``) still evaluates
|
|
|
|
|
+ the actual setting instead of the always-truthy wrapper object.
|
|
|
|
|
+ """
|
|
|
|
|
+
|
|
|
|
|
+ __slots__ = ("value", "field_name", "env_name")
|
|
|
|
|
|
|
|
- # 5. Ensure the IDE knows only type T can be assigned
|
|
|
|
|
- def __set__(self, instance: Any, value: T) -> None:
|
|
|
|
|
|
|
+ def __init__(self, value: T, field_name: str, env_name: str) -> None:
|
|
|
self.value = value
|
|
self.value = value
|
|
|
- self._type = type(value)
|
|
|
|
|
- logger.info("Chatto: configuration field '%s' set to '%s'", self.field_name, str(value))
|
|
|
|
|
|
|
+ self.field_name = field_name
|
|
|
|
|
+ self.env_name = env_name
|
|
|
|
|
+
|
|
|
|
|
+ def __bool__(self) -> bool:
|
|
|
|
|
+ return bool(self.value)
|
|
|
|
|
+
|
|
|
|
|
+ def __eq__(self, other: Any) -> bool:
|
|
|
|
|
+ if isinstance(other, ConfigValue):
|
|
|
|
|
+ return self.value == other.value
|
|
|
|
|
+ return self.value == other
|
|
|
|
|
+
|
|
|
|
|
+ def __hash__(self) -> int:
|
|
|
|
|
+ return hash(self.value)
|
|
|
|
|
+
|
|
|
|
|
+ def __contains__(self, item: Any) -> bool:
|
|
|
|
|
+ return item in self.value # type: ignore[operator]
|
|
|
|
|
+
|
|
|
|
|
+ def __iter__(self):
|
|
|
|
|
+ return iter(self.value) # type: ignore[call-overload]
|
|
|
|
|
|
|
|
def __str__(self) -> str:
|
|
def __str__(self) -> str:
|
|
|
return str(self.value)
|
|
return str(self.value)
|
|
|
|
|
|
|
|
- def __set_name__(self, owner: Any, name: str) -> None:
|
|
|
|
|
- temp: str = str(name).lower().replace("chatto_", "", 1)
|
|
|
|
|
- self.field_name = temp
|
|
|
|
|
- self.env_name = f"CHATTO_{temp.upper()}"
|
|
|
|
|
|
|
+ def __repr__(self) -> str:
|
|
|
|
|
+ return f"{self.field_name}={self.value!r}"
|
|
|
|
|
|
|
|
|
|
|
|
|
-@dataclass
|
|
|
|
|
-class ChattoConfiguration:
|
|
|
|
|
- """
|
|
|
|
|
- Chatto Platform Config
|
|
|
|
|
|
|
+class ConfigField(Generic[T]):
|
|
|
|
|
+ """Declarative descriptor for one config field.
|
|
|
|
|
|
|
|
- Some constants and getting from environment/config.yaml
|
|
|
|
|
|
|
+ Declares *how* a field is read (kind, env var name, default); the resolved
|
|
|
|
|
+ payload lives per configuration instance in ``instance._values``, never on
|
|
|
|
|
+ the descriptor itself. Reading a field on the class (rather than on an
|
|
|
|
|
+ instance) yields the descriptor, so ``ChattoConfiguration.token.env_name``
|
|
|
|
|
+ keeps working for the registration hooks.
|
|
|
"""
|
|
"""
|
|
|
- base_url = ConfigField(str)
|
|
|
|
|
- token = ConfigField(str | None)
|
|
|
|
|
- login = ConfigField(str)
|
|
|
|
|
- password = ConfigField(str)
|
|
|
|
|
- channels = ConfigField(str)
|
|
|
|
|
- channels_list = ConfigField(list[str])
|
|
|
|
|
- home_channel = ConfigField(str)
|
|
|
|
|
- allowed_users = ConfigField(list[str])
|
|
|
|
|
- require_mention = ConfigField(bool)
|
|
|
|
|
- free_response_channels_list = ConfigField(list[str])
|
|
|
|
|
- auto_thread = ConfigField(bool)
|
|
|
|
|
- allow_all_users = ConfigField(bool)
|
|
|
|
|
- reactions = ConfigField(bool)
|
|
|
|
|
|
|
|
|
|
- def __init__(self, pconfig: PlatformConfig):
|
|
|
|
|
- """PlatformConfig from Hermes provides our own configuration within the "extra"
|
|
|
|
|
- object. But here, we allow overriding via environment variables again.
|
|
|
|
|
|
|
+ field_name: str = ""
|
|
|
|
|
+ env_name: str = ""
|
|
|
|
|
|
|
|
- We take our Configuration from this typed Class, because it is easier access than using extra.get["base_url"].
|
|
|
|
|
- """
|
|
|
|
|
- self.base_url.value = _get_env_or_extra_str(self.base_url.env_name,
|
|
|
|
|
- pconfig.extra.get(self.base_url.field_name), ChattoClient.DEFAULT_BASE_URL)
|
|
|
|
|
|
|
+ def __init__(
|
|
|
|
|
+ self,
|
|
|
|
|
+ kind: str,
|
|
|
|
|
+ *,
|
|
|
|
|
+ default: Any = None,
|
|
|
|
|
+ config_key: Optional[str] = None,
|
|
|
|
|
+ doc: str = "",
|
|
|
|
|
+ ) -> None:
|
|
|
|
|
+ # kind: "str" | "str_opt" | "bool" | "list"
|
|
|
|
|
+ self.kind = kind
|
|
|
|
|
+ self.default = default
|
|
|
|
|
+ # config_key: the name used in config.yaml's "extra" block and (upper-cased,
|
|
|
|
|
+ # CHATTO_-prefixed) as the env var, where it differs from the attribute name.
|
|
|
|
|
+ self._config_key = config_key
|
|
|
|
|
+ self.__doc__ = doc
|
|
|
|
|
+
|
|
|
|
|
+ def __set_name__(self, owner: Any, name: str) -> None:
|
|
|
|
|
+ self.field_name = str(name).lower().replace("chatto_", "", 1)
|
|
|
|
|
+ self.config_key = self._config_key or self.field_name
|
|
|
|
|
+ self.env_name = f"CHATTO_{self.config_key.upper()}"
|
|
|
|
|
+
|
|
|
|
|
+ def __get__(self, instance: Any, owner: Any = None) -> "ConfigValue[T]":
|
|
|
|
|
+ if instance is None:
|
|
|
|
|
+ return self # type: ignore[return-value]
|
|
|
|
|
+ return instance._values[self.field_name]
|
|
|
|
|
+
|
|
|
|
|
+ def __set__(self, instance: Any, value: Any) -> None:
|
|
|
|
|
+ raise AttributeError(
|
|
|
|
|
+ f"Chatto: config field '{self.field_name}' is read-only; "
|
|
|
|
|
+ f"set '{self.field_name}.value' if you really need to override it."
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ def resolve(self, extra: Dict[str, Any]) -> ConfigValue:
|
|
|
|
|
+ """Read this field from the environment, then from ``extra``, then the default."""
|
|
|
|
|
+ raw = extra.get(self.config_key)
|
|
|
|
|
+
|
|
|
|
|
+ if self.kind == "list":
|
|
|
|
|
+ value: Any = _get_env_or_extra_list(self.env_name, raw)
|
|
|
|
|
+ elif self.kind == "bool":
|
|
|
|
|
+ value = _get_env_or_extra_truthy(self.env_name, raw, bool(self.default))
|
|
|
|
|
+ elif self.kind == "str_opt":
|
|
|
|
|
+ value = _get_env_or_extra_str_opt(self.env_name, raw, self.default)
|
|
|
|
|
+ else:
|
|
|
|
|
+ value = _get_env_or_extra_str(self.env_name, raw, self.default)
|
|
|
|
|
|
|
|
- self.token.value = _get_env_or_extra_str_opt(self.token.env_name, pconfig.extra.get(self.token.field_name))
|
|
|
|
|
- self.login.value = _get_env_or_extra_str(self.login.env_name, pconfig.extra.get(self.login.field_name))
|
|
|
|
|
- self.password.value = _get_env_or_extra_str(self.password.env_name, pconfig.extra.get(self.password.field_name))
|
|
|
|
|
-
|
|
|
|
|
- self.channels.value = str(_get_env_or_extra_list(self.channels.env_name, pconfig.extra.get(self.channels.field_name)))
|
|
|
|
|
- self.channels_list.value = _get_env_or_extra_list(self.channels.env_name, pconfig.extra.get(self.channels.field_name))
|
|
|
|
|
|
|
+ return ConfigValue(value, self.field_name, self.env_name)
|
|
|
|
|
|
|
|
- self.home_channel.value = _get_env_or_extra_str(self.home_channel.env_name, pconfig.extra.get(self.home_channel.field_name))
|
|
|
|
|
|
|
|
|
|
- self.allowed_users.value = _get_env_or_extra_list(self.allowed_users.env_name, pconfig.extra.get(self.allowed_users.field_name))
|
|
|
|
|
- logger.info("self.allowed_users.value: %s", self.allowed_users.value)
|
|
|
|
|
|
|
+class ChattoConfiguration:
|
|
|
|
|
+ """Chatto platform configuration.
|
|
|
|
|
|
|
|
- self.require_mention.value = _get_env_or_extra_truthy(self.require_mention.env_name, pconfig.extra.get(self.require_mention.field_name), False)
|
|
|
|
|
- logger.info("self.require_mention.value: %s", self.require_mention.value)
|
|
|
|
|
|
|
+ Every field is resolved once per instance, in this order:
|
|
|
|
|
+ environment variable → ``PlatformConfig.extra`` → declared default.
|
|
|
|
|
+ """
|
|
|
|
|
|
|
|
- # free_response_channels: room IDs where the bot responds without being mentioned via "@botname" when require_mention is true.
|
|
|
|
|
- self.free_response_channels_list.value = _get_env_or_extra_list(self.free_response_channels_list.env_name,
|
|
|
|
|
- pconfig.extra.get(self.free_response_channels_list.field_name))
|
|
|
|
|
|
|
+ base_url = ConfigField("str", default=ChattoClient.DEFAULT_BASE_URL)
|
|
|
|
|
+ token = ConfigField("str_opt")
|
|
|
|
|
+ login = ConfigField("str")
|
|
|
|
|
+ password = ConfigField("str")
|
|
|
|
|
+ channels_list = ConfigField("list", config_key="channels")
|
|
|
|
|
+ home_channel = ConfigField("str")
|
|
|
|
|
+ allowed_users = ConfigField("list")
|
|
|
|
|
+ require_mention = ConfigField("bool", default=False)
|
|
|
|
|
+ # free_response_channels: room IDs where the bot responds without being
|
|
|
|
|
+ # mentioned via "@botname" even when require_mention is true.
|
|
|
|
|
+ free_response_channels_list = ConfigField("list", config_key="free_response_channels")
|
|
|
|
|
+ # Auto-thread: by default, Chatto creates a thread for replies to room
|
|
|
|
|
+ # messages (not DMs, not already in a thread). This keeps conversations
|
|
|
|
|
+ # organized in the room. Can be disabled via extra.auto_thread=false.
|
|
|
|
|
+ auto_thread = ConfigField("bool", default=True)
|
|
|
|
|
+ allow_all_users = ConfigField("bool", default=False)
|
|
|
|
|
+ reactions = ConfigField("bool", default=True)
|
|
|
|
|
|
|
|
- # Auto-thread: by default, Chatto creates a thread for replies to room
|
|
|
|
|
- # messages (not DMs, not already in a thread). This keeps conversations
|
|
|
|
|
- # organized in the room. Can be disabled via extra.auto_thread=false.
|
|
|
|
|
- self.auto_thread.value = _get_env_or_extra_truthy(self.auto_thread.env_name, pconfig.extra.get(self.auto_thread.field_name), True)
|
|
|
|
|
|
|
+ def __init__(self, pconfig: PlatformConfig):
|
|
|
|
|
+ """Resolve every declared ConfigField against env vars and
|
|
|
|
|
+ ``PlatformConfig.extra`` (which Hermes pre-populates from config.yaml).
|
|
|
|
|
+ """
|
|
|
|
|
+ extra: Dict[str, Any] = getattr(pconfig, "extra", None) or {}
|
|
|
|
|
|
|
|
- self.allow_all_users.value = _get_env_or_extra_truthy(self.allow_all_users.env_name, pconfig.extra.get(self.allow_all_users.field_name))
|
|
|
|
|
- self.reactions.value = _get_env_or_extra_truthy(self.reactions.env_name, pconfig.extra.get(self.reactions.field_name), True)
|
|
|
|
|
|
|
+ self._values: Dict[str, ConfigValue] = {
|
|
|
|
|
+ field.field_name: field.resolve(extra) for field in self.fields()
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
- logger.info("ChattoConfiguration: %s", self)
|
|
|
|
|
|
|
+ logger.debug("ChattoConfiguration: %s", self)
|
|
|
|
|
+
|
|
|
|
|
+ @classmethod
|
|
|
|
|
+ def fields(cls) -> list[ConfigField]:
|
|
|
|
|
+ """All declared config fields, in declaration order."""
|
|
|
|
|
+ return [v for v in vars(cls).values() if isinstance(v, ConfigField)]
|
|
|
|
|
+
|
|
|
|
|
+ def __str__(self) -> str:
|
|
|
|
|
+ redacted = {"token", "password"}
|
|
|
|
|
+ parts = [
|
|
|
|
|
+ f"{name}={'***' if name in redacted and cv.value else cv.value!r}"
|
|
|
|
|
+ for name, cv in self._values.items()
|
|
|
|
|
+ ]
|
|
|
|
|
+ return "ChattoConfiguration(" + ", ".join(parts) + ")"
|