1.0.0 : Web Implemented.
This commit is contained in:
47
web/database.py
Normal file
47
web/database.py
Normal file
@@ -0,0 +1,47 @@
|
||||
import sqlite3
|
||||
from flask import g
|
||||
from web.config import Config
|
||||
|
||||
def get_db(db_name):
|
||||
"""
|
||||
db_name: 'l2', 'l3', or 'web'
|
||||
"""
|
||||
db_attr = f'db_{db_name}'
|
||||
db = getattr(g, db_attr, None)
|
||||
|
||||
if db is None:
|
||||
if db_name == 'l2':
|
||||
path = Config.DB_L2_PATH
|
||||
elif db_name == 'l3':
|
||||
path = Config.DB_L3_PATH
|
||||
elif db_name == 'web':
|
||||
path = Config.DB_WEB_PATH
|
||||
else:
|
||||
raise ValueError(f"Unknown database: {db_name}")
|
||||
|
||||
# Connect with check_same_thread=False if needed for dev, but default is safer per thread
|
||||
db = sqlite3.connect(path)
|
||||
db.row_factory = sqlite3.Row
|
||||
setattr(g, db_attr, db)
|
||||
|
||||
return db
|
||||
|
||||
def close_dbs(e=None):
|
||||
for db_name in ['l2', 'l3', 'web']:
|
||||
db_attr = f'db_{db_name}'
|
||||
db = getattr(g, db_attr, None)
|
||||
if db is not None:
|
||||
db.close()
|
||||
|
||||
def query_db(db_name, query, args=(), one=False):
|
||||
cur = get_db(db_name).execute(query, args)
|
||||
rv = cur.fetchall()
|
||||
cur.close()
|
||||
return (rv[0] if rv else None) if one else rv
|
||||
|
||||
def execute_db(db_name, query, args=()):
|
||||
db = get_db(db_name)
|
||||
cur = db.execute(query, args)
|
||||
db.commit()
|
||||
cur.close()
|
||||
return cur.lastrowid
|
||||
Reference in New Issue
Block a user