(self, index_id: str)
| 340 | return idx_id |
| 341 | |
| 342 | def get_index(self, index_id: str) -> dict | None: |
| 343 | conn = sqlite3.connect(self.db_path) |
| 344 | conn.row_factory = sqlite3.Row |
| 345 | cur = conn.execute('SELECT * FROM indexes WHERE id=?', (index_id,)) |
| 346 | row = cur.fetchone() |
| 347 | if not row: |
| 348 | conn.close() |
| 349 | return None |
| 350 | idx = dict(row) |
| 351 | idx['metadata'] = json.loads(idx['metadata'] or '{}') |
| 352 | cur = conn.execute('SELECT original_filename, stored_path FROM index_documents WHERE index_id=?', (index_id,)) |
| 353 | docs = [{'filename': r[0], 'stored_path': r[1]} for r in cur.fetchall()] |
| 354 | idx['documents'] = docs |
| 355 | conn.close() |
| 356 | return idx |
| 357 | |
| 358 | def list_indexes(self) -> list[dict]: |
| 359 | conn = sqlite3.connect(self.db_path) |
no outgoing calls
no test coverage detected