Sprawdź, czy funkcja lub metoda jest normalna lub asynchroniczna.

Question

Jak mogę się dowiedzieć, czy funkcja lub metoda jest normalną funkcją lub asynchroniczną? Chciałbym, aby mój kod automatycznie obsługiwał normalne lub asynchroniczne wywołania zwrotne i potrzebował sposobu na sprawdzenie, jaki typ funkcji został przekazany.

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) 

A w zależności od tego, jaki typ funkcji zostanie przekazany, należy uruchomić go normalnie lub z wyczekiwaniem. Próbowałem różnych rzeczy, ale nie mam pojęcia, jak wdrożyć isAsync().

Answer 1

Użyj modułu Python .

inspect.iscoroutinefunction(object)

powrotu ważne, jeśli celem jest funkcja współprogram (funkcja zdefiniowana składni def asynchroniczny).

Ta funkcja jest dostępna od wersji Python 3.5. Moduł jest dostępny dla Pythona 2 z mniejszymi funkcjami iz pewnością bez tego, którego szukasz: inspect

Moduł inspekcji, jak sugeruje nazwa, jest przydatny do inspekcji wielu rzeczy. Dokumentacja mówi

sprawdzić moduł zawiera kilka użytecznych funkcji, aby pomóc uzyskać informacje o żywych obiektów, takich jak moduły, klasy, metody, funkcje, tracebacks, obiektów kadru i obiektów kodu. Na przykład może pomóc w zbadaniu zawartości klasy, pobrać kod źródłowy metody, wyodrębnić i sformatować listę argumentów dla funkcji lub uzyskać wszystkie informacje potrzebne do wyświetlenia szczegółowych informacji zwrotnych.

Moduł ten oferuje cztery główne rodzaje usług: sprawdzanie typu, uzyskiwanie kodu źródłowego, kontrola klas i funkcji oraz sprawdzanie stosu tłumacza.

Niektóre podstawowe funkcje tego modułu to:

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

pakuje również możliwość pobierania kodu źródłowego

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

metody nazywane są intuicyjnie. Opis w razie potrzeby można znaleźć w dokumentacji.

Answer 2

Co-procedury ma zestaw COROUTINE flagi bit 6 w flagi KOD:

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

Wartość 128 jest przechowywany w postaci stałej w module inspect:

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

inspect.iscoroutinefunction() function czyni tylko to; sprawdź, czy obiekt jest funkcją lub metodą (aby upewnić się, że istnieje atrybut __code__) i przetestuj dla tej flagi. Zobacz source code.

Oczywiście, używając inspect.iscoroutinefunction() jest najbardziej czytelny i gwarantowane do dalszej pracy, jeśli w ogóle flagi kodowe były zmiany:

>>> inspect.iscoroutinefunction(foo) 
True 

Answer 3

Jeśli nie chcą wprowadzić kolejny import z inspect, iscoroutine jest dostępny również w wersji asyncio.

import asyncio 

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

Answer 4

Rozwiązania powyżej będą działać w prostych przypadkach, gdy przejdziesz funkcję coroutine. W niektórych przypadkach możesz przesłać niezaprzeczalną funkcję obiektu, która działa jak funkcja coroutine, ale nie jest funkcją coroutine. Dwa przykłady to klasa Future lub Przyszłościowa klasa obiektu (klasa, która jest metodą magiczną). W takich przypadkach iscoroutinefunction zwróci False, czego nie potrzebujesz.

Łatwiej zrozumieć na non-asynchronicznym przykład przy przejściu zakaz funkcję wpłacone jako callback:

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

await callCallback(SmartCallback) # Should work, right? 

Powrót do asynchronicznej świecie, podobna sytuacja:

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 

sposobem rozwiązania go nie używać iscoroutine lub iscoroutinefunction, ale zamiast tego użyj inspect.isawaitable. Działa z gotowym obiektem, więc musisz go najpierw utworzyć. Innymi słowy, rozwiązanie, które powinienem zastosować:

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

To bardziej uniwersalne (i jestem pewien, logicznie poprawne) rozwiązanie.