Test of functie of methode normaal of asynchroon is

Hoe kom ik erachter of een functie of methode een normale functie of een asynchrone functie is? Ik wil dat mijn code automatisch normale of asynchrone callbacks ondersteunt en ik heb een manier nodig om te testen welk type functie wordt doorgegeven.

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)

En afhankelijk van het type functie dat wordt doorgegeven, moet het normaal of met wachten worden uitgevoerd. Ik heb verschillende dingen geprobeerd, maar heb geen idee hoe ik isAsync()moet implementeren.

Antwoord 1, autoriteit 100%

Gebruik de inspect-module van Python.

inspect.iscoroutinefunction(object)

Retourneer true als het object een coroutine-functie is (een functie die is gedefinieerd met een async def-syntaxis).

Deze functie is beschikbaar sinds Python 3.5.
De module is beschikbaar voor Python 2 met minder functionaliteiten en zeker zonder degene die je zoekt: inspect

Inspecteer module, zoals de naam al doet vermoeden, is handig om een ​​heleboel dingen te inspecteren. De documentatie zegt

De inspect-module biedt verschillende handige functies om informatie te krijgen over live-objecten, zoals modules, klassen, methoden, functies, tracebacks, frame-objecten en code-objecten. Het kan u bijvoorbeeld helpen de inhoud van een klasse te onderzoeken, de broncode van een methode op te halen, de lijst met argumenten voor een functie te extraheren en op te maken, of alle informatie te krijgen die u nodig hebt om een ​​gedetailleerde traceback weer te geven.

Deze module biedt vier soorten services: typecontrole, broncode ophalen, klassen en functies inspecteren en de tolkstack onderzoeken.

Enkele basismogelijkheden van deze module zijn:

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

Het bevat ook de mogelijkheid om de broncode op te halen

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

Methoden worden intuïtief benoemd. Beschrijving is indien nodig te vinden in de documentatie.

Antwoord 2, autoriteit 77%

Als je geen nieuwe import wilt invoeren met inspect, is iscoroutineook beschikbaar in asyncio.

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

Antwoord 3, autoriteit 61%

TLDR

Als je wilt controleren of iets moet worden gebruikt met await, gebruik dan inspect.isawaitable(zoals wanneer je test dat iets callable()is behalve dat het een functie is).

In tegenstelling tot iscoroutineof iscoroutinefunctionzal het ook werken voor Futures en objecten die de __await__methode implementeren.

Gedetailleerd

Bovenstaande oplossingen werken voor eenvoudige gevallen, wanneer u de coroutine-functie doorgeeft. In sommige gevallen wilt u misschien de functie wachtbaar objectdoorgeven die als coroutine-functie werkt, maar geen coroutine-functie is. Twee voorbeelden zijn Futureclass of Future- like objectclass (klasse die implementeert__await__magische methode). In dit geval retourneert iscoroutinefunctionFalse, wat u niet nodig hebt.

Het is gemakkelijker te begrijpen in een niet-async-voorbeeld door niet-functionele oproepbaar als callback door te geven:

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

Terug naar asynchrone wereld, een vergelijkbare situatie:

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

Manier om het op te lossen door iscoroutineof iscoroutinefunctionniet te gebruiken, maar in plaats daarvan inspect.isawaitablete gebruiken. Het werkt met een gereed object, dus u moet het eerst maken. Met andere woorden, de oplossing die ik zou aanraden te gebruiken:

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

Het is een meer universele (en ik weet zeker dat het logisch correct is) oplossing.

Antwoord 4, autoriteit 39%

Co-routines hebben de vlag COROUTINE, bit 7 in de codevlaggen:

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

De waarde 128 wordt opgeslagen als een constante in de module inspect:

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

De inspect.iscoroutinefunction()functiedoet precies dat; test of het object een functie of methode is (om er zeker van te zijn dat er een __code__attribuut is) en test voor die vlag. Zie de broncode.

Natuurlijk is het gebruik van inspect.iscoroutinefunction()het meest leesbaar en blijft het gegarandeerd werken als de codevlaggen ooit zouden veranderen:

>>> inspect.iscoroutinefunction(foo)
True