(user_id: str, node_ids: List[str])
| 459 | |
| 460 | |
| 461 | def get_citations_for_nodes(user_id: str, node_ids: List[str]) -> Dict[str, List[Dict[str, Any]]]: |
| 462 | init_wiki_schema() |
| 463 | if not node_ids: |
| 464 | return {} |
| 465 | placeholders = ",".join("?" for _ in node_ids) |
| 466 | conn = db_ops.get_connection() |
| 467 | rows = conn.execute( |
| 468 | f""" |
| 469 | SELECT * FROM wiki_citations |
| 470 | WHERE user_id = ? AND node_id IN ({placeholders}) |
| 471 | ORDER BY id ASC |
| 472 | """, |
| 473 | (user_id, *node_ids), |
| 474 | ).fetchall() |
| 475 | conn.close() |
| 476 | grouped: Dict[str, List[Dict[str, Any]]] = {} |
| 477 | for row in rows: |
| 478 | item = dict(row) |
| 479 | grouped.setdefault(item["node_id"], []).append(item) |
| 480 | return grouped |
| 481 | |
| 482 | |
| 483 | def list_nodes(user_id: str, node_type: Optional[str] = None, limit: int = 20) -> List[Dict[str, Any]]: |
nothing calls this directly
no test coverage detected