-import time
+#!/usr/bin/env python3
+
+from abc import ABC, abstractmethod
from datetime import datetime
-from decorators import invokation_logged
+from decorators import invocation_logged
+import time
+from typing import Dict, List, Optional, Set
-class renderer(object):
+
+class renderer(ABC):
"""Base class for something that can render."""
- @invokation_logged
+ @abstractmethod
def render(self):
pass
+ @abstractmethod
def get_name(self):
- return self.__class__.__name__
+ pass
+
class abstaining_renderer(renderer):
"""A renderer that doesn't do it all the time."""
- def __init__(self, name_to_timeout_dict):
- self.name_to_timeout_dict = name_to_timeout_dict;
+
+ def __init__(self, name_to_timeout_dict: Dict[str, int]) -> None:
+ self.name_to_timeout_dict = name_to_timeout_dict
self.last_runs = {}
for key in name_to_timeout_dict:
- self.last_runs[key] = 0
+ self.last_runs[key] = 0.0
- def should_render(self, keys_to_skip):
+ def should_render(self, keys_to_skip: Set[str]) -> Optional[str]:
now = time.time()
for key in self.name_to_timeout_dict:
- if (((now - self.last_runs[key]) > self.name_to_timeout_dict[key]) and
- key not in keys_to_skip):
+ if (
+ (now - self.last_runs[key]) > self.name_to_timeout_dict[key]
+ ) and key not in keys_to_skip:
return key
return None
- def render(self):
- tries_per_key = {}
- keys_to_skip = set()
+ @invocation_logged
+ def render(self) -> None:
+ tries_per_key: Dict[str, int] = {}
+ keys_to_skip: Set[str] = set()
while True:
key = self.should_render(keys_to_skip)
- if key == None:
+ if key is None:
break
if key in tries_per_key:
tries_per_key[key] = 0
if tries_per_key[key] >= 3:
- print('renderer: Too many failures for "%s.%s", giving up' % (
- self.get_name(), key))
+ print(
+ 'renderer: Too many failures for "%s.%s", giving up'
+ % (self.get_name(), key)
+ )
keys_to_skip.add(key)
else:
msg = 'renderer: executing "%s.%s"' % (self.get_name(), key)
- if (tries_per_key[key] > 1):
+ if tries_per_key[key] > 1:
msg = msg + " (retry #%d)" % tries_per_key[key]
print(msg)
- if (self.periodic_render(key)):
+ if self.periodic_render(key):
self.last_runs[key] = time.time()
- @invokation_logged
- def periodic_render(self, key):
+ @invocation_logged
+ @abstractmethod
+ def periodic_render(self, key) -> bool:
pass
+ def get_name(self) -> str:
+ return self.__class__.__name__
+
+
class debuggable_abstaining_renderer(abstaining_renderer):
- def __init__(self, name_to_timeout_dict, debug):
- super(debuggable_abstaining_renderer, self).__init__(name_to_timeout_dict);
+ def __init__(self, name_to_timeout_dict: Dict[str, int], debug: bool) -> None:
+ super(debuggable_abstaining_renderer, self).__init__(name_to_timeout_dict)
self.debug = debug
- def debug_prefix(self):
+ def debug_prefix(self) -> str:
return self.get_name()
- def being_debugged(self):
+ def being_debugged(self) -> bool:
return self.debug
- def debug_print(self, template, *args):
+ def debug_print(self, template: str, *args) -> None:
try:
if self.being_debugged():
if args: