File size: 4,031 Bytes
2facf1f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | import logging
import subprocess
from typing import Any, Optional
def setup_logging(name: Optional[str] = None):
"""Configure and setup logging for the application"""
formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger = logging.getLogger(name or __name__)
logger.setLevel(logging.INFO)
if not logger.handlers:
logger.addHandler(console_handler)
return logger
class KernelBotError(Exception):
"""
This class represents an Exception that has been sanitized,
i.e., whose message can be safely displayed to the user without
risk of leaking internal bot details.
"""
def __init__(self, message, code: int = 400):
super().__init__(message)
self.http_code = code
def get_github_branch_name():
try:
result = subprocess.run(
["git", "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"],
capture_output=True,
text=True,
check=True,
)
return result.stdout.strip().split("/", 1)[1]
except subprocess.CalledProcessError:
return "main"
class LRUCache:
def __init__(self, max_size: int):
"""LRU Cache implementation, as functools.lru doesn't work in async code
Note: Implementation uses list for convenience because cache is small, so
runtime complexity does not matter here.
Args:
max_size (int): Maximum size of the cache
"""
self._cache = {}
self._max_size = max_size
self._q = []
def __getitem__(self, key: Any) -> Any | None:
if key not in self._cache:
return None
self._q.remove(key)
self._q.append(key)
return self._cache[key]
def __setitem__(self, key: Any, value: Any) -> None:
if key in self._cache:
self._q.remove(key)
self._q.append(key)
self._cache[key] = value
return
if len(self._cache) >= self._max_size:
self._cache.pop(self._q.pop(0))
self._cache[key] = value
self._q.append(key)
def __contains__(self, key: Any) -> bool:
return key in self._cache
def __len__(self) -> int:
return len(self._cache)
def invalidate(self):
"""Invalidate the cache, clearing all entries, should be called when updating the underlying
data in db
"""
self._cache.clear()
self._q.clear()
def format_time(nanoseconds: float | str, err: Optional[float | str] = None): # noqa: C901
if nanoseconds is None:
logging.warning("Expected a number, got None", stack_info=True)
return "–"
# really ugly, but works for now
nanoseconds = float(nanoseconds)
scale = 1 # nanoseconds
unit = "ns"
if nanoseconds > 2_000_000:
scale = 1000_000
unit = "ms"
elif nanoseconds > 2000:
scale = 1000
unit = "µs"
time_in_unit = nanoseconds / scale
if err is not None:
err = float(err)
err /= scale
if time_in_unit < 1:
if err:
return f"{time_in_unit} ± {err} {unit}"
else:
return f"{time_in_unit} {unit}"
elif time_in_unit < 10:
if err:
return f"{time_in_unit:.2f} ± {err:.3f} {unit}"
else:
return f"{time_in_unit:.2f} {unit}"
elif time_in_unit < 100:
if err:
return f"{time_in_unit:.1f} ± {err:.2f} {unit}"
else:
return f"{time_in_unit:.1f} {unit}"
else:
if err:
return f"{time_in_unit:.0f} ± {err:.1f} {unit}"
else:
return f"{time_in_unit:.0f} {unit}"
def limit_length(text: str, maxlen: int):
assert maxlen > 6
if len(text) > maxlen:
return text[: maxlen - 6] + " [...]"
else:
return text
|