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>
36 lines
1.0 KiB
Python
36 lines
1.0 KiB
Python
"""数据库后端选择：配置了 MYSQL_DSN 时走 MySQL，否则回退本地 SQLite。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Literal
|
||
from urllib.parse import unquote, urlparse
|
||
|
||
from jiangchang_skill_core import config
|
||
|
||
DbBackend = Literal["sqlite", "mysql"]
|
||
|
||
|
||
def get_mysql_dsn() -> str:
|
||
return (config.get("MYSQL_DSN") or "").strip()
|
||
|
||
|
||
def resolve_backend() -> DbBackend:
|
||
return "mysql" if get_mysql_dsn() else "sqlite"
|
||
|
||
|
||
def parse_mysql_dsn(dsn: str) -> dict[str, object]:
|
||
text = dsn.strip()
|
||
if not text.startswith(("mysql://", "mysql2://")):
|
||
raise ValueError("MYSQL_DSN must be a mysql:// URL.")
|
||
parsed = urlparse(text)
|
||
database = (parsed.path or "").lstrip("/")
|
||
if not database:
|
||
raise ValueError("MYSQL_DSN must include a database name.")
|
||
return {
|
||
"host": parsed.hostname or "127.0.0.1",
|
||
"port": parsed.port or 3306,
|
||
"user": unquote(parsed.username or ""),
|
||
"password": unquote(parsed.password or ""),
|
||
"database": database,
|
||
}
|