How to filter efficiently huge list by multiple rules?

user_3479860 0
Viewed 58

I am writing an open-source PyPi package, that should filter the AWS EC2 instances.

In my function ec_compare__from_dict, I am filtering a list of 350+ elements that takes 364Kb on disk.

The following example of execution returns 1 filtered element:

>>> ec_compare__from_dict(_partial=_partial,InstanceType='z1d',FreeTierEligible=False,SupportedUsageClasses='spot',BareMetal=True)
[{'InstanceType': 'z1d.metal', 'CurrentGeneration': True, 'FreeTierEligible': False, 'SupportedUsageClasses': ['on-demand', 'spot'], 'SupportedRootDeviceTypes': ['ebs'], 'BareMetal': True, 'ProcessorInfo': {'SupportedArchitectures': ['x86_64'], 'SustainedClockSpeedInGhz': 4.0}, 'VCpuInfo': {'DefaultVCpus': 48}, 'MemoryInfo': {'SizeInMiB': 393216}, 'InstanceStorageSupported': True, 'InstanceStorageInfo': {'TotalSizeInGB': 1800, 'Disks': [{'SizeInGB': 900, 'Count': 2, 'Type': 'ssd'}]}, 'EbsInfo': {'EbsOptimizedSupport': 'default', 'EncryptionSupport': 'supported'}, 'NetworkInfo': {'NetworkPerformance': '25 Gigabit', 'MaximumNetworkInterfaces': 15, 'Ipv4AddressesPerInterface': 50, 'Ipv6AddressesPerInterface': 50, 'Ipv6Supported': True, 'EnaSupport': 'required'}, 'PlacementGroupInfo': {'SupportedStrategies': ['cluster', 'partition', 'spread']}, 'HibernationSupported': False, 'BurstablePerformanceSupported': False, 'DedicatedHostsSupported': True, 'AutoRecoverySupported': False}]

My problem is the following: I want to filter the list with all filters that have different rules in one single list comprehension.

But I am losing readability and I am creating a spaghetti code. Please point me to the better design decisions.

from typing import List


def ec2keys(*arg) -> List:
    values = {'str': ['InstanceType', 'Hypervisor'], 'bool': ['FreeTierEligible', 'HibernationSupported', 'CurrentGeneration', 'BurstablePerformanceSupported', 'AutoRecoverySupported', 'DedicatedHostsSupported', 'InstanceStorageSupported', 'BareMetal'], 'list': ['SupportedUsageClasses', 'SupportedRootDeviceTypes'], 'dict': ['InstanceStorageInfo', 'VCpuInfo', 'EbsInfo', 'FpgaInfo', 'PlacementGroupInfo', 'GpuInfo', 'InferenceAcceleratorInfo', 'MemoryInfo', 'NetworkInfo', 'ProcessorInfo'], 'other': []} 
    return [elem for k,v in  values.items() if k in arg or not arg for elem in v]

def ec_compare__from_dict(_partial: List,**kwargs):
    _instance_type = kwargs.get('InstanceType')
    flat_keys = set(ec2keys('str', 'bool')).intersection(
        set(kwargs.keys())) - {'InstanceType'}
    complex_filter_keys = set(ec2keys()).intersection(
        set(kwargs.keys()))
    list_keys_dict = {k: list(
        (lambda x: x if isinstance(x, list) else [x])(kwargs.get(k)))
        for k in set(ec2keys('list')).intersection(
            set(kwargs.keys()))
    }
    # here I started with list comprehension
    _partial = [x for x in _partial
                if all(elem in x.keys() for elem in flat_keys)
                and all(elem in x.keys() for elem in complex_filter_keys)
                and all(x[elem] == kwargs[elem] for elem in flat_keys)
                ]
    # this is re-apply filter again to all elements 
    if isinstance(_instance_type, str) and _instance_type:
        _partial = [x for x in _partial
                    if str(x['InstanceType']).startswith(_instance_type)
                    ]
    elif isinstance(_instance_type, (list, set)) and _instance_type:
        _partial = [x for x in _partial
                    if any(str(x['InstanceType']).startswith(elem)
                           for elem in _instance_type)
                    ]

    # this is how I filter list values
    if list_keys_dict:
        _partial = [x for x in _partial
                    if any(set(x[k]).intersection(v) for k, v in list_keys_dict.items())
                    ]
    return _partial

Example data

_partial = [{'InstanceType': 'z1d.metal', 'CurrentGeneration': True, 'FreeTierEligible': False, 'SupportedUsageClasses': ['on-demand', 'spot'], 'SupportedRootDeviceTypes': ['ebs'], 'BareMetal': True, 'ProcessorInfo': {'SupportedArchitectures': ['x86_64'], 'SustainedClockSpeedInGhz': 4.0}, 'VCpuInfo': {'DefaultVCpus': 48}, 'MemoryInfo': {'SizeInMiB': 393216}, 'InstanceStorageSupported': True, 'InstanceStorageInfo': {'TotalSizeInGB': 1800, 'Disks': [{'SizeInGB': 900, 'Count': 2, 'Type': 'ssd'}]}, 'EbsInfo': {'EbsOptimizedSupport': 'default', 'EncryptionSupport': 'supported'}, 'NetworkInfo': {'NetworkPerformance': '25 Gigabit', 'MaximumNetworkInterfaces': 15, 'Ipv4AddressesPerInterface': 50, 'Ipv6AddressesPerInterface': 50, 'Ipv6Supported': True, 'EnaSupport': 'required'}, 'PlacementGroupInfo': {'SupportedStrategies': ['cluster', 'partition', 'spread']}, 'HibernationSupported': False, 'BurstablePerformanceSupported': False, 'DedicatedHostsSupported': True, 'AutoRecoverySupported': False}]
list-comprehension python-3.x design-patterns python
1 Answers

Because of your nested list and dict structure, I think a class comparison is not the easiest one. But in a class comparison you can generate a comparison method for every item separately, which will cut the big function in many small ones. This will lead to some maintaining issues if the interface changes.

Your dictionary comparison approach is better in that case, but I would rewrite it using recursion for the nested dictionaries. By using recursion, you can simplify the nesting a bit.

By using your provided input:

data = {
    'InstanceType': 'z1d.metal',
    'CurrentGeneration': True,
    'FreeTierEligible': False,
    'SupportedUsageClasses': ['on-demand', 'spot'],
    'SupportedRootDeviceTypes': ['ebs'],
    'BareMetal': True,
    'ProcessorInfo': {'SupportedArchitectures': ['x86_64'],
                      'SustainedClockSpeedInGhz': 4.0},
    'VCpuInfo': {'DefaultVCpus': 48},
    'MemoryInfo': {'SizeInMiB': 393216},
    'InstanceStorageSupported': True,
    'InstanceStorageInfo': {'TotalSizeInGB': 1800,
                            'Disks': [{'SizeInGB': 900, 'Count': 2, 'Type': 'ssd'}]},
    'EbsInfo': {'EbsOptimizedSupport': 'default', 'EncryptionSupport': 'supported'},
    'NetworkInfo': {'NetworkPerformance': '25 Gigabit',
                    'MaximumNetworkInterfaces': 15,
                    'Ipv4AddressesPerInterface': 50,
                    'Ipv6AddressesPerInterface': 50,
                    'Ipv6Supported': True,
                    'EnaSupport': 'required'},
    'PlacementGroupInfo': {'SupportedStrategies': ['cluster', 'partition', 'spread']},
    'HibernationSupported': False,
    'BurstablePerformanceSupported': False,
    'DedicatedHostsSupported': True,
    'AutoRecoverySupported': False}

I generated a few possible filters (valid are True, invalid are False):

data_check_valid = {
    'InstanceType': 'z1d.metal',
    'InstanceStorageInfo': {'TotalSizeInGB': 1800},
    'PlacementGroupInfo': {'SupportedStrategies': ['spread']},
}

data_check_invalid_strategy = {
    'InstanceType': 'z1d.metal',
    'InstanceStorageInfo': {'TotalSizeInGB': 1800},
    'PlacementGroupInfo': {'SupportedStrategies': ['clustering']},  # Clustering is not supported.
}

data_check_invalid_count = {
    'InstanceType': 'z1d.metal',
    'InstanceStorageInfo': {'TotalSizeInGB': 1800,
                            'Disks': [{'SizeInGB': 900, 'Count': 4, 'Type': 'ssd'}]},  # Counts are unequal
}

Then we will compare the two dictionaries element for element, including nested elements. For this the following function is used:

def verify_element(original, check) -> bool:

    # Compare the types
    if type(original) != type(check):
        return False

    # recursively call this function for every element in the dictionary (if key exists)
    if isinstance(check, dict):
        for key, value in check.items():
            if key not in original:
                return False
            if not verify_element(value, original[key]):
                return False
        return True

    # The value inside check has to occur in any of the original elements
    # This behaviour is required, because we do not know where the check elements is positioned.
    if isinstance(check, (tuple, list)):
        for element in check:
            if not any(verify_element(each, element) for each in original):
                return False
        return True

    # Verify the element directly.
    if isinstance(check, (str, bool, int, float)):
        return original == check

    # Handle any unknown data types.
    raise TypeError(f"Type {type(check)}, with value {check} cannot be compared.")

To compare both dictionaries with each other, the final check will then look like this:

if __name__ == '__main__':
    print(verify_element(data, data_check_valid))             # True
    print(verify_element(data, data_check_invalid_strategy))  # False

    print(verify_element(data, data_check_invalid_count))     # False
    # When you change 'Count' to 2, the answer will become    # True

If you want to use this cleanly you can put it in a class and compare every element individually using the above function. This makes it possible to also include custom validators, such as should be bigger or smaller than the original value (which is currently impossible with the above code).

user_109613420
Related