dekun
ef4b2f17ca
P1-P4: configurable nav/section visibility in system settings, full .env editor with restart badges, password change, runtime hot overrides, and single-instance pm2 restart. Works in hub embed iframe. Co-authored-by: Cursor <cursoragent@cursor.com>
72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
"""实例 SQLite 运行时配置（导航开关、env 热覆盖等）。"""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from typing import Any, Callable, Optional
|
|
|
|
RUNTIME_TABLE_SQL = """
|
|
CREATE TABLE IF NOT EXISTS app_runtime_settings (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
|
|
|
|
def ensure_runtime_settings_table(conn: sqlite3.Connection) -> None:
|
|
conn.execute(RUNTIME_TABLE_SQL)
|
|
conn.commit()
|
|
|
|
|
|
def runtime_get(conn: sqlite3.Connection, key: str) -> Optional[str]:
|
|
row = conn.execute(
|
|
"SELECT value FROM app_runtime_settings WHERE key=?",
|
|
(key,),
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
val = row["value"] if isinstance(row, sqlite3.Row) else row[0]
|
|
return None if val is None else str(val)
|
|
|
|
|
|
def runtime_set(conn: sqlite3.Connection, key: str, value: str) -> None:
|
|
now = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
|
|
conn.execute(
|
|
"INSERT INTO app_runtime_settings(key, value, updated_at) VALUES (?,?,?) "
|
|
"ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at",
|
|
(key, value, now),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def runtime_get_prefix(conn: sqlite3.Connection, prefix: str) -> dict[str, str]:
|
|
rows = conn.execute(
|
|
"SELECT key, value FROM app_runtime_settings WHERE key LIKE ?",
|
|
(prefix + "%",),
|
|
).fetchall()
|
|
out: dict[str, str] = {}
|
|
for row in rows:
|
|
k = row["key"] if isinstance(row, sqlite3.Row) else row[0]
|
|
v = row["value"] if isinstance(row, sqlite3.Row) else row[1]
|
|
if k.startswith(prefix):
|
|
out[k[len(prefix) :]] = v if v is not None else ""
|
|
return out
|
|
|
|
|
|
def runtime_set_many(conn: sqlite3.Connection, mapping: dict[str, str]) -> None:
|
|
for key, value in mapping.items():
|
|
runtime_set(conn, key, value)
|
|
|
|
|
|
def with_db(
|
|
get_db: Callable[[], sqlite3.Connection],
|
|
fn: Callable[[sqlite3.Connection], Any],
|
|
) -> Any:
|
|
conn = get_db()
|
|
try:
|
|
ensure_runtime_settings_table(conn)
|
|
return fn(conn)
|
|
finally:
|
|
conn.close()
|