1  import contextlib
  2  from typing import Dict
  3  from typing import Iterator
  4  from typing import List
  5  from typing import Type
  6  from typing import TypeVar
  7  
  8  import dynaconf
  9  from dynaconf import LazySettings
 10  from dynaconf.utils.boxing import DynaBox
 11  from litestar import Litestar
 12  from litestar.di import Provide
 13  
 14  from evidently._pydantic_compat import BaseModel
 15  from evidently._pydantic_compat import PrivateAttr
 16  from evidently._pydantic_compat import parse_obj_as
 17  from evidently.legacy.ui.components.base import SECTION_COMPONENT_TYPE_MAPPING
 18  from evidently.legacy.ui.components.base import AppBuilder
 19  from evidently.legacy.ui.components.base import Component
 20  from evidently.legacy.ui.components.base import ComponentContext
 21  from evidently.legacy.ui.components.base import ServiceComponent
 22  from evidently.legacy.ui.components.base import T
 23  from evidently.legacy.ui.components.security import SecurityComponent
 24  
 25  
 26  def _convert_keys(box):
 27      if isinstance(box, (DynaBox, LazySettings)):
 28          return {k.lower(): _convert_keys(v) for k, v in box.items()}
 29      return box
 30  
 31  
 32  class ConfigContext(ComponentContext):
 33      def __init__(self, config: "Config", components_mapping: Dict[Type[Component], Component]):
 34          self.config = config
 35          self.components_mapping = components_mapping
 36  
 37      def get_component(self, type_: Type[T]) -> T:
 38          for cls in self.components_mapping:
 39              if issubclass(cls, type_):
 40                  return self.components_mapping[cls]  # type: ignore[return-value]
 41          raise ValueError(f"Component of type {type_.__name__} not found")
 42  
 43      @property
 44      def components(self) -> List[Component]:
 45          return list(sorted(self.components_mapping.values(), key=lambda x: -x.get_priority()))
 46  
 47      def get_dependencies(self) -> Dict[str, Provide]:
 48          res = {}
 49          for c in self.components:
 50              dependencies = c.get_dependencies(self)
 51              print(f"{c.__class__.__name__} deps: " + ", ".join(dependencies))
 52              res.update(dependencies)
 53          return res
 54  
 55      def get_middlewares(self):
 56          res = []
 57          for c in self.components:
 58              res.extend(c.get_middlewares(self))
 59          return res
 60  
 61      def apply(self, builder: AppBuilder):
 62          for c in self.components:
 63              c.apply(self, builder)
 64  
 65      def finalize(self, app: Litestar):
 66          for c in self.components:
 67              c.finalize(self, app)
 68  
 69      def validate(self):
 70          for c in self.components_mapping.values():
 71              reqs = c.get_requirements()
 72              for r in reqs:
 73                  try:
 74                      self.get_component(r)
 75                  except ValueError as e:
 76                      raise ValueError(f"Component {c.__class__.__name__} missing {r.__name__} requirement") from e
 77  
 78  
 79  class Config(BaseModel):
 80      additional_components: Dict[str, Component] = {}
 81  
 82      _components: List[Component] = PrivateAttr(default_factory=list)
 83      _ctx: ComponentContext = PrivateAttr()
 84  
 85      @property
 86      def components(self) -> List[Component]:
 87          return [getattr(self, name) for name in self.__fields__ if isinstance(getattr(self, name), Component)] + list(
 88              self.additional_components.values()
 89          )
 90  
 91      @contextlib.contextmanager
 92      def context(self) -> Iterator[ConfigContext]:
 93          ctx = ConfigContext(self, {type(c): c for c in self.components})
 94          ctx.validate()
 95          self._ctx = ctx
 96          yield ctx
 97          del self._ctx
 98  
 99  
100  class AppConfig(Config):
101      security: SecurityComponent
102      service: ServiceComponent
103  
104  
105  TConfig = TypeVar("TConfig", bound=Config)
106  
107  
108  def load_config(config_type: Type[TConfig], box: dict) -> TConfig:
109      new_box = _convert_keys(box)
110      components = {}
111      named_components = {}
112      for section, component_dict in new_box.items():
113          # todo
114          if not isinstance(component_dict, dict):
115              continue
116          if section.endswith("for_dynaconf"):
117              continue
118          if section in ("renamed_vars", "dict_itemiterator"):
119              continue
120          if section == "additional_components":
121              for subsection, compoennt_subdict in component_dict.items():
122                  component = parse_obj_as(SECTION_COMPONENT_TYPE_MAPPING.get(subsection, Component), compoennt_subdict)
123                  components[subsection] = component
124          elif section in config_type.__fields__:
125              type_ = config_type.__fields__[section].type_
126              component = parse_obj_as(type_, component_dict)
127              named_components[section] = component
128          elif section in SECTION_COMPONENT_TYPE_MAPPING:
129              component = parse_obj_as(SECTION_COMPONENT_TYPE_MAPPING[section], component_dict)
130              components[section] = component
131          else:
132              raise ValueError(f"unknown config section {section}")
133  
134      # todo: we will get validation error if not all components configured, but we can wrap it more nicely
135      return config_type(additional_components=components, **named_components)
136  
137  
138  def load_config_from_file(cls: Type[TConfig], path: str, envvar_prefix: str = "EVIDENTLY") -> TConfig:
139      dc = dynaconf.Dynaconf(
140          envvar_prefix=envvar_prefix,
141      )
142      dc.configure(settings_module=path)
143      config = load_config(cls, dc)
144      return config
145  
146  
147  settings = dynaconf.Dynaconf(
148      envvar_prefix="EVIDENTLY",
149  )