renderer.py

   1 #!/usr/bin/env python3
   2
   3 from abc import ABC, abstractmethod
   4 from decorators import invocation_logged
   5 import logging
   6 import time
   7 from typing import Dict, Optional, Set
   8
   9
  10 logger = logging.getLogger(__file__)
  11
  12
  13 class renderer(ABC):
  14     """Base class for something that can render."""
  15
  16     @abstractmethod
  17     def render(self):
  18         pass
  19
  20     @abstractmethod
  21     def get_name(self):
  22         pass
  23
  24
  25 class abstaining_renderer(renderer):
  26     """A renderer that doesn't do it all the time."""
  27
  28     def __init__(self, name_to_timeout_dict: Dict[str, int]) -> None:
  29         self.name_to_timeout_dict = name_to_timeout_dict
  30         self.last_runs = {}
  31         for key in name_to_timeout_dict:
  32             self.last_runs[key] = 0.0
  33
  34     def should_render(self, keys_to_skip: Set[str]) -> Optional[str]:
  35         now = time.time()
  36         for key in self.name_to_timeout_dict:
  37             if (
  38                 (now - self.last_runs[key]) > self.name_to_timeout_dict[key]
  39             ) and key not in keys_to_skip:
  40                 return key
  41         return None
  42
  43     @invocation_logged
  44     def render(self) -> None:
  45         tries_per_key: Dict[str, int] = {}
  46         keys_to_skip: Set[str] = set()
  47         while True:
  48             key = self.should_render(keys_to_skip)
  49             if key is None:
  50                 logger.info(
  51                     f'renderer: Found nothing to do in "{self.get_name()}"; returning.'
  52                 )
  53                 break
  54
  55             if key in tries_per_key:
  56                 tries_per_key[key] += 1
  57             else:
  58                 tries_per_key[key] = 0
  59             op = f'{self.get_name()}.{key}'
  60
  61             if tries_per_key[key] >= 3:
  62                 logger.warning(
  63                     f'renderer: Too many failures in "{op}"; giving up.'
  64                 )
  65                 keys_to_skip.add(key)
  66             else:
  67                 msg = f'renderer: executing "{op}"'
  68                 if tries_per_key[key] > 1:
  69                     msg = msg + f' (retry #{tries_per_key[key]})'
  70                 logger.info(msg)
  71                 if self.periodic_render(key):
  72                     logger.debug(f'renderer: {op} succeeded.')
  73                     self.last_runs[key] = time.time()
  74                 else:
  75                     logger.warning(f'renderer: {op} failed; returned False.')
  76
  77     @invocation_logged
  78     @abstractmethod
  79     def periodic_render(self, key) -> bool:
  80         pass
  81
  82     def get_name(self) -> str:
  83         return self.__class__.__name__