Method get

code_review_graph/registry.py:233–265

Get or create a connection for the given database path. Args: db_path: Path to the SQLite database file. Returns: An open SQLite connection.

(self, db_path: str)

        self._lock = threading.Lock()

    def get(self, db_path: str) -> sqlite3.Connection:
        """Get or create a connection for the given database path.

        Args:
            db_path: Path to the SQLite database file.

        Returns:
            An open SQLite connection.
        """
        key = str(Path(db_path).resolve())
        with self._lock:
            if key in self._pool:
                self._pool.move_to_end(key)
                return self._pool[key]

            # Evict LRU if full
            while len(self._pool) >= self._max_size:
                evict_key, evict_conn = self._pool.popitem(last=False)
                try:
                    evict_conn.close()
                except sqlite3.Error:
                    logger.debug("Failed to close evicted connection: %s", evict_key)
                logger.debug("Evicted connection: %s", evict_key)

            conn = sqlite3.connect(
                key, timeout=30, check_same_thread=False,
                isolation_level=None,
            )
            conn.row_factory = sqlite3.Row
            conn.execute("PRAGMA journal_mode=WAL")
            conn.execute("PRAGMA busy_timeout=5000")
            self._pool[key] = conn
            return conn

    def close_all(self) -> None:
        """Close all connections in the pool."""