3 from abc import ABC, abstractmethod
4 from datetime import datetime
5 from decorators import invocation_logged
7 from typing import Dict, List, Optional, Set
11 """Base class for something that can render."""
22 class abstaining_renderer(renderer):
23 """A renderer that doesn't do it all the time."""
25 def __init__(self, name_to_timeout_dict: Dict[str, int]) -> None:
26 self.name_to_timeout_dict = name_to_timeout_dict
28 for key in name_to_timeout_dict:
29 self.last_runs[key] = 0.0
31 def should_render(self, keys_to_skip: Set[str]) -> Optional[str]:
33 for key in self.name_to_timeout_dict:
35 (now - self.last_runs[key]) > self.name_to_timeout_dict[key]
36 ) and key not in keys_to_skip:
41 def render(self) -> None:
42 tries_per_key: Dict[str, int] = {}
43 keys_to_skip: Set[str] = set()
45 key = self.should_render(keys_to_skip)
49 if key in tries_per_key:
50 tries_per_key[key] += 1
52 tries_per_key[key] = 0
54 if tries_per_key[key] >= 3:
56 'renderer: Too many failures for "%s.%s", giving up'
57 % (self.get_name(), key)
61 msg = 'renderer: executing "%s.%s"' % (self.get_name(), key)
62 if tries_per_key[key] > 1:
63 msg = msg + " (retry #%d)" % tries_per_key[key]
65 if self.periodic_render(key):
66 self.last_runs[key] = time.time()
70 def periodic_render(self, key) -> bool:
73 def get_name(self) -> str:
74 return self.__class__.__name__
77 class debuggable_abstaining_renderer(abstaining_renderer):
78 def __init__(self, name_to_timeout_dict: Dict[str, int], debug: bool) -> None:
79 super(debuggable_abstaining_renderer, self).__init__(name_to_timeout_dict)
82 def debug_prefix(self) -> str:
83 return self.get_name()
85 def being_debugged(self) -> bool:
88 def debug_print(self, template: str, *args) -> None:
90 if self.being_debugged():
92 msg = template.format(args)
96 # current date and time
98 timestamp = now.strftime("%d-%b-%Y (%H:%M:%S.%f)")
99 print("%s(%s): %s" % (self.debug_prefix(), timestamp, msg))
100 except Exception as e:
101 print("Exception in debug_print!")