refactor(applets): drop omegaconf for a dotted config lookup

bbot_server/applets/base.py

@@ -5,7 +5,6 @@
 import traceback
 
 from fastapi import APIRouter
-from omegaconf import OmegaConf
 from typing import Annotated, Any, get_origin, get_args, Union, Callable, cast  # noqa
 from functools import cached_property
 from pydantic import BaseModel, Field  # noqa
@@ -309,7 +308,7 @@ async def register_worker_tasks(self, broker):
                     raise ValueError(
                         f"{self.name}.{method_name}: When specifying a crontab config value, you must also give a default crontab value (kwarg: 'cron')"
                     )
-                cron = OmegaConf.select(self.global_config, cron_config_key, default=cron_default)
+                cron = bbot_server_misc.select_dotted(self.global_config, cron_config_key, default=cron_default)
                 kwargs["schedule"] = [{"cron": cron}]
             elif cron_default is not None:
                 kwargs["schedule"] = [{"cron": cron_default}]

bbot_server/utils/misc.py

@@ -5,6 +5,7 @@
 from functools import wraps
 from types import UnionType
 from inspect import signature
+from collections.abc import Mapping
 from pydantic import BaseModel, create_model
 from datetime import datetime, timezone, timedelta
 from typing import Any, Union, get_origin, get_args, get_type_hints, Annotated
@@ -152,6 +153,31 @@ def wrapper(*args, **kwargs):
     return fn
 
 
+def select_dotted(obj: Any, key: str, default: Any = None) -> Any:
+    """
+    Resolve a dotted config key against a pydantic model or mapping.
+
+    Args:
+        obj: Root config object to traverse
+        key: Dotted path, e.g. "modules.foo.cron"
+        default: Returned if any segment is missing or resolves to None
+
+    Returns:
+        The resolved value, or default.
+    """
+    current = obj
+    for segment in key.split("."):
+        if isinstance(current, Mapping):
+            if segment not in current:
+                return default
+            current = current[segment]
+        else:
+            if not hasattr(current, segment):
+                return default
+            current = getattr(current, segment)
+    return default if current is None else current
+
+
 def utc_now() -> float:
     return datetime.now(timezone.utc).timestamp()
 

pyproject.toml

@@ -24,7 +24,6 @@ dependencies = [
     "pymongo>=4.15.3",
     "cloudcheck>=9.2.0",
     "textual>=7.5.0",
-    "omegaconf>=2.3.0,<3",
     "bbot",
 ]