omicron
b56c5314f8
- Add a function to automatically initialize the db with the schema if it isn't yet initialized - Remove the DROP TABLE queries from the schema in case the above^ goes wrong somehow - Make Item.write_to_db and Item.load_from_db take a db argument so that multiple threads can be supported This also changes the schema to never drop existing tables, just in case something goes wrong in the logic.
47 lines
878 B
Python
47 lines
878 B
Python
import atexit
|
|
import os
|
|
import sqlite3
|
|
from flask import g
|
|
|
|
|
|
_schema_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "schema.sql")
|
|
_db = None
|
|
_path = "items.sqlite3"
|
|
|
|
|
|
def set_db_path(f: str) -> None:
|
|
global _path
|
|
_path = f
|
|
|
|
|
|
def get_db() -> sqlite3.Connection:
|
|
global _db
|
|
if not _db:
|
|
_db = sqlite3.connect(
|
|
_path,
|
|
detect_types=sqlite3.PARSE_DECLTYPES,
|
|
)
|
|
_db.row_factory = sqlite3.Row
|
|
return _db
|
|
|
|
|
|
@atexit.register
|
|
def close_db() -> None:
|
|
global _db
|
|
if _db:
|
|
_db.close()
|
|
_db = None
|
|
|
|
|
|
def init_db() -> None:
|
|
db = get_db()
|
|
count = db.execute("SELECT count(*) FROM sqlite_master").fetchone()[0]
|
|
if count == 0:
|
|
create_db()
|
|
|
|
|
|
def create_db() -> None:
|
|
db = get_db()
|
|
with open(_schema_path, encoding="utf-8") as f:
|
|
db.executescript(f.read())
|