def bugzilla_fetch_history()

in jobs/webcompat-kb/webcompat_kb/bugzilla.py [0:0]

• 42 lines of code
• 11 McCabe index (conditional complexity)

    def bugzilla_fetch_history(self, ids: Iterable[int]) -> HistoryByBug:
        history: dict[int, list[bugdantic.bugzilla.History]] = {}
        chunk_size = 100
        ids_list = list(ids)
        max_retries = 3

        for retry in range(max_retries):
            retry += 1
            retry_ids: list[int] = []

            chunks: list[list[int]] = []
            for i in range((len(ids_list) // chunk_size) + 1):
                offset = i * chunk_size
                bug_ids = ids_list[offset : offset + chunk_size]
                if bug_ids:
                    chunks.append(bug_ids)

            for i, bug_ids in enumerate(chunks):
                logging.info(
                    f"Fetching history from bugzilla for {','.join(str(item) for item in bug_ids)} ({i + 1}/{len(chunks)})"
                )
                try:
                    results = self.bz_client.search(
                        query={"id": bug_ids}, include_fields=["id", "history"]
                    )
                except Exception as e:
                    logging.warning(f"Search request failed:\n{e}")
                    results = []

                for bug in results:
                    assert bug.id is not None
                    assert bug.history is not None
                    history[bug.id] = bug.history

                # Add anything missing from the response to the retry list
                retry_ids.extend(item for item in bug_ids if item not in history)

                time.sleep(1)

            ids_list = retry_ids
            if retry_ids:
                time.sleep(10)
                logging.info(f"Retrying {len(retry_ids)} bugs with missing history")
            else:
                break

        if len(ids_list) != 0:
            raise BugLoadError(
                f"Failed to fetch bug history for {','.join(str(item) for item in ids_list)}"
            )

        return self.bugzilla_to_history_entry(history)