⚡️ Speed up function is_coroutine_callable by 142% in src/aviary/utils.py

[codeflash-ai]

📄 is_coroutine_callable() in src/aviary/utils.py

📈 Performance improved by 142% (1.42x faster)

⏱️ Runtime went down from 420 microseconds to 174 microseconds (best of 270 runs)

Explanation and details

Here's a more optimized version of the given Python program.

.

1. Removed the inspect module and used isinstance() with FunctionType and MethodType, which is faster.
2. Directly checked the co_flags of a function's code object to determine if it is a coroutine function. This bypasses the overhead of introspection functions.
3. Conditional checks for __call__ attribute only if necessary and evaluates one less function call when obj is a usual function or method.

Correctness verification

The new optimized code was tested for correctness. The results are listed below.

🔘 (none found) − ⚙️ Existing Unit Tests

✅ 1036 Passed − 🌀 Generated Regression Tests

(click to show generated tests)

import functools
import inspect

import pytest  # used for our unit tests
from src.aviary.utils import is_coroutine_callable

# unit tests

# Basic Functionality Tests
async def async_function():
    pass
    # Outputs were verified to be equal to the original implementation

def regular_function():
    pass
    # Outputs were verified to be equal to the original implementation

def test_coroutine_function():
    codeflash_output = is_coroutine_callable(async_function)  # Should be True for coroutine functions
    # Outputs were verified to be equal to the original implementation

def test_non_coroutine_function():
    codeflash_output = is_coroutine_callable(regular_function)  # Should be False for regular functions
    # Outputs were verified to be equal to the original implementation

# Method Tests
class ExampleClass:
    async def async_method(self):
        pass
        # Outputs were verified to be equal to the original implementation

    def regular_method(self):
        pass
        # Outputs were verified to be equal to the original implementation

def test_coroutine_method():
    codeflash_output = is_coroutine_callable(ExampleClass().async_method)  # Should be True for coroutine methods
    # Outputs were verified to be equal to the original implementation

def test_non_coroutine_method():
    codeflash_output = is_coroutine_callable(ExampleClass().regular_method)  # Should be False for regular methods
    # Outputs were verified to be equal to the original implementation

# Callable Object Tests
class CallableObject:
    async def __call__(self):
        pass
        # Outputs were verified to be equal to the original implementation

class NonCallableObject:
    def __call__(self):
        pass
        # Outputs were verified to be equal to the original implementation

def test_object_with_coroutine_call():
    codeflash_output = is_coroutine_callable(CallableObject())  # Should be True for objects with coroutine __call__
    # Outputs were verified to be equal to the original implementation

def test_object_with_non_coroutine_call():
    codeflash_output = is_coroutine_callable(NonCallableObject())  # Should be False for objects with non-coroutine __call__
    # Outputs were verified to be equal to the original implementation

# Non-Callable Object Tests
def test_non_callable_object():
    codeflash_output = is_coroutine_callable(object())  # Should be False for non-callable objects
    # Outputs were verified to be equal to the original implementation

def test_primitive_types():
    codeflash_output = is_coroutine_callable(42)  # Should be False for integers
    codeflash_output = is_coroutine_callable("string")  # Should be False for strings
    codeflash_output = is_coroutine_callable([1, 2, 3])  # Should be False for lists
    # Outputs were verified to be equal to the original implementation

# Edge Case Tests
def test_bound_coroutine_method():
    obj = ExampleClass()
    codeflash_output = is_coroutine_callable(obj.async_method)  # Should be True for bound coroutine methods
    # Outputs were verified to be equal to the original implementation

def test_bound_non_coroutine_method():
    obj = ExampleClass()
    codeflash_output = is_coroutine_callable(obj.regular_method)  # Should be False for bound regular methods
    # Outputs were verified to be equal to the original implementation

class DynamicCallable:
    def __init__(self):
        self.__call__ = lambda: None
        # Outputs were verified to be equal to the original implementation

def test_dynamic_callable():
    obj = DynamicCallable()
    codeflash_output = is_coroutine_callable(obj)  # Initially non-coroutine
    obj.__call__ = async_function
    codeflash_output = is_coroutine_callable(obj)  # After changing to coroutine
    # Outputs were verified to be equal to the original implementation

# Performance and Scalability Tests
def test_large_number_of_calls():
    for _ in range(1000):
        codeflash_output = is_coroutine_callable(async_function)  # Test performance with repeated calls
    # Outputs were verified to be equal to the original implementation

# Error Handling Tests
def test_none_input():
    codeflash_output = is_coroutine_callable(None)  # Should be False for None input
    # Outputs were verified to be equal to the original implementation

# Complex Object Tests
class NestedCallable:
    def __call__(self):
        return async_function
        # Outputs were verified to be equal to the original implementation

def test_nested_callable():
    obj = NestedCallable()
    codeflash_output = is_coroutine_callable(obj)  # __call__ returns a coroutine but is not coroutine itself
    # Outputs were verified to be equal to the original implementation

# Integration-Like Tests
class MixedClass:
    async def async_method(self):
        pass
        # Outputs were verified to be equal to the original implementation

    def __call__(self):
        return self.async_method
        # Outputs were verified to be equal to the original implementation

def test_mixed_class():
    obj = MixedClass()
    codeflash_output = is_coroutine_callable(obj)  # __call__ is not coroutine
    # Outputs were verified to be equal to the original implementation

# Rare or Unexpected Edge Cases
class Metaclass(type):
    async def __call__(cls):
        pass
        # Outputs were verified to be equal to the original implementation

class MetaClassWithCoroutine(metaclass=Metaclass):
    pass

def test_metaclass_with_coroutine_call():
    codeflash_output = is_coroutine_callable(MetaClassWithCoroutine)  # Metaclass with coroutine __call__
    # Outputs were verified to be equal to the original implementation

class Descriptor:
    async def __get__(self, instance, owner):
        pass
        # Outputs were verified to be equal to the original implementation

def test_descriptor():
    desc = Descriptor()
    codeflash_output = is_coroutine_callable(desc)  # Descriptors are not directly callable
    # Outputs were verified to be equal to the original implementation

class Proxy:
    def __init__(self, target):
        self._target = target
        # Outputs were verified to be equal to the original implementation

    def __call__(self):
        return self._target()
        # Outputs were verified to be equal to the original implementation

def test_proxy_with_coroutine_target():
    proxy = Proxy(async_function)
    codeflash_output = is_coroutine_callable(proxy)  # Proxy itself is not coroutine
    # Outputs were verified to be equal to the original implementation

# Run tests using pytest
if __name__ == "__main__":
    pytest.main()
import inspect  # used within the function to test

import pytest  # used for our unit tests
from src.aviary.utils import is_coroutine_callable

# unit tests

# Basic Functionality Tests

async def async_function():
    pass
    # Outputs were verified to be equal to the original implementation

def test_async_function():
    # Test with a simple coroutine function
    codeflash_output = is_coroutine_callable(async_function)
    # Outputs were verified to be equal to the original implementation

def regular_function():
    pass
    # Outputs were verified to be equal to the original implementation

def test_regular_function():
    # Test with a regular function
    codeflash_output = is_coroutine_callable(regular_function)
    # Outputs were verified to be equal to the original implementation

def test_lambda_function():
    # Test with a lambda function
    codeflash_output = is_coroutine_callable(lambda x: x)
    # Outputs were verified to be equal to the original implementation

# Methods Tests

class MyClass:
    async def async_method(self):
        pass
        # Outputs were verified to be equal to the original implementation

    def regular_method(self):
        pass
        # Outputs were verified to be equal to the original implementation

def test_async_method():
    # Test with a coroutine method in a class
    codeflash_output = is_coroutine_callable(MyClass().async_method)
    # Outputs were verified to be equal to the original implementation

def test_regular_method():
    # Test with a regular method in a class
    codeflash_output = is_coroutine_callable(MyClass().regular_method)
    # Outputs were verified to be equal to the original implementation

# Callable Objects Tests

class CallableAsync:
    async def __call__(self):
        pass
        # Outputs were verified to be equal to the original implementation

class CallableRegular:
    def __call__(self):
        pass
        # Outputs were verified to be equal to the original implementation

def test_callable_async():
    # Test with a callable class with coroutine __call__
    codeflash_output = is_coroutine_callable(CallableAsync())
    # Outputs were verified to be equal to the original implementation

def test_callable_regular():
    # Test with a callable class with regular __call__
    codeflash_output = is_coroutine_callable(CallableRegular())
    # Outputs were verified to be equal to the original implementation

# Non-Callable Objects Tests

def test_non_callable_objects():
    # Test with non-callable objects
    codeflash_output = is_coroutine_callable(42)
    codeflash_output = is_coroutine_callable("string")
    codeflash_output = is_coroutine_callable([1, 2, 3])
    # Outputs were verified to be equal to the original implementation

# Edge Cases Tests

def test_none_input():
    # Test with None as input
    codeflash_output = is_coroutine_callable(None)
    # Outputs were verified to be equal to the original implementation

def test_builtin_function():
    # Test with a built-in function
    codeflash_output = is_coroutine_callable(len)
    # Outputs were verified to be equal to the original implementation

# Decorated Functions Tests

def decorator(func):
    async def wrapper(*args, **kwargs):
        return await func(*args, **kwargs)
        # Outputs were verified to be equal to the original implementation
    return wrapper
    # Outputs were verified to be equal to the original implementation

@decorator
async def decorated_async_function():
    pass
    # Outputs were verified to be equal to the original implementation

def test_decorated_async_function():
    # Test with a decorated coroutine function
    codeflash_output = is_coroutine_callable(decorated_async_function)
    # Outputs were verified to be equal to the original implementation

# Performance and Scalability Tests

class LargeClass:
    def __init__(self):
        self.attributes = [i for i in range(1000)]
        # Outputs were verified to be equal to the original implementation

    async def __call__(self):
        pass
        # Outputs were verified to be equal to the original implementation

def test_large_class():
    # Test with a large class definition
    codeflash_output = is_coroutine_callable(LargeClass())
    # Outputs were verified to be equal to the original implementation

# Complex Scenarios Tests

def outer_function():
    async def inner_async_function():
        pass
        # Outputs were verified to be equal to the original implementation
    return inner_async_function
    # Outputs were verified to be equal to the original implementation

def test_nested_coroutine_function():
    # Test with a nested coroutine function
    codeflash_output = is_coroutine_callable(outer_function())
    # Outputs were verified to be equal to the original implementation

class BaseClass:
    async def method(self):
        pass
        # Outputs were verified to be equal to the original implementation

class SubClass(BaseClass):
    def method(self):
        pass
        # Outputs were verified to be equal to the original implementation

def test_inherited_method_override():
    # Test with a subclass that overrides a coroutine method with a regular method
    codeflash_output = is_coroutine_callable(SubClass().method)
    # Outputs were verified to be equal to the original implementation

🔘 (none found) − ⏪ Replay Tests