def parse()

in mysqloperator/controller/backup/backup_api.py [0:0]
33 lines of code
12 McCabe index (conditional complexity)

    def parse(self, spec: dict) -> Optional[ApiSpecError]:
        self.clusterName = dget_str(spec, "clusterName", "spec")
        self.backupProfileName = dget_str(spec, "backupProfileName", "spec", default_value="")
        self.backupProfile = self.parse_backup_profile(dget_dict(spec, "backupProfile", "spec", {}), "spec.backupProfile")
        self.deleteBackupData = dget_bool(spec, "deleteBackupData", "spec", default_value=False)
        self.timeZone = dget_str(spec, "timeZone", "spec", default_value="") #marking timeZone with default_value None will make it non-optional
        self.addTimestampToBackupDirectory = dget_bool(spec, "addTimestampToBackupDirectory", "spec", default_value=True)

        self.incremental = dget_bool(spec, "incremental", "spec", default_value=False)
        self.incrementalBase = dget_str(spec, "incrementalBase", "spec", default_value="last_backup")

        if self.backupProfileName and self.backupProfile:
            raise ApiSpecError("Only one of spec.backupProfileName or spec.backupProfile must be set")
        if not self.backupProfileName and not self.backupProfile:
            raise ApiSpecError("One of spec.backupProfileName or spec.backupProfile must be set")

        try:
            cluster = cluster_api.InnoDBCluster.read(self.namespace, self.clusterName)
        except ApiException as e:
            if e.status == 404:
                return ApiSpecError(f"Invalid clusterName {self.namespace}/{self.clusterName}")
            raise

        self.operator_image = cluster.parsed_spec.operator_image
        self.operator_image_pull_policy = cluster.parsed_spec.operator_image_pull_policy
        self.serviceAccountName = cluster.parsed_spec.serviceAccountName

        if self.backupProfileName:
            self.backupProfile = cluster.parsed_spec.get_backup_profile(self.backupProfileName)
            if not self.backupProfile:
                err_msg = f"Invalid backupProfileName '{self.backupProfileName}' in cluster {self.namespace}/{self.clusterName}"
                raise ApiSpecError(err_msg)

        if self.incremental:
            if not self.backupProfile.meb:
                raise ApiSpecError("Incremental Backup is only supported with a profile using MySQL Enterprise Backup")

            if self.incrementalBase not in ('last_backup', 'last_full_backup'):
                raise ApiSpecError("BackupBase must be last_backup or last_full_backup")


        return None