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

Question

Как узнать, является ли функция или метод нормальной или асинхронной? Я хотел бы, чтобы мой код автоматически поддерживал обычные или асинхронные обратные вызовы и нуждался в способе проверить, какой тип функции передан.

async def exampleAsyncCb(): 
    pass 

def exampleNomralCb(): 
    pass 

def isAsync(someFunc): 
    #do cool dynamic python stuff on the function 
    return True/False 

async def callCallback(cb, arg): 
    if isAsync(cb): 
     await cb(arg) 
    else: 
     cb(arg) 

И в зависимости от того, какой тип функции передается, он должен запускать его нормально или с ожиданием. Я пробовал разные вещи, но понятия не имею, как реализовать isAsync().

Answer 1

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

inspect.iscoroutinefunction(object)

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

Эта функция доступна с 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) 

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

Answer 2

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() function делает только то; проверьте, является ли объект функцией или методом (для обеспечения наличия атрибута __code__) и проверьте этот флаг. См. source code.

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

>>> inspect.iscoroutinefunction(foo) 
True 

Answer 3

Если вы не хотите, чтобы ввести еще один импорт с inspect, iscoroutine является также доступно внутри asyncio.

import asyncio 

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

Answer 4

Решения, указанные выше, будут работать в простых случаях, когда вы передадите функцию сопрограммы. В некоторых случаях вам может понравиться awaitable object функция, которая действует как функция coroutine, но не является функцией coroutine. Два примера: Future класс или Будущий объект класс (класс implements__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's 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') 

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