Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/fetcher/fetcher_register.py: 34%

1from typing import Dict, Optional, Type 

2 

3from opal_common.config import opal_common_config 

4from opal_common.fetcher.events import FetchEvent 

5from opal_common.fetcher.fetch_provider import BaseFetchProvider 

6from opal_common.fetcher.logger import get_logger 

7from opal_common.fetcher.providers.http_fetch_provider import HttpFetchProvider 

8 

9logger = get_logger("opal.fetcher_register") 

10 

11 

12class FetcherRegisterException(Exception): 

13 pass 

14 

15 

16class NoMatchingFetchProviderException(FetcherRegisterException): 

17 pass 

18 

19 

20class FetcherRegister: 

21 """A store for fetcher providers.""" 

22 

23 # Builtin fetchers 

24 BASIC_CONFIG = { 

25 "HttpFetchProvider": HttpFetchProvider, 

26 } 

27 

28 def __init__(self, config: Optional[Dict[str, BaseFetchProvider]] = None) -> None: 

29 if config is not None: 

30 self._config = config 

31 else: 

32 from opal_common.emport import emport_objects_by_class 

33 

34 # load fetchers 

35 fetchers = [] 

36 for module_path in opal_common_config.FETCH_PROVIDER_MODULES: 

37 try: 

38 providers_to_register = emport_objects_by_class( 

39 module_path, BaseFetchProvider, ["*"] 

40 ) 

41 for provider_name, provider_class in providers_to_register: 

42 logger.info( 

43 f"Loading FetcherProvider '{provider_name}' found at: {repr(provider_class)}" 

44 ) 

45 fetchers.extend(providers_to_register) 

46 except: 

47 logger.exception( 

48 f"Failed to load FetchingProvider module: {module_path}" 

49 ) 

50 self._config = {name: fetcher for name, fetcher in fetchers} 

51 logger.info("Fetcher Register loaded", extra={"config": self._config}) 

52 

53 def register_fetcher(self, name: str, fetcher_class: Type[BaseFetchProvider]): 

54 self._config[name] = fetcher_class 

55 

56 def get_fetcher(self, name: str, event: FetchEvent) -> BaseFetchProvider: 

57 """Init a fetcher instance from a registered fetcher class name. 

58 

59 Args: 

60 name (str): Name of a registered fetcher 

61 event (FetchEvent): Event used to configure the fetcher 

62 

63 Returns: 

64 BaseFetchProvider: A fetcher instance 

65 """ 

66 provider_class = self._config.get(name, None) 

67 if provider_class is None: 

68 raise NoMatchingFetchProviderException( 

69 f"Couldn't find a match for - {name} , {event}" 

70 ) 

71 fetcher = provider_class(event) 

72 if event.retry is not None: 

73 fetcher.set_retry_config(event.retry) 

74 return fetcher 

75 

76 def get_fetcher_for_event(self, event: FetchEvent) -> BaseFetchProvider: 

77 """Same as get_fetcher, using the event information to deduce the 

78 fetcher class. 

79 

80 Args: 

81 event (FetchEvent): Event used to choose and configure the fetcher 

82 

83 Returns: 

84 BaseFetchProvider: A fetcher instance 

85 """ 

86 return self.get_fetcher(event.fetcher, event)