def get_probes()

in mozilla_schema_generator/glean_ping.py [0:0]

• 48 lines of code
• 17 McCabe index (conditional complexity)

    def get_probes(self) -> List[GleanProbe]:
        data = self._get_json(self.probes_url)
        probes = list(data.items())

        for dependency in self.get_dependencies():
            dependency_probes = self._get_json(
                self.probes_url_template.format(dependency)
            )
            probes += list(dependency_probes.items())

        pings = self.get_pings()

        processed = []
        for _id, defn in probes:
            probe = GleanProbe(_id, defn, pings=pings)
            processed.append(probe)

            # Manual handling of incompatible schema changes
            issue_118_affected = {
                "fenix",
                "fenix-nightly",
                "firefox-android-nightly",
                "firefox-android-beta",
                "firefox-android-release",
            }
            if (
                self.repo_name in issue_118_affected
                and probe.get_name() == "installation.timestamp"
            ):
                logging.info(f"Writing column {probe.get_name()} for compatibility.")
                # See: https://github.com/mozilla/mozilla-schema-generator/issues/118
                # Search through history for the "string" type and add a copy of
                # the probe at that time in history. The changepoint signifies
                # this event.
                changepoint_index = 0
                for definition in probe.definition_history:
                    if definition["type"] != probe.get_type():
                        break
                    changepoint_index += 1
                # Modify the definition with the truncated history.
                hist_defn = defn.copy()
                hist_defn[probe.history_key] = probe.definition_history[
                    changepoint_index:
                ]
                hist_defn["type"] = hist_defn[probe.history_key][0]["type"]
                incompatible_probe_type = GleanProbe(_id, hist_defn, pings=pings)
                processed.append(incompatible_probe_type)

            # Handling probe type changes (Bug 1870317)
            probe_types = {hist["type"] for hist in defn[probe.history_key]}
            if len(probe_types) > 1:
                # The probe type changed at some point in history.
                # Create schema entry for each type.
                hist_defn = defn.copy()

                # No new entry needs to be created for the current probe type
                probe_types.remove(defn["type"])

                for hist in hist_defn[probe.history_key]:
                    # Create a new entry for a historic type
                    if hist["type"] in probe_types:
                        hist_defn["type"] = hist["type"]
                        probe = GleanProbe(_id, hist_defn, pings=pings)
                        processed.append(probe)

                        # Keep track of the types entries were already created for
                        probe_types.remove(hist["type"])

        return processed