dekun
47 lines
1.2 KiB
Python
47 lines
1.2 KiB
Python
"""SQLite 注册表."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
from typing import Any
|
|
|
|
|
|
def db_path() -> str:
|
|
return (os.getenv("DB_PATH") or "hub.db").strip() or "hub.db"
|
|
|
|
|
|
def get_db() -> sqlite3.Connection:
|
|
conn = sqlite3.connect(db_path(), timeout=30)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def init_db() -> None:
|
|
conn = get_db()
|
|
try:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS hub_instances (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
base_url TEXT NOT NULL,
|
|
bridge_token TEXT NOT NULL,
|
|
embed_path TEXT NOT NULL DEFAULT '/embed',
|
|
tags TEXT DEFAULT '',
|
|
enabled INTEGER NOT NULL DEFAULT 1,
|
|
sort_order INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def row_to_dict(row: sqlite3.Row | None) -> dict[str, Any] | None:
|
|
if row is None:
|
|
return None
|
|
return {k: row[k] for k in row.keys()}
|