109 lines
4.4 KiB
Python
109 lines
4.4 KiB
Python
"""SQLite 连接与幂等初始化。
|
||
|
||
``init_db`` 建两张表：
|
||
1. ``todos`` — 业务表（demo：待办事项）
|
||
2. ``_jiangchang_tables`` / ``_jiangchang_columns`` — 宿主数据管理展示元数据
|
||
|
||
业务表字段顺序就是宿主列展示顺序（cid 顺序）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
|
||
from db.display_metadata import (
|
||
init_display_metadata_tables,
|
||
seed_column_metadata_batch,
|
||
upsert_table_metadata,
|
||
)
|
||
from util.runtime_paths import get_db_path
|
||
from db.timestamp_columns import init_updated_at_trigger
|
||
|
||
ALL_BUSINESS_TABLES:tuple[str, ...] = ("todos")
|
||
|
||
# ---- 业务表 DDL（待办事项）----
|
||
# 字段顺序 = 数据管理列展示顺序；状态/优先级用 options_json 给宿主做下拉筛选
|
||
_TODOS_DDL = """
|
||
CREATE TABLE IF NOT EXISTS todos (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
title TEXT NOT NULL,
|
||
status TEXT NOT NULL DEFAULT 'pending',
|
||
priority TEXT NOT NULL DEFAULT 'normal',
|
||
due_at INTEGER,
|
||
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
|
||
updated_at INTEGER NOT NULL DEFAULT (unixepoch())
|
||
)
|
||
"""
|
||
|
||
_TODOS_INDEX = """
|
||
CREATE INDEX IF NOT EXISTS idx_todos_status ON todos(status)
|
||
"""
|
||
|
||
|
||
def get_conn() -> sqlite3.Connection:
|
||
return sqlite3.connect(get_db_path())
|
||
|
||
|
||
def _init_business_timestamp_triggers(cursor) -> None:
|
||
""" for table_name in ALL_BUSINESS_TABLES:
|
||
init_updated_at_trigger(cursor, table_name) """
|
||
init_updated_at_trigger(cursor, "todos")
|
||
|
||
|
||
def _seed_todos_display_metadata(cursor) -> None:
|
||
"""写 ``todos`` 表与字段的中文展示元数据。"""
|
||
upsert_table_metadata(
|
||
cursor,
|
||
table_name="todos",
|
||
display_name="待办事项",
|
||
description="本技能的核心业务表：每条待办事项一行。",
|
||
grid_help="所有的待办事项，包括待办、已办、已删除。",
|
||
visible=1,
|
||
readonly=1, # 整表只读：增删改通过技能 Action 走，宿主不直写
|
||
default_sort_json='[{"column":"created_at","direction":"desc"}]',
|
||
default_visible_columns_json='["title","status","priority","due_at","created_at"]',
|
||
)
|
||
seed_column_metadata_batch(
|
||
cursor,
|
||
table_name="todos",
|
||
columns=[
|
||
{"column_name": "id", "display_name": "编号", "description": "主键",
|
||
"visible": 1, "searchable": 1, "editable": 0, "display_type": "number"},
|
||
{"column_name": "title", "display_name": "标题", "description": "待办事项内容",
|
||
"grid_help": "简明描述要做的事", "visible": 1, "searchable": 1, "editable": 0,
|
||
"display_type": "text"},
|
||
{"column_name": "status", "display_name": "状态", "description": "pending/done",
|
||
"grid_help": "pending=待办，done=完成", "visible": 1, "searchable": 1, "editable": 0,
|
||
"display_type": "enum",
|
||
"options_json": '[{"value":"pending","label":"待办"},{"value":"done","label":"完成"}]'},
|
||
{"column_name": "priority", "display_name": "优先级", "description": "high/normal/low",
|
||
"visible": 1, "searchable": 1, "editable": 0, "display_type": "enum",
|
||
"options_json": '[{"value":"high","label":"高"},{"value":"normal","label":"中"},{"value":"low","label":"低"}]'},
|
||
{"column_name": "due_at", "display_name": "截止时间", "description": "Unix 秒；可空",
|
||
"visible": 1, "searchable": 1, "editable": 0, "display_type": "datetime_unix_seconds"},
|
||
{"column_name": "created_at", "display_name": "创建时间", "description": "Unix 秒",
|
||
"visible": 1, "searchable": 0, "editable": 0, "display_type": "datetime_unix_seconds"},
|
||
{"column_name": "updated_at", "display_name": "更新时间", "description": "Unix 秒",
|
||
"visible": 0, "searchable": 0, "editable": 0, "display_type": "datetime_unix_seconds"},
|
||
],
|
||
)
|
||
|
||
|
||
def init_db() -> None:
|
||
"""幂等创建/迁移本地 SQLite 与数据管理元数据。"""
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
# 1. 元数据表（先建，方便其它步骤 upsert）
|
||
init_display_metadata_tables(cur)
|
||
# 2. 业务表
|
||
cur.execute(_TODOS_DDL)
|
||
cur.execute(_TODOS_INDEX)
|
||
_init_business_timestamp_triggers(cur)
|
||
# 3. 业务表展示元数据
|
||
_seed_todos_display_metadata(cur)
|
||
|
||
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|