renderer.py

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