base/backend/database.py
Hera Zhao d19183923c
Some checks are pending
Deploy Production / test (push) Waiting to run
Deploy Production / deploy (push) Blocked by required conditions
Test / unit-test (push) Waiting to run
Test / e2e-test (push) Blocked by required conditions
Test / build-check (push) Waiting to run
Initial template: Vue 3 + FastAPI + SQLite full-stack with K8s deployment
Extracted from oil project — business logic removed, auth/db/deploy infrastructure
generalized with APP_NAME placeholders.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 22:13:06 +00:00

63 lines
2.0 KiB
Python

import sqlite3
import os
import secrets
DB_PATH = os.environ.get("DB_PATH", "/data/app.db")
def get_db():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
return conn
def init_db():
"""Initialize database schema. Add your tables here."""
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
conn = get_db()
c = conn.cursor()
c.executescript("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password TEXT,
token TEXT UNIQUE NOT NULL,
role TEXT NOT NULL DEFAULT 'viewer',
display_name TEXT,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
action TEXT NOT NULL,
target_type TEXT,
target_id TEXT,
target_name TEXT,
detail TEXT,
created_at TEXT DEFAULT (datetime('now'))
);
""")
# Seed admin user if no users exist
count = c.execute("SELECT COUNT(*) FROM users").fetchone()[0]
if count == 0:
admin_token = os.environ.get("ADMIN_TOKEN", secrets.token_hex(24))
c.execute(
"INSERT INTO users (username, token, role, display_name) VALUES (?, ?, ?, ?)",
("admin", admin_token, "admin", "Admin"),
)
print(f"[INIT] Admin user created. Token: {admin_token}")
conn.commit()
conn.close()
def log_audit(conn, user_id, action, target_type=None, target_id=None, target_name=None, detail=None):
conn.execute(
"INSERT INTO audit_log (user_id, action, target_type, target_id, target_name, detail) "
"VALUES (?, ?, ?, ?, ?, ?)",
(user_id, action, target_type, str(target_id) if target_id else None, target_name, detail),
)