def fetch_socorro_info()

in bugbot/crash/analyzer.py [0:0]

• 50 lines of code
• 4 McCabe index (conditional complexity)

    def fetch_socorro_info(self) -> tuple[list[dict], int]:
        """Fetch the signature data from Socorro."""
        if not self._signatures:
            return [], 0

        end_date = lmdutils.get_date_ymd("today")
        start_date = end_date - self.SUMMARY_DURATION
        date_range = socorro.SuperSearch.get_search_date(start_date, end_date)

        params = {
            "product": self._product,
            # TODO(investigate): should we included all release channels?
            "release_channel": self._channel,
            # TODO(investigate): should we limit based on the build date as well?
            "date": date_range,
            # TODO: split signatures into chunks to avoid very long query URLs
            "signature": ["=" + signature for signature in self._signatures],
            "_aggs.signature": [
                "address",
                "build_id",
                "cpu_arch",
                "proto_signature",
                "_cardinality.user_comments",
                "cpu_arch",
                "platform_pretty_version",
                "_histogram.date",
                "phc_alloc_stack",
                # The following are needed for SignatureStats:
                "platform",
                "is_garbage_collecting",
                "_cardinality.install_time",
                "startup_crash",
                "_histogram.uptime",
                "process_type",
                "moz_crash_reason",
            ],
            "_results_number": 0,
            "_facets_size": 10000,
        }

        def handler(search_results: dict, data: dict):
            data["num_total_crashes"] = search_results["total"]
            data["signatures"] = search_results["facets"]["signature"]

        logger.debug(
            "Fetch from Socorro: requesting info for %d signatures",
            len(self._signatures),
        )

        data: dict = {}
        socorro.SuperSearchUnredacted(
            params=params,
            handler=handler,
            handlerdata=data,
        ).wait()

        logger.debug(
            "Fetch from Socorro: received info for %d signatures",
            len(data["signatures"]),
        )

        return data["signatures"], data["num_total_crashes"]