(user_id: str, node_type: Optional[str] = None, limit: int = 20)
| 481 | |
| 482 | |
| 483 | def list_nodes(user_id: str, node_type: Optional[str] = None, limit: int = 20) -> List[Dict[str, Any]]: |
| 484 | init_wiki_schema() |
| 485 | conn = db_ops.get_connection() |
| 486 | cursor = conn.cursor() |
| 487 | params: List[Any] = [user_id] |
| 488 | where = "WHERE user_id = ?" |
| 489 | if node_type: |
| 490 | where += " AND node_type = ?" |
| 491 | params.append(node_type) |
| 492 | params.append(max(1, int(limit))) |
| 493 | rows = cursor.execute( |
| 494 | f""" |
| 495 | SELECT * FROM wiki_nodes |
| 496 | {where} |
| 497 | ORDER BY updated_at DESC, id DESC |
| 498 | LIMIT ? |
| 499 | """, |
| 500 | params, |
| 501 | ).fetchall() |
| 502 | conn.close() |
| 503 | return _rows_to_nodes(rows) |
| 504 | |
| 505 | |
| 506 | def get_node(user_id: str, node_id: str) -> Optional[Dict[str, Any]]: |
no test coverage detected