def count_objects()

in smallpond/execution/workqueue.py [0:0]

• 32 lines of code
• 18 McCabe index (conditional complexity)

def count_objects(obj, object_cnt=None, visited_objs=None, depth=0):
    object_cnt = {} if object_cnt is None else object_cnt
    visited_objs = set() if visited_objs is None else visited_objs

    if id(obj) in visited_objs:
        return
    else:
        visited_objs.add(id(obj))

    if isinstance(obj, dict):
        for key, value in obj.items():
            count_objects(key, object_cnt, visited_objs, depth + 1)
            count_objects(value, object_cnt, visited_objs, depth + 1)
    elif isinstance(obj, list) or isinstance(obj, tuple):
        for item in obj:
            count_objects(item, object_cnt, visited_objs, depth + 1)
    else:
        class_name = obj.__class__.__name__
        if class_name not in object_cnt:
            object_cnt[class_name] = (0, 0)
        cnt, size = object_cnt[class_name]
        object_cnt[class_name] = (cnt + 1, size + sys.getsizeof(obj))

        key_attributes = ("__self__", "__dict__", "__slots__")
        if not isinstance(obj, (bool, str, int, float, type(None))) and any(attr_name in key_attributes for attr_name in dir(obj)):
            logger.debug(f"{' ' * depth}{class_name}@{id(obj):x}")
            for attr_name in dir(obj):
                try:
                    if (not attr_name.startswith("__") or attr_name in key_attributes) and not isinstance(
                        getattr(obj.__class__, attr_name, None), property
                    ):
                        logger.debug(f"{' ' * depth}{class_name}.{attr_name}@{id(obj):x}")
                        count_objects(getattr(obj, attr_name), object_cnt, visited_objs, depth + 1)
                except Exception as ex:
                    logger.warning(f"failed to get '{attr_name}' from {repr(obj)}: {ex}")