def _load()

in mysqloperator/controller/innodbcluster/cluster_api.py [0:0]


    def _load(self, spec_root: dict, spec_specific: dict, where_specific: str) -> None:
        # initialize now or all instances will share the same list initialized before the ctor
        self.add_to_initconf_cbs: dict[str, List[AddToInitconfHandler]] = {}
        self.remove_from_sts_cbs: dict[str, List[RemoveFromStsHandler]] = {}
        self.add_to_sts_cbs: dict[str, List[AddToStsHandler]] = {}
        self.get_configmaps_cbs: dict[str, List[GetConfigMapHandler]] = {}
        self.get_add_to_svc_cbs: dict[str, List[AddToSvcHandler]] = {}
        self.get_svc_monitor_cbs: dict[str, List[GetSvcMonitorHandler]] = {}

        self.secretName = dget_str(spec_root, "secretName", "spec")

        if "tlsCASecretName" in spec_root:
            self.tlsCASecretName = dget_str(spec_root, "tlsCASecretName", "spec")
        else:
            self.tlsCASecretName = f"{self.name}-ca"

        if "tlsSecretName" in spec_root:
            self.tlsSecretName = dget_str(spec_root, "tlsSecretName", "spec")
        else:
            self.tlsSecretName = f"{self.name}-tls"

        if "tlsUseSelfSigned" in spec_root:
            self.tlsUseSelfSigned = dget_bool(spec_root, "tlsUseSelfSigned", "spec")

        self.instances = dget_int(spec_specific, "instances", where_specific)
        if "version" in spec_specific:
            self.version = dget_str(spec_specific, "version", where_specific)

        if "edition" in spec_root:
            self.edition = dget_enum(
                spec_root, "edition", "spec", default_value=config.OPERATOR_EDITION,
                enum_type=Edition)

            if "imageRepository" not in spec_root:
                self.imageRepository = config.DEFAULT_IMAGE_REPOSITORY

        if "imagePullPolicy" in spec_root:
            self.imagePullPolicy = dget_enum(
                spec_root, "imagePullPolicy", "spec",
                default_value=config.default_image_pull_policy,
                enum_type=ImagePullPolicy)

        if "imagePullSecrets" in spec_root:
            self.imagePullSecrets = dget_list(
                spec_root, "imagePullSecrets", "spec", content_type=dict)

        self.serviceAccountName = dget_str(spec_root, "serviceAccountName", "spec", default_value=f"{self.name}-sidecar-sa")
        self.roleBindingName = f"{self.name}-sidecar-rb"
        self.switchoverServiceAccountName = "mysql-switchover-sa"
        self.switchoverRoleBindingName = "mysql-switchover-rb"

        if "imageRepository" in spec_root:
            self.imageRepository = dget_str(spec_root, "imageRepository", "spec")

        if "podSpec" in spec_specific:  # TODO - replace with something more specific
            self.podSpec = dget_dict(spec_specific, "podSpec", where_specific)

        if "podAnnotations" in spec_specific:
            self.podAnnotations = dget_dict(spec_specific, "podAnnotations", where_specific)

        if "podLabels" in spec_specific:
            self.podLabels = dget_dict(spec_specific, "podLabels", where_specific)

        if "datadirVolumeClaimTemplate" in spec_specific:
            self.datadirVolumeClaimTemplate = dget_dict(spec_specific, "datadirVolumeClaimTemplate", where_specific)

        self.keyring = KeyringSpec(self.namespace, self.name)
        if "keyring" in spec_root:
            self.keyring.parse(dget_dict(spec_root, "keyring", "spec"), "spec.keyring")

        if "mycnf" in spec_root:
            self.mycnf = dget_str(spec_root, "mycnf", "spec")

        self.metrics = MetriscSpec(self.namespace, self.cluster_name)
        if "metrics" in spec_root:
            self.metrics.parse(dget_dict(spec_root, "metrics", "spec"), "spec.metrics")

        if cb := self.metrics.get_configmaps_cb():
            self.get_configmaps_cbs[InnoDBClusterSpecProperties.METRICS.value] = [cb]

        if cb := self.metrics.get_add_to_sts_cb():
            self.add_to_sts_cbs[InnoDBClusterSpecProperties.METRICS.value] = [cb]

        if cb := self.metrics.get_add_to_svc_cb():
            self.get_add_to_svc_cbs[InnoDBClusterSpecProperties.METRICS.value] = [cb]

        if cb := self.metrics.get_svc_monitor_cb():
            self.get_svc_monitor_cbs[InnoDBClusterSpecProperties.METRICS.value] = [cb]

        self.logs = LogsSpec(self.namespace, self.cluster_name)
        if (section:= InnoDBClusterSpecProperties.LOGS.value) in spec_root:
            self.logs.parse(dget_dict(spec_root, section, "spec"), f"spec.{section}", getLogger())

            if cb := self.logs.get_configmaps_cb():
                self.get_configmaps_cbs[InnoDBClusterSpecProperties.LOGS.value] = [cb]

            if cb := self.logs.get_add_to_initconf_cb():
                self.add_to_initconf_cbs[InnoDBClusterSpecProperties.LOGS.value] = [cb]

            if cb := self.logs.get_remove_from_sts_cb():
                self.remove_from_sts_cbs[InnoDBClusterSpecProperties.LOGS.value] = [cb]

            if cb := self.logs.get_add_to_sts_cb():
                self.add_to_sts_cbs[InnoDBClusterSpecProperties.LOGS.value] = [cb]

        # Initialization Options
        section = InnoDBClusterSpecProperties.INITDB.value
        if section in spec_root:
            self.load_initdb(dget_dict(spec_root, section, "spec"))

        # TODO keep a list of base_server_id in the operator to keep things globally unique?
        if "baseServerId" in spec_specific:
            self.baseServerId = dget_int(spec_specific, "baseServerId", "spec")

        self.serviceFqdnTemplate = None
        if "serviceFqdnTemplate" in spec_root:
            self.serviceFqdnTemplate = dget_str(spec_root, "serviceFqdnTemplate",
                                                "spec",
                                                default_value="{service}.{namespace}.svc.{domain}")

        self.dataDirPermissions = DataDirPermissionsSpec()
        section = "datadirPermissions"
        if section in spec_root:
            self.dataDirPermissions.parse(dget_dict(spec_root, section, "spec"), f"spec.{section}")

        self.instanceService = InstanceServiceSpec()
        section = InnoDBClusterSpecProperties.INSTANCE_SERVICE.value
        if section in spec_root:
            self.instanceService.parse(dget_dict(spec_root, section, "spec"), f"spec.{section}")