Method _generate_test_data

database-lib/database_lib/main.py:218–240

(self, num_rows: int, batch_size: int = 10000)

                logger.debug(f"Attempted to delete non-existing key: {key}")

    def _generate_test_data(self, num_rows: int, batch_size: int = 10000):
        logger.info("Generating test data")
        self.ensure_connection()
        with self.connection:
            # Fetch a random key-value pair just once
            self.cursor.execute(
                "SELECT key, value FROM cache ORDER BY RANDOM() LIMIT 1"
            )
            key, value = self.cursor.fetchone()

            # Use a generator to create batches of test data
            def data_generator():
                for i in range(num_rows):
                    yield (f"{uuid.uuid4()}", value)

            # Process data in batches
            for offset in range(0, num_rows, batch_size):
                batch = list(islice(data_generator(), batch_size))
                self.cursor.executemany("INSERT INTO cache VALUES (?, ?)", batch)
                self.connection.commit()
                print(
                    f"Inserted {min(offset + batch_size, num_rows)} / {num_rows} rows"
                )

    def maintenance(self, time: Optional[int] = None, blocking_time: float = 0.5):
        connection = sqlite3.connect(self.database)