from threading import Thread
from typing import Any, Callable, NamedTuple, Optional
# ourselves
-from ircplom.events import AffectiveEvent, ExceptionEvent, QueueMixin
+from ircplom.events import (
+ AffectiveEvent, CrashingException, ExceptionEvent, QueueMixin)
from ircplom.irc_conn import (BaseIrcConnection, IrcConnAbortException,
IrcMessage, PORT_SSL)
if value is None:
if old_value is None:
if confirm:
- raise Exception
+ raise CrashingException('called to unset non-set entry')
del self._dict[key]
elif value_changed:
self._dict[key] = value
elif targeted is None:
self._dict[key] = [value]
else:
- raise Exception
+ raise CrashingException('called on non-list entry')
def get_force(self, key: str) -> tuple[Optional[int | str | list[str]],
bool]:
self._client_trigger('_on_connect')
except IrcConnAbortException as e:
self._log(f'failed to connect: {e}', alert=True)
+ # catch _all_ just so they exit readably with the main loop
except Exception as e: # pylint: disable=broad-exception-caught
- self._put(ExceptionEvent(e))
+ self._put(ExceptionEvent(CrashingException(e)))
self._log('connecting …', conn_setup=self._db.conn_setup)
Thread(target=connect, daemon=True, args=(self,)).start()
return wrap()(t_method=t_method, **kwargs)
+class CrashingException(BaseException):
+ 'To explicitly crash, because it should never happen, but explaining why.'
+
+
@dataclass
class ExceptionEvent(Event):
'To deliver Exception to main loop for handling.'
- exception: Exception
+ exception: CrashingException
@dataclass
break
if it_yield is not None:
self._put(it_yield)
+ # catch _all_ just so they exit readably with the main loop
except Exception as e: # pylint: disable=broad-exception-caught
- self._put(ExceptionEvent(e))
+ self._put(ExceptionEvent(CrashingException(e)))