Simplify SQLite database table creation

Table creation is a little complicated right now and I'm simplifying it
for clarity. Perhaps at a later date when each server/client database
has more tables, I will revisit it.
This commit is contained in:
Xevion
2021-05-30 16:30:14 -05:00
parent 0aa78628de
commit 5a6824f3a9

View File

@@ -1,8 +1,9 @@
import abc
import logging import logging
import os import os
import sqlite3 import sqlite3
import threading import threading
import abc from typing import List, Optional, Union
import constants import constants
@@ -14,11 +15,12 @@ lock = threading.Lock()
class Database(abc.ABC): class Database(abc.ABC):
def __init__(self, database: str): def __init__(self, database: str):
logger.debug(f"Connected to './{os.path.basename(database)}'")
self.conn = sqlite3.connect(database, detect_types=sqlite3.PARSE_DECLTYPES) self.conn = sqlite3.connect(database, detect_types=sqlite3.PARSE_DECLTYPES)
logger.debug(f"Connected to './{os.path.basename(database)}'")
self.__isClosed = False self.__isClosed = False
self.__table = None self._tables: List[str] = []
self.construct() self.construct()
logger.debug('Completed database construction.')
@property @property
def is_closed(self) -> bool: def is_closed(self) -> bool:
@@ -34,29 +36,17 @@ class Database(abc.ABC):
self.conn.close() self.conn.close()
self.__isClosed = True self.__isClosed = True
def construct(self) -> None:
with lock:
cur = self.conn.cursor()
try:
cur.execute('''SELECT name FROM sqlite_master WHERE type='table' AND name = ?;''', [self.__table])
if cur.fetchone() is None:
self._construct()
logger.info(f"'{self.__table}' table created.")
finally:
cur.close()
@abc.abstractmethod @abc.abstractmethod
def _construct(self) -> None: def construct(self) -> None:
pass self._construct()
class ClientDatabase(Database): class ClientDatabase(Database):
def __init__(self): def __init__(self):
super().__init__(constants.CLIENT_DATABASE) super().__init__(constants.CLIENT_DATABASE)
self.__table = 'connection'
def _construct(self) -> None: def construct(self) -> None:
self.conn.execute('''CREATE TABLE connection self.conn.execute('''CREATE TABLE IF NOT EXISTS connection
(id INTEGER PRIMARY KEY, (id INTEGER PRIMARY KEY,
address TEXT NOT NULL, address TEXT NOT NULL,
port INTEGER NOT NULL, port INTEGER NOT NULL,
@@ -72,8 +62,8 @@ class ServerDatabase(Database):
def __init__(self): def __init__(self):
super().__init__(constants.SERVER_DATABASE) super().__init__(constants.SERVER_DATABASE)
def _construct(self): def construct(self):
self.conn.execute('''CREATE TABLE message self.conn.execute('''CREATE TABLE IF NOT EXISTS message
(id INTEGER PRIMARY KEY, (id INTEGER PRIMARY KEY,
nickname TEXT NOT NULL, nickname TEXT NOT NULL,
connection_hash TEXT NOT NULL, connection_hash TEXT NOT NULL,
@@ -98,7 +88,7 @@ class ServerDatabase(Database):
try: try:
cur.execute('''INSERT INTO message (nickname, connection_hash, color, message, timestamp) cur.execute('''INSERT INTO message (nickname, connection_hash, color, message, timestamp)
VALUES (?, ?, ?, ?, ?)''', [nickname, user_hash, color, message, timestamp]) VALUES (?, ?, ?, ?, ?)''', [nickname, user_hash, color, message, timestamp])
logger.debug(f'Message {cur.lastrowid} recorded.') logger.debug(f'Message #{cur.lastrowid} recorded.')
return cur.lastrowid return cur.lastrowid
finally: finally:
cur.close() cur.close()