Coverage for src/extratools_cloud/common/router.py: 0%

1from abc import ABC, abstractmethod 

2from collections.abc import Iterable 

3from logging import getLogger 

4from typing import Any, cast 

5 

6logger = getLogger(__name__) 

7 

8 

9class BaseRouter[RT: Any, TT: Any](ABC): 

10 def __init__( 

11 self, 

12 *, 

13 default_target_resource: RT, 

14 ) -> None: 

15 self._default_target_resource: RT = default_target_resource 

16 

17 self.targets: dict[TT, RT] = {} 

18 

19 def register_targets( 

20 self, 

21 resource: RT, 

22 targets: Iterable[TT], 

23 ) -> None: 

24 targets = list(targets) 

25 if not targets: 

26 return 

27 

28 logger.info(f"Registering resource {resource} for targets {targets}") 

29 

30 self.targets.update(dict.fromkeys(targets, resource)) 

31 

32 @abstractmethod 

33 def _route_to_resource[DT: Any]( 

34 self, 

35 data: Iterable[DT], 

36 resource: RT, 

37 target: TT, 

38 ) -> Iterable[Any]: 

39 ... 

40 

41 def route_to_target[DT: Any]( 

42 self, 

43 data: Iterable[DT], 

44 target: TT, 

45 ) -> Iterable[Any]: 

46 resource: RT = cast("RT", self.targets.get(target, self._default_target_resource)) 

47 

48 logger.info(f"Routing to resource {resource} for target {target}") 

49 

50 return self._route_to_resource(data, resource, target)