Probar si la función o el método es normal o asíncrono

¿Cómo puedo saber si una función o un método es una función normal o una función asíncrona? Me gustaría que mi código soporte automáticamente las devoluciones de llamadas normales o asíncronas y necesito una forma de probar qué tipo de función se pasa.

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) 

Y dependiendo del tipo de función que se pase, se debe ejecutar normalmente o con espera. isAsync() varias cosas, pero no tengo idea de cómo implementar isAsync() .

Utilice el módulo de inspección de Python.

inspect.iscoroutinefunction(object)

Devuelve verdadero si el objeto es una función de rutina (una función definida con una syntax de definición asíncrona).

Esta función está disponible desde Python 3.5. El módulo está disponible para Python 2 con menos funcionalidades y ciertamente sin el que está buscando: inspeccionar

Inspeccionar el módulo como su nombre indica es útil para inspeccionar un montón de cosas. La documentación dice

El módulo de inspección proporciona varias funciones útiles para ayudar a obtener información sobre objetos en vivo, tales como módulos, clases, métodos, funciones, seguimientos, objetos de marco y objetos de código. Por ejemplo, puede ayudarlo a examinar el contenido de una clase, recuperar el código fuente de un método, extraer y formatear la lista de argumentos para una función u obtener toda la información que necesita para mostrar un seguimiento detallado.

Este módulo proporciona cuatro tipos principales de servicios: verificación de tipos, obtención de código fuente, inspección de clases y funciones y examen de la stack de intérpretes.

Algunas capacidades básicas de este módulo son:

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

También tiene capacidad para recuperar el código fuente.

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

Los métodos se nombran intuitivamente. Descripción si es necesario se puede encontrar en la documentación.

Si no desea introducir otra importación con inspect , iscoroutine también está disponible dentro de asyncio .

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

Las co-rutinas tienen el COROUTINE indicadores COROUTINE , bit 6 en los indicadores de código:

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

El valor 128 se almacena como una constante en el módulo de inspect :

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

La función inspect.iscoroutinefunction() hace justamente eso; pruebe si el objeto es una función o un método (para asegurarse de que hay un atributo __code__ ) y pruebe ese indicador. Consulte el código fuente .

Por supuesto, el uso de inspect.iscoroutinefunction() es el más legible y está garantizado para continuar trabajando si alguna vez inspect.iscoroutinefunction() código:

 >>> inspect.iscoroutinefunction(foo) True 

Las soluciones anteriores funcionarán para casos simples, cuando pase la función de coroutine. En algunos casos, es posible que desee pasar una función de objeto esperada que actúa como una función de coroutine, pero no es una función de coroutine. Dos ejemplos son la clase Future o la clase de objeto similar a Future (clase que implementa el método mágico __await__ ). En este caso, iscoroutinefunction devolverá False , lo que no es necesario.

Es más fácil de entender en el ejemplo no asíncrono al pasar la función no llamable como callback:

 class SmartCallback: def __init__(self): print('SmartCallback is not function, but can be used as function') await callCallback(SmartCallback) # Should work, right? 

De vuelta al mundo asíncrono, una situación similar:

 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 

Una forma de resolverlo es no usar la función iscoroutine o iscoroutinefunction , sino usar inspect.isawaitable en inspect.isawaitable lugar. Funciona con un objeto listo, por lo que debe crearlo primero. En otras palabras, la solución que aconsejaría utilizar:

 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') 

Es una solución más universal (y estoy seguro de que es lógicamente correcta).