from .utils import deindex
STAR = os.getenv("DELTA_VELORUM", "B")
+PATH = os.getenv("DELTA_PATH")
+
import argo.admin as admin
from argo.database.read import read_views as database_read_views
auth = HTTPBasicAuth()
CORS(app)
+app.root_path = f"{PATH}/argo/"
+
# import authentication data
-with open(f"{app.root_path}/../../secrets.json", "r", encoding="utf-8") as f:
+with open(f"{PATH}/../secrets.json", "r", encoding="utf-8") as f:
users = json.loads(f.read())
users = {k: generate_password_hash(v) for (k, v) in users.items()}
@app.route("/script/<path:filename>", methods=["get"])
def script(filename):
- return send_from_directory(os.path.join(app.root_path, "build"), filename)
+ return send_from_directory(os.path.join(PATH, "argo", "build"), filename)
@app.route("/db", methods=["get"])
@auth.login_required
def download_db():
- return send_from_directory(f"{app.root_path}/../data", "main.db")
+ return send_from_directory(f"{PATH}/data", "main.db")
from argo import STAR
def admin_main():
- from argo import app
- assets = glob(os.path.join(app.root_path, "static", "media", "assets", "*"))
+ assets = glob(os.path.join(os.getenv("DELTA_PATH"), "argo", "static", "media", "assets", "*"))
assets = [path.split("argo")[-1] for path in assets]
return render_template(
check if file exists and pragam check the database on init
"""
-from .utils import integrity_check, optimize
import sqlite3
import os
-assert os.path.exists("./data/main.db"), "Database missing"
+PATH = f"{os.getenv('DELTA_PATH')}/data/main.db"
+
+from .utils import integrity_check, optimize
+
+assert os.path.exists(PATH), "Database missing"
assert integrity_check(), "Database integrity error"
optimize()
import sqlite3
from .utils import list_all, list_views
+from . import PATH
def read_all_current():
tables = list_all()
indexes = read_indexes()
- with sqlite3.connect("./data/main.db") as connection:
+ with sqlite3.connect(PATH) as connection:
cursor = connection.cursor()
data = {}
"""
reads a table by name
"""
- with sqlite3.connect("./data/main.db") as connection:
+ with sqlite3.connect(PATH) as connection:
cursor = connection.cursor()
cursor.execute(f"SELECT * FROM {table};")
return data
def read_indexes():
- with sqlite3.connect("./data/main.db") as connection:
+ with sqlite3.connect(PATH) as connection:
cursor = connection.cursor()
cursor.execute("SELECT * FROM indexes;")
import sqlite3
+from . import PATH
+
def integrity_check():
"""
check the database is ok and return true if it is
"""
- with sqlite3.connect("./data/main.db") as connection:
+ with sqlite3.connect(PATH) as connection:
cursor = connection.cursor()
integrity = cursor.execute('PRAGMA integrity_check')
"""
check the database is ok and return true if it is
"""
- with sqlite3.connect("./data/main.db") as connection:
+ with sqlite3.connect(PATH) as connection:
cursor = connection.cursor()
integrity = cursor.execute('PRAGMA optimize')
"""
get a list of all the tables and views in the database
"""
- with sqlite3.connect("./data/main.db") as connection:
+ with sqlite3.connect(PATH) as connection:
cursor = connection.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type IN ('table', 'view');")
return [table[0] for table in cursor.fetchall()]
"""
get a list of all the views in the database
"""
- with sqlite3.connect("./data/main.db") as connection:
+ with sqlite3.connect(PATH) as connection:
cursor = connection.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='view';")
return [table[0] for table in cursor.fetchall()]
from .utils import list_all
from .read import read
+from . import PATH
+
def write(data):
- with sqlite3.connect("./data/main.db") as connection:
+ with sqlite3.connect(PATH) as connection:
cursor = connection.cursor()
for key in data: