Skip to content

Commit

Permalink
Add is_awaitable utility function
Browse files Browse the repository at this point in the history
This can act as a faster replacement for inspect.isawaitable.
  • Loading branch information
Cito committed Mar 21, 2020
1 parent 85635a2 commit 9feec00
Show file tree
Hide file tree
Showing 3 changed files with 129 additions and 0 deletions.
2 changes: 2 additions & 0 deletions src/graphql/pyutils/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
from .event_emitter import EventEmitter, EventEmitterAsyncIterator
from .identity_func import identity_func
from .inspect import inspect
from .is_awaitable import is_awaitable
from .is_collection import is_collection
from .is_finite import is_finite
from .is_integer import is_integer
Expand All @@ -47,6 +48,7 @@
"EventEmitterAsyncIterator",
"identity_func",
"inspect",
"is_awaitable",
"is_collection",
"is_finite",
"is_integer",
Expand Down
24 changes: 24 additions & 0 deletions src/graphql/pyutils/is_awaitable.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
import inspect
from typing import Any
from types import CoroutineType, GeneratorType

__all__ = ["is_awaitable"]

CO_ITERABLE_COROUTINE = inspect.CO_ITERABLE_COROUTINE


def is_awaitable(value: Any) -> bool:
"""Return true if object can be passed to an ``await`` expression.
Instead of testing if the object is an instance of abc.Awaitable, it checks
the existence of an `__await__` attribute. This is much faster.
"""
return (
# check for coroutine objects
isinstance(value, CoroutineType)
# check for old-style generator based coroutine objects
or isinstance(value, GeneratorType)
and bool(value.gi_code.co_flags & CO_ITERABLE_COROUTINE)
# check for other awaitables (e.g. futures)
or hasattr(value, "__await__")
)
103 changes: 103 additions & 0 deletions tests/pyutils/test_is_awaitable.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
import asyncio
from inspect import isawaitable

from pytest import mark # type: ignore

from graphql.pyutils import is_awaitable


def describe_is_awaitable():
def declines_the_none_value():
assert not isawaitable(None)
assert not is_awaitable(None)

def declines_a_boolean_value():
assert not isawaitable(True)
assert not is_awaitable(True)

def declines_an_int_value():
assert not is_awaitable(42)

def declines_a_string_value():
assert not isawaitable("some_string")
assert not is_awaitable("some_string")

def declines_a_dict_value():
assert not isawaitable({})
assert not is_awaitable({})

def declines_an_object_instance():
assert not isawaitable(object())
assert not is_awaitable(object())

def declines_the_type_class():
assert not isawaitable(type)
assert not is_awaitable(type)

def declines_a_lambda_function():
assert not isawaitable(lambda: True) # pragma: no cover
assert not is_awaitable(lambda: True) # pragma: no cover

def declines_a_normal_function():
def some_function():
return True

assert not isawaitable(some_function())
assert not is_awaitable(some_function)

def declines_a_normal_generator_function():
def some_generator():
yield True # pragma: no cover

assert not isawaitable(some_generator)
assert not is_awaitable(some_generator)

def declines_a_normal_generator_object():
def some_generator():
yield True # pragma: no cover

assert not isawaitable(some_generator())
assert not is_awaitable(some_generator())

def declines_a_coroutine_function():
async def some_coroutine():
return True # pragma: no cover

assert not isawaitable(some_coroutine)
assert not is_awaitable(some_coroutine)

@mark.filterwarnings("ignore::RuntimeWarning")
def recognizes_a_coroutine_object():
async def some_coroutine():
return False # pragma: no cover

assert isawaitable(some_coroutine())
assert is_awaitable(some_coroutine())

@mark.filterwarnings("ignore::RuntimeWarning")
@mark.filterwarnings("ignore::DeprecationWarning")
def recognizes_an_old_style_coroutine():
@asyncio.coroutine
def some_old_style_coroutine():
yield False # pragma: no cover

assert is_awaitable(some_old_style_coroutine())
assert is_awaitable(some_old_style_coroutine())

@mark.filterwarnings("ignore::RuntimeWarning")
def recognizes_a_future_object():
async def some_coroutine():
return False # pragma: no cover

some_future = asyncio.ensure_future(some_coroutine())

assert is_awaitable(some_future)
assert is_awaitable(some_future)

@mark.filterwarnings("ignore::RuntimeWarning")
def declines_an_async_generator():
async def some_async_generator():
yield True # pragma: no cover

assert not isawaitable(some_async_generator())
assert not is_awaitable(some_async_generator())

0 comments on commit 9feec00

Please sign in to comment.