File size: 1,827 Bytes
6f8c8ab | 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 | import asyncio
import os
import threading
from typing import Generic, TypeVar
K = TypeVar("K")
V = TypeVar("V")
def assert_not_in_asyncio_loop():
try:
asyncio.get_running_loop()
raise AssertionError(
"This function should not be called from within an asyncio loop"
)
except RuntimeError:
pass
def module_logging_filename(module_name: str, logging_filename: str) -> str:
"""
Parse logging_filename = STEM EXTENSION,
return STEM - MODULE_NAME EXTENSION
Example:
module_name = "TestModule"
logging_filename = "dashboard.log"
STEM = "dashboard"
EXTENSION = ".log"
return "dashboard-TestModule.log"
"""
stem, extension = os.path.splitext(logging_filename)
return f"{stem}-{module_name}{extension}"
class ThreadSafeDict(Generic[K, V]):
"""A thread-safe dictionary that only allows certain operations."""
def __init__(self):
self._lock = threading.Lock()
self._dict: dict[K, V] = {}
def put_new(self, key: K, value: V):
with self._lock:
if key in self._dict:
raise KeyError(f"Key {key} already exists in {self._dict}")
self._dict[key] = value
def get_or_raise(self, key: K) -> V:
with self._lock:
value = self._dict.get(key)
if value is None:
raise KeyError(f"Key {key} not found in {self._dict}")
return value
def pop_or_raise(self, key: K) -> V:
with self._lock:
value = self._dict.pop(key)
if value is None:
raise KeyError(f"Key {key} not found in {self._dict}")
return value
def pop_all(self) -> dict[K, V]:
with self._lock:
d = self._dict
self._dict = {}
return d
|