1  from abc import ABC
  2  from typing import Any
  3  from typing import Callable
  4  from typing import ClassVar
  5  from typing import Dict
  6  from typing import Generic
  7  from typing import List
  8  from typing import Literal
  9  from typing import Optional
 10  from typing import Type
 11  from typing import TypeVar
 12  from typing import overload
 13  
 14  from litestar import Litestar
 15  from litestar import Router
 16  from litestar.di import Provide
 17  from litestar.types import ControllerRouterHandler
 18  from litestar.types import ExceptionHandlersMap
 19  from litestar.types import Middleware
 20  
 21  from evidently._pydantic_compat import Extra
 22  from evidently.legacy.ui.utils import parse_json
 23  from evidently.pydantic_utils import PolymorphicModel
 24  
 25  SECTION_COMPONENT_TYPE_MAPPING: Dict[str, Type["Component"]] = {}
 26  
 27  
 28  T = TypeVar("T", bound="Component")
 29  
 30  
 31  class AppBuilder:
 32      def __init__(self, context: "ComponentContext"):
 33          self.context = context
 34          self.dependencies: Dict[str, Provide] = {}
 35          self.route_handlers: List[ControllerRouterHandler] = []
 36          self.api_route_handlers: List[ControllerRouterHandler] = []
 37          self.exception_handlers: ExceptionHandlersMap = {}
 38          self.middlewares: List[Middleware] = []
 39          self.type_decoders: List[tuple[Callable[[Any], bool], Callable[[Any, Any], Any]]] = []
 40          self.kwargs: Dict[str, Any] = {}
 41  
 42      def build_api_router(self):
 43          return Router(path="/api", route_handlers=self.api_route_handlers)
 44  
 45      def build(self) -> Litestar:
 46          api_router = self.build_api_router()
 47          return Litestar(
 48              route_handlers=[api_router] + self.route_handlers,
 49              exception_handlers=self.exception_handlers,
 50              dependencies=self.dependencies,
 51              middleware=self.middlewares,
 52              type_decoders=self.type_decoders,
 53              **self.kwargs,
 54          )
 55  
 56  
 57  class ComponentContext:
 58      @overload
 59      def get_component(self, type_: Type[T], required: Literal[True] = True) -> T:
 60          pass
 61  
 62      @overload
 63      def get_component(self, type_: Type[T], required: Literal[False] = False) -> Optional[T]:
 64          pass
 65  
 66      def get_component(self, type_: Type[T], required: bool = True) -> Optional[T]:
 67          raise NotImplementedError
 68  
 69  
 70  class Component(PolymorphicModel, ABC):
 71      __section__: ClassVar[str] = ""
 72      __require__: ClassVar[List[Type["Component"]]] = []
 73      __priority__: ClassVar[int] = 0
 74      priority: Optional[int] = None
 75  
 76      def get_priority(self) -> int:
 77          return self.priority if self.priority is not None else self.__priority__
 78  
 79      def get_requirements(self) -> List[Type["Component"]]:
 80          return self.__require__
 81  
 82      class Config:
 83          extra = Extra.forbid
 84          alias_required = False
 85          is_base_type = True
 86  
 87      def __init_subclass__(cls):
 88          super().__init_subclass__()
 89          if cls.__section__:
 90              SECTION_COMPONENT_TYPE_MAPPING[cls.__section__] = cls
 91  
 92      def get_dependencies(self, ctx: ComponentContext) -> Dict[str, Provide]:
 93          return {}
 94  
 95      def get_middlewares(self, ctx: ComponentContext):
 96          return []
 97  
 98      def get_route_handlers(self, ctx: ComponentContext):
 99          return []
100  
101      def get_api_route_handlers(self, ctx: ComponentContext):
102          return []
103  
104      def apply(self, ctx: ComponentContext, builder: AppBuilder):
105          builder.dependencies.update(self.get_dependencies(ctx))
106          builder.middlewares.extend(self.get_middlewares(ctx))
107          builder.route_handlers.extend(self.get_route_handlers(ctx))
108          builder.api_route_handlers.extend(self.get_api_route_handlers(ctx))
109  
110      def finalize(self, ctx: ComponentContext, app: Litestar):
111          pass
112  
113  
114  DT = TypeVar("DT")
115  
116  
117  class FactoryComponent(Component, Generic[DT], ABC):
118      dependency_name: ClassVar[str]
119      use_cache: ClassVar[bool] = True
120      sync_to_thread: ClassVar[Optional[bool]] = False
121  
122      def dependency_factory(self) -> Callable[..., DT]:
123          raise NotImplementedError(self.__class__)
124  
125      def get_dependencies(self, ctx: ComponentContext) -> Dict[str, Provide]:
126          return {
127              self.dependency_name: Provide(
128                  self.dependency_factory(), use_cache=self.use_cache, sync_to_thread=self.sync_to_thread
129              )
130          }
131  
132  
133  class ServiceComponent(Component):
134      host: str = "127.0.0.1"
135      port: int = 8000
136  
137      def get_dependencies(self, ctx: ComponentContext) -> Dict[str, Provide]:
138          # todo: maybe not put utils here
139          return {
140              "parsed_json": Provide(parse_json, sync_to_thread=False),
141          }