def execute_meb()

in mysqloperator/backup_main.py [0:0]

• 33 lines of code
• 4 McCabe index (conditional complexity)

def execute_meb(backup: MySQLBackup, backup_source: dict, backup_name: str, logger: logging.Logger) -> dict:
    cert = ("/tls/client.pem", "/tls/client.key")
    ca = "/tls/ca.pem"

    backup_obj = MySQLBackup.read(backup.name, backup.namespace)

    profile = backup_obj.get_profile().meb

    request = {
        "spec": profile,
        "source": backup_source
    }

    if "s3" in profile.storage:
        request["secret"] = get_secret(profile.storage["s3"]["credentials"],
                                       backup.namespace, logger)
    elif "oci" in profile.storage:
        request["secret"] = get_secret(profile.storage["oci"]["credentials"],
                                       backup.namespace, logger)
    else:
        raise Exception("Need either meb or s3 storage specification")

    request["incremental"] = backup.parsed_spec.incremental
    request["incremental_base"] = backup.parsed_spec.incrementalBase

    request_s = json.dumps(request, default=lambda o: o.__dict__, indent=4)

    name = urlquote(backup_name, safe="")
    url = f"https://{backup_source['host']}:4443/backup/{name}"
    logger.info(f"Triggering {url}")

    response = requests.post(url, data=request_s,
                             cert=cert, verify=False) #, verify=ca)

    print(response.content.decode())

    if response.status_code != 200:
        raise Exception(f"Failed to take backup. Backup daemon returned {response.status_code}: {response.content}")

    info = {
        "method": "meb",
        "source": f"{backup_source['user']}@{backup_source['host']}:{backup_source['port']}",
        #"spaceAvailable": f"{gb_avail:.4}G",
        #"size": f"{backup_size:.4}G"
    }
    return info