chendelian
166007b460
All checks were successful
技能自动化发布 / release (push) Successful in 9s
新增 DATA_SOURCES.md、data-sources 示例与双后端 connection 骨架；CLI 与宿主 UI 可读写同一库。要求匠厂 Host ≥ 6.9.28。 Co-authored-by: Cursor <cursoragent@cursor.com>
61 lines
1.7 KiB
Python
61 lines
1.7 KiB
Python
"""SQLite / MySQL 连接与幂等初始化模板。
|
||
|
||
模板默认只建数据管理元数据表。业务表由复制后的技能自行 CREATE，并同步展示元数据。
|
||
配置了 MYSQL_DSN 时走 MySQL 云库，否则使用本地 {skill_slug}.db。见 references/DATA_SOURCES.md。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
|
||
from db.backend import get_mysql_dsn, parse_mysql_dsn, resolve_backend
|
||
from db.cursor_compat import AdaptedConnection
|
||
from db.display_metadata import init_display_metadata
|
||
from util.runtime_paths import get_db_path
|
||
|
||
|
||
def get_conn() -> AdaptedConnection:
|
||
backend = resolve_backend()
|
||
if backend == "mysql":
|
||
import pymysql
|
||
|
||
params = parse_mysql_dsn(get_mysql_dsn())
|
||
raw = pymysql.connect(
|
||
host=str(params["host"]),
|
||
port=int(params["port"]),
|
||
user=str(params["user"]),
|
||
password=str(params["password"]),
|
||
database=str(params["database"]),
|
||
charset="utf8mb4",
|
||
autocommit=False,
|
||
)
|
||
return AdaptedConnection(raw, "mysql")
|
||
|
||
raw = sqlite3.connect(get_db_path())
|
||
return AdaptedConnection(raw, "sqlite")
|
||
|
||
|
||
def init_db() -> None:
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
init_display_metadata(cur)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def describe_active_store() -> dict[str, str]:
|
||
backend = resolve_backend()
|
||
if backend == "mysql":
|
||
params = parse_mysql_dsn(get_mysql_dsn())
|
||
return {
|
||
"backend": "mysql",
|
||
"host": str(params["host"]),
|
||
"database": str(params["database"]),
|
||
}
|
||
return {
|
||
"backend": "sqlite",
|
||
"db_path": get_db_path(),
|
||
}
|