def depricated_api_check()

in eksupdate/src/preflight_module.py [0:0]

• 47 lines of code
• 11 McCabe index (conditional complexity)

def depricated_api_check(log_details,errors,cluster_name,region,report,customer_report,update_version):
    loading_config(cluster_name,region)
    s3 = boto3.resource('s3')
    # Depricated API dictionary
    # depricated_api = s3.Object('eks-one-click-upgrade', 'depricatedApi')
    # depricated_api = depricated_api.get()['Body'].read().decode('utf-8')
    # depricated_api = json.loads(depricated_api)
    f = open('eksupdate/src/S3Files/depricatedApi',)
    depricated_api = json.load(f)
    #print(depricated_api)
    print('\n')
    log_pusher(log_details,cluster_name,region,"Checking for any depricated API being used....")
    print("Checking for any depricated API being used....")
    customer_report['depricated Api'] = []
    try :
        dict = depricated_api[update_version]
        # print(dict)
      
        for key in dict.keys():
            # print(key)
            if key == 'all-resources':
                for k in dict[key].keys():
                    if (dict[key][k] == "permanent") :
                        customer_report['depricated Api'].append(k + ' API has been depricated permanently ')
                        log_pusher(log_details,cluster_name,region,k + ' API has been depricated permanently ')
                        print(k + ' API has been depricated permanently ')
                    else :
                        customer_report['depricated Api'].append(k + ' API has been depricated use ' + dict[key][k] + ' instead')
                        log_pusher(log_details,cluster_name,region,k + ' API has been depricated use ' + dict[key][k] + ' instead')
                        print(k + ' API has been depricated use ' + dict[key][k] + ' instead')
            else : 
                depricated_resource = []
                new_resource = []
                v1 = eval(key)
                res = v1.get_api_resources()
                for resource in res.resources:
                    depricated_resource.append(resource.name)
                for k in dict[key].keys():
                    v2 = eval(k)
                    ret = v2.get_api_resources()
                    for resource in ret.resources:
                        new_resource.append(resource.name)
                    if (dict[key][k] in depricated_resource and dict[key][k] not in new_resource):
                        customer_report['depricated Api'].append("Resource "+ dict[key][k] + " is present in depricated API "+ key + " to be shifted to " + k)
                        errors.append("Resource "+ dict[key][k] + " is present in depricated API "+ key + " to be shifted to " + k)
                        log_pusher(log_details,cluster_name,region,"Resource "+ dict[key][k] + " is present in depricated API "+ key + " to be shifted to " + k)
                        print("Resource "+ dict[key][k] + " is present in depricated API "+ key + " to be shifted to " + k)
        log_pusher(log_details,cluster_name,region,"Depricated Api check completed")
        print("Depricated Api check completed")
    except Exception as e:
        errors.append("Depricated API check failed {err}".format(err=e))
        customer_report['depricated Api'].append("Depricated API check failed")
        log_pusher(log_details,cluster_name,region,"Depricated API check failed {err}".format(err=e))
        print("Depricated API check failed {err}".format(err=e))
        report['preflight_status'] = False