| 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 |
|
|
