def parse()

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


    def parse(self, spec: dict, prefix: str) -> None:
        # TODO other storage types ....
        storagespec = dget_dict(spec, "storage", prefix)

        if "ociObjectStorage" in storagespec:
            ocistorage = dget_dict(storagespec, "ociObjectStorage", prefix+".storage")
            self.oci_credentials = dget_str(ocistorage, "credentials", prefix+".storage.credentials")

        if "s3" in storagespec:
            s3storage = dget_dict(storagespec, "s3", prefix+".storage")
            self.s3_region = dget_str(s3storage, "region", prefix+".storage.s3.region")
            self.s3_bucket= dget_str(s3storage, "region", prefix+".storage.s3.bucket")
            self.s3_object_key_prefix = dget_str(s3storage, "region", prefix+".storage.s3.object_key_prefix")
            self.s3_credentials = dget_str(s3storage, "region", prefix+".storage.s3.credentials")
            self.s3_host = dget_str(s3storage, "region", prefix+".storage.s3.host", default_value="")

        if (not self.oci_credentials and not self.s3_bucket) or (self.oci_credentials and self.s3_bucket):
            raise kopf.TemporaryError("Need one of either s3 or ociObjectStorage for MEB Restore")

        self.full_backup = dget_str(spec, "fullBackup", prefix)
        self.incremental_backups = dget_list(spec, "incrementalBackups", prefix, default_value = [])

        if "pitr" in spec:
            pitr_spec = dget_dict(spec, "pitr", prefix)
            self.pitr_backup_file = dget_str(pitr_spec, "backupFile", prefix+".pitr", default_value="")
            self.pitr_binlog_name = dget_str(pitr_spec, "binlogName", prefix+".pitr", default_value="") # cluster.name)
            self.pitr_gtid_purge = dget_str(pitr_spec, "gtidPurge", prefix+".pitr", default_value="")

            if "end" in pitr_spec:
                pitr_end_spec = dget_dict(pitr_spec, "end", prefix+".pitr.end")

                if "afterGtids" in pitr_end_spec:
                    self.pitr_end_term = "SQL_AFTER_GTIDS"
                    self.pitr_end_value = dget_str(pitr_end_spec, "afterGtids",
                                                   prefix+".pitr.end.afterGtids")
                elif "beforeGtids" in pitr_end_spec:
                    self.pitr_end_term = "SQL_BEFORE_GTIDS"
                    self.pitr_end_value = dget_str(pitr_end_spec, "beforeGtids",
                                                   prefix+".pitr.end.beforeGtids")