Atomically replace data for a batch of files in one transaction.
(
self, batch: list[tuple[str, list[NodeInfo], list[EdgeInfo], str]]
)
| 287 | self._invalidate_cache() |
| 288 | |
| 289 | def store_file_batch( |
| 290 | self, batch: list[tuple[str, list[NodeInfo], list[EdgeInfo], str]] |
| 291 | ) -> None: |
| 292 | """Atomically replace data for a batch of files in one transaction.""" |
| 293 | self._begin_immediate() |
| 294 | try: |
| 295 | for file_path, nodes, edges, fhash in batch: |
| 296 | self.remove_file_data(file_path) |
| 297 | for node in nodes: |
| 298 | self.upsert_node(node, file_hash=fhash) |
| 299 | for edge in edges: |
| 300 | self.upsert_edge(edge) |
| 301 | self._conn.commit() |
| 302 | except BaseException: |
| 303 | self._conn.rollback() |
| 304 | raise |
| 305 | self._invalidate_cache() |
| 306 | |
| 307 | def set_metadata(self, key: str, value: str) -> None: |
| 308 | self._conn.execute( |