Pull "is main thread" logic out to utils, force Database to be accessed on main

thread

src/Database.py

@@ -1,5 +1,5 @@
 import json, os, sqlite3, threading, time, typing
-from src import Logging
+from src import Logging, utils
 
 sqlite3.register_converter("BOOLEAN", lambda v: bool(int(v)))
 
@@ -309,6 +309,9 @@ class Database(object):
     def _execute_fetch(self, query: str,
             fetch_func: typing.Callable[[sqlite3.Cursor], typing.Any],
             params: typing.List=[]):
+        if not utils.is_main_thread():
+            raise RuntimeError("Can't access Database outside of main thread")
+
         printable_query = " ".join(query.split())
         start = time.monotonic()
 

src/IRCBot.py

@@ -38,7 +38,8 @@ class Bot(object):
             func: typing.Optional[typing.Callable[[], typing.Any]]=None
             ) -> typing.Any:
         func = func or (lambda: None)
-        if threading.current_thread() is threading.main_thread():
+
+        if utils.is_main_thread():
             returned = func()
             self._trigger_client.send(b"TRIGGER")
             return returned

src/utils/__init__.py

@@ -1,4 +1,4 @@
-import datetime, decimal, enum, io, ipaddress, re, typing
+import datetime, decimal, enum, io, ipaddress, re, threading, typing
 from src.utils import cli, consts, irc, http, parse, security
 
 class Direction(enum.Enum):
@@ -199,3 +199,6 @@ def is_ip(s: str) -> bool:
     except ValueError:
         return False
     return True
+
+def is_main_thread() -> bool:
+    return threading.current_thread() is threading.main_thread()