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