Function upsert_node

skills/wiki-store/scripts/wiki_db.py:288–349

Insert or update a wiki node and optionally refresh its Markdown mirror.

(
    *,
    user_id: str,
    node_id: str,
    node_type: str,
    title: str,
    body: str,
    metadata: Optional[Dict[str, Any]] = None,
    keywords: Optional[str] = None,
    source_type: Optional[str] = None,
    source_ref: Optional[str] = None,
    file_path: Optional[str] = None,
    write_mirror: bool = True,
)

def upsert_node(
    *,
    user_id: str,
    node_id: str,
    node_type: str,
    title: str,
    body: str,
    metadata: Optional[Dict[str, Any]] = None,
    keywords: Optional[str] = None,
    source_type: Optional[str] = None,
    source_ref: Optional[str] = None,
    file_path: Optional[str] = None,
    write_mirror: bool = True,
) -> Dict[str, Any]:
    """Insert or update a wiki node and optionally refresh its Markdown mirror."""
    init_wiki_schema()
    now = _now()
    relative_path = file_path or _relative_file_path(user_id, node_type, node_id)
    conn = db_ops.get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """
        INSERT INTO wiki_nodes
            (node_id, user_id, node_type, title, body, metadata_json,
             keywords, file_path, source_type, source_ref, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(user_id, node_id) DO UPDATE SET
            node_type=excluded.node_type,
            title=excluded.title,
            body=excluded.body,
            metadata_json=excluded.metadata_json,
            keywords=excluded.keywords,
            file_path=excluded.file_path,
            source_type=excluded.source_type,
            source_ref=excluded.source_ref,
            updated_at=excluded.updated_at
        """,
        (
            node_id,
            user_id,
            node_type,
            title or node_id,
            body or "",
            _json_dumps(metadata),
            keywords or "",
            relative_path,
            source_type,
            source_ref,
            now,
            now,
        ),
    )
    conn.commit()
    row = cursor.execute(
        "SELECT * FROM wiki_nodes WHERE user_id = ? AND node_id = ?",
        (user_id, node_id),
    ).fetchone()
    conn.close()