Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/sources/base_policy_source.py: 37%

1import asyncio 

2import os 

3from functools import partial 

4from typing import Callable, Coroutine, List, Union 

5 

6from git.objects.commit import Commit 

7from opal_common.logger import logger 

8 

9OnNewPolicyCallback = Callable[[Commit, Commit], Coroutine] 

10OnPolicyFailureCallback = Callable[[Exception], Coroutine] 

11 

12 

13class BasePolicySource: 

14 """Base class to support git and api policy source. 

15 

16 Args: 

17 remote_source_url(str): the base address to request the policy from 

18 local_clone_path(str): path for the local git to manage policies 

19 polling_interval(int): how many seconds need to wait between polling 

20 """ 

21 

22 def __init__( 

23 self, 

24 remote_source_url: str, 

25 local_clone_path: str, 

26 polling_interval: int = 0, 

27 ): 

28 self._on_failure_callbacks: List[OnNewPolicyCallback] = [] 

29 self._on_new_policy_callbacks: List[OnPolicyFailureCallback] = [] 

30 self._polling_interval = polling_interval 

31 self._polling_task = None 

32 self.remote_source_url = remote_source_url 

33 self.local_clone_path = os.path.expanduser(local_clone_path) 

34 

35 def add_on_new_policy_callback(self, callback: OnNewPolicyCallback): 

36 """Register a callback that will be called when new policy are detected 

37 on the monitored repo (after a pull).""" 

38 self._on_new_policy_callbacks.append(callback) 

39 

40 def add_on_failure_callback(self, callback: OnPolicyFailureCallback): 

41 """Register a callback that will be called when failure occurred.""" 

42 self._on_failure_callbacks.append(callback) 

43 

44 async def get_initial_policy_state_from_remote(self): 

45 """Init remote data to local repo.""" 

46 raise NotImplementedError() 

47 

48 async def check_for_changes(self): 

49 """Trigger check for policy change.""" 

50 raise NotImplementedError() 

51 

52 async def run(self): 

53 """Potentially starts the polling task.""" 

54 await self.get_initial_policy_state_from_remote() 

55 

56 if self._polling_interval > 0: 

57 logger.info( 

58 "Launching polling task, interval: {interval} seconds", 

59 interval=self._polling_interval, 

60 ) 

61 self._start_polling_task(self.check_for_changes) 

62 else: 

63 logger.info("Polling task is off") 

64 

65 async def stop(self): 

66 return await self._stop_polling_task() 

67 

68 def _start_polling_task(self, polling_task): 

69 if self._polling_task is None and self._polling_interval > 0: 

70 self._polling_task = asyncio.create_task(self._do_polling(polling_task)) 

71 

72 async def _do_polling(self, polling_task): 

73 """Optional task to periodically check the remote for changes (git pull 

74 and compare hash).""" 

75 while True: 

76 try: 

77 await polling_task() 

78 except Exception as ex: 

79 logger.error( 

80 "Error occurred during polling task {task}: {err}", 

81 task=polling_task.__name__, 

82 err=ex, 

83 ) 

84 await asyncio.sleep(self._polling_interval) 

85 

86 async def _stop_polling_task(self): 

87 if self._polling_task is not None: 

88 self._polling_task.cancel() 

89 try: 

90 await self._polling_task 

91 except asyncio.CancelledError: 

92 pass 

93 

94 async def _on_new_policy(self, old: Commit, new: Commit): 

95 """Triggers callbacks registered with on_new_policy().""" 

96 await self._run_callbacks(self._on_new_policy_callbacks, old, new) 

97 

98 async def _on_failed(self, exc: Exception): 

99 """Will be triggered if a failure occurred. 

100 

101 triggers callbacks registered with on_git_failed(). 

102 """ 

103 await self._run_callbacks(self._on_failure_callbacks, exc) 

104 

105 async def _run_callbacks(self, handlers, *args, **kwargs): 

106 """Triggers a list of callbacks.""" 

107 await asyncio.gather(*(callback(*args, **kwargs) for callback in handlers)) 

108 

109 async def _on_git_failed(self, exc: Exception): 

110 """Will be triggered if a git failure occurred (i.e: repo does not 

111 exist, can't clone, etc). 

112 

113 triggers callbacks registered with on_git_failed(). 

114 """ 

115 await self._run_callbacks(self._on_failure_callbacks, exc)