Проверьте, нормально или асинхронно, если функция или метод

Ответ 1

Используйте inspect модуль Python.

inspect.iscoroutinefunction(object)

Возвращает true, если объект является сопроводительной функцией (функция, определенная с синтаксисом aync def).

Эта функция доступна с Python 3.5. Модуль доступен для Python 2 с меньшей функциональностью и, конечно же, без того, который вы ищете: inspect

Проверьте, что модуль, как следует из названия, полезен для проверки множества вещей. В документации указано

Модуль проверки предоставляет несколько полезных функций, помогающих получать информацию о живых объектах, таких как модули, классы, методы, функции, трассировки, объекты фрейма и объекты кода. Например, он может помочь вам изучить содержимое класса, получить исходный код метода, извлечь и отформатировать список аргументов для функции или получить всю необходимую информацию для отображения подробной трассировки.

Существует четыре основных вида услуг, предоставляемых этим модулем: проверка типов, получение исходного кода, проверка классов и функций и проверка стека интерпретаторов.

Некоторые основные возможности этого модуля:

inspect.ismodule(object)
inspect.isclass(object)
inspect.ismethod(object)
inspect.isfunction(object)

Он также имеет возможность извлекать исходный код

inspect.getdoc(object)
inspect.getcomments(object)
inspect.getfile(object) 
inspect.getmodule(object)

Способы называются интуитивно. Описание при необходимости можно найти в документации.

Ответ 2

Если вы не хотите вводить другой импорт с inspect, iscoroutine также доступен внутри asyncio.

import asyncio

def isAsync(someFunc):
    return asyncio.iscoroutinefunction(someFunc)

Ответ 3

Co-подпрограммы имеют флаг COROUTINE, бит 6 в флагах кода:

>>> async def foo(): pass
>>> foo.__code__.co_flags & (2 << 6)
128   # not 0, so the flag is set.

Значение 128 сохраняется как константа в модуле inspect:

>>> import inspect
>>> inspect.CO_COROUTINE
128
>>> foo.__code__.co_flags & inspect.CO_COROUTINE
128

Функция inspect.iscoroutinefunction() делает именно это; проверьте, является ли объект функцией или методом (для обеспечения атрибута __code__) и проверки для этого флага. Смотрите исходный код .

Конечно, использование inspect.iscoroutinefunction() является наиболее читаемым и гарантировано продолжит работу, если бы когда-либо изменялись флагов кода:

>>> inspect.iscoroutinefunction(foo)
True

Ответ 4

TL;DR

Если вы хотите проверить, что-то должно использоваться с await, используйте inspect.isawaitable (например, когда вы тестируете что-то callable() не просто как функцию).

В отличие от функции iscoroutine или iscoroutinefunction она также будет работать для Future и объектов, которые реализуют метод __await__.

Решения выше будут работать для простых случаев, когда вы передаете функцию сопрограммы. В некоторых случаях вам может потребоваться передать ожидаемую объектную функцию, которая действует как функция сопрограммы, но не является функцией сопрограммы. Два примера - класс Future или класс объектов, подобный Future (класс, реализующий магический метод __await__). В этом случае iscoroutinefunction вернет False, то, что вам не нужно.

Это легче понять на не асинхронном примере с передачей функции без вызова в качестве обратного вызова:

class SmartCallback:
    def __init__(self):
        print('SmartCallback is not function, but can be used as function')

await callCallback(SmartCallback)  # Should work, right?

Вернемся к асинхронному миру, похожая ситуация:

class AsyncSmartCallback:
    def __await__(self):
        return self._coro().__await__()

    async def _coro(self):
        print('AsyncSmartCallback is not coroutine function, but can be used as coroutine function')
        await asyncio.sleep(1)

await callCallback(AsyncSmartCallback)  # Should work, but oops! iscoroutinefunction(AsyncSmartCallback) == False

Чтобы решить эту проблему, не используйте iscoroutine или iscoroutinefunction, а вместо этого используйте inspect.isawaitable. Он работает с готовым объектом, поэтому сначала вы должны его создать. Другими словами, решение я бы посоветовал использовать:

async def callCallback(cb, arg):
    if callable(cb):
        res = cb()  # here result of regular func or awaitable
        if inspect.isawaitable(res):
            res = await res  # await if awaitable
        return res  # return final result
    else:
        raise ValueError('cb is not callable')

Это более универсальное (и я уверен, логически правильное) решение.