def _process_depots()

in dusty/models/config.py [0:0]

• 61 lines of code
• 14 McCabe index (conditional complexity)

    def _process_depots(self, current_context_config):  # pylint: disable=R0912
        context_config = current_context_config
        # Check depot config section sanity
        if "depots" not in context_config["settings"]:
            context_config["settings"]["depots"] = dict()
        depot_sections = ["secret", "object", "state"]
        for depot_section in depot_sections:
            if depot_section not in context_config["settings"]["depots"]:
                context_config["settings"]["depots"][depot_section] = dict()
        # Support legacy depot configuration
        legacy_depot_names = [
            item for item in list(
                context_config["settings"].get("depots", dict())
            ) if item not in depot_sections
        ]
        legacy_section_map = {
            SecretDepotModel: "secret",
            ObjectDepotModel: "object",
            StateDepotModel: "state"
        }
        for depot_name in legacy_depot_names:
            try:
                depot_class = importlib.import_module(
                    f"dusty.tools.depots.{depot_name}.depot"
                ).Depot
                for depot_type in legacy_section_map:
                    if issubclass(depot_class, depot_type):
                        depot_section = legacy_section_map[depot_type]
                        context_config["settings"]["depots"][depot_section][depot_name] = \
                            context_config["settings"]["depots"][depot_name]
                        log.info("Legacy depot %s added to section %s", depot_name, depot_section)
                context_config["settings"]["depots"].pop(depot_name)
            except:
                log.exception("Failed to process legacy depot %s", depot_name)
        # Make depot instances
        for depot_section in depot_sections:
            for depot_name in list(context_config["settings"]["depots"][depot_section]):
                try:
                    depot_class = importlib.import_module(
                        f"dusty.tools.depots.{depot_name}.depot"
                    ).Depot
                    if depot_class.get_name() in self.context.depots[depot_section]:
                        continue  # Depot already enabled in first iteration
                    depot = depot_class(
                        self.context,
                        context_config["settings"]["depots"][depot_section][depot_name]
                    )
                    self.context.depots[depot_section][depot.get_name()] = depot
                    log.info("Enabled %s depot %s", depot_section, depot_name)
                    if isinstance(depot, SecretDepotModel):
                        self.context.set_meta("depots_resolved_secrets", 0)
                        context_config = self._depot_substitution(context_config)
                        log.info(
                            "Resolved %d secrets from depots",
                            self.context.get_meta("depots_resolved_secrets", 0)
                        )
                except ModuleNotFoundError:
                    log.warning(
                        "Depot %s is not enabled, probably needed module is not loaded yet",
                        depot_name
                    )
                except:
                    log.exception("Failed to enable depot %s", depot_name)
        #
        return context_config