İşlev veya yöntemin normal veya asenkronize olup olmadığını sınama

Question

Bir işlev veya yöntemin normal bir işlev mi yoksa bir uyumsuzluk işlevi mi olduğunu nasıl anlayabilirim? Kodumun normal veya async geri aramalarını otomatik olarak desteklemesini ve ne tür bir işlevin geçtiğini test etmenin bir yoluna ihtiyaç duyarım.

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) 

Ve normalde ya beklemektedir ile çalışmalıdır ya iletilir fonksiyonun türüne bağlı. Çeşitli şeyler denedim ama isAsync()'u nasıl uygulayacağımı bilmiyorum.

Answer 1

Python'un inspect modülünü kullanın.

inspect.iscoroutinefunction(object) doğru

Dönüş nesnesi, bir eşyordam fonksiyonu (zaman uyumsuz def sözdizimi ile tanımlanan bir fonksiyonu) ise. Bu işlev Python 3.5'den beri kullanılabilir. Bu işlev Python 3.5'den beri kullanılabilir. Modül, Python 2 için daha az işlevsellik ile ve tam olarak aradığınız model olmadan kullanılabilir: inspect

Modülü, adından da anlaşılacağı gibi, bir çok şeyi incelemek yararlıdır. Dokümantasyonda, modüller, sınıflar, metotlar, fonksiyonlar, geridönüşümler, çerçeve nesneleri ve kod nesneleri gibi canlı nesneler hakkında bilgi edinmeye yardımcı olacak birçok yararlı fonksiyon bulunmaktadır. Örneğin, bir sınıfın içeriğini incelemenize, bir yöntemin kaynak kodunu almanıza, bir işlev için argüman listesini ayıklamanıza ve biçimlendirmenize ya da ayrıntılı bir geri izleme görüntülemek için gereken tüm bilgileri almanıza yardımcı olabilir.

Bu modül tarafından sağlanan dört temel hizmet türü vardır: tip denetimi, kaynak kodu alma, sınıfları ve işlevleri inceleme ve yorumlayıcı kümesini inceleme.

Bu modülün bazı temel yetenekleri şunlardır:

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

Ayrıca

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

Yöntemleri sezgisel olarak adlandırılır kaynak kodunu almak için yeteneği paketleri. Gerektiğinde açıklama dokümantasyonda bulunabilir.

Answer 2

Co-rutinleri COROUTINE bayrağı ayarlanmışsa, kod bayrakları bit 6:

değeri 128 inspect modülünde bir sabit olarak depolanır

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

:

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

inspect.iscoroutinefunction() function yapar sadece bu; Nesnenin bir işlev veya yöntem olup olmadığını sınayın (__code__ özniteliğinin bulunduğundan emin olmak için) ve bu işaret için sınayın. source code'a bakın.inspect.iscoroutinefunction() kullanarak Tabii

, hiç kod bayrakları değiştirmek olsaydı çalışmaya devam etmek en okunabilir ve garantili geçerli:

>>> inspect.iscoroutinefunction(foo) 
True 

Answer 3

Eğer inspect başka ithalat tanıtmak istemiyorsanız, iscoroutine olduğunu asyncio da mevcuttur.

import asyncio 

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

Answer 4

Yukarıdaki çözümler, coroutine işlevini ilettiğinizde basit durumlar için çalışacaktır. Bazı durumlarda, coroutine işlevi gibi davranan algılanan nesne işlevini geçmek isteyebilirsiniz, ancak coroutine işlevi değildir. İki örnek Future sınıfı veya Gelecek benzeri nesne sınıfıdır (sınıf implements__await__ sihirli yöntemidir). Bu durumda iscoroutinefunction, False'u döndürecek, ihtiyacınız olmayan şey.

Bu çağrılabilir geri arama olmayan işlevi geçirerek ile uyumsuz olmayan örnek üzerinde anlaşılması kolay:

Geri zaman uyumsuz dünyaya

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

await callCallback(SmartCallback) # Should work, right? 

benzer bir durum: öyle değil çözmek için

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 

Way iscoroutine veya iscoroutinefunction kullanmak için, ancak bunun yerine inspect.isawaitable kullanın. Hazır nesne ile çalışır, bu yüzden önce onu oluşturmanız gerekir. Bu daha evrensel olduğunu

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

(ve ben mantıken doğru eminim) çözümü: Diğer bir deyişle, ben tavsiye ediyorum çözüm kullanmak.