Возвращает оболочку в зависимости от того, был ли создан экземпляр async или sync - PullRequest
Новая
       10

Возвращает оболочку в зависимости от того, был ли создан экземпляр async или sync

0 голосов
/ 01 ноября 2018

Есть ли способ, которым я могу извлечь, какой подкласс был создан для данного метода из базового класса?

Я знаю, что вопрос немного сложен, поэтому вот пример: 

from functools import wraps

def my_wrapper(fn_to_wrap):

    @wraps(fn_to_wrap)
    async def async_wrapper(*args, **kwargs):
        await do_some_async_stuff()
        print('Did some async stuff')
        return fn_to_wrap(*args, **kwargs)

    @wraps(fn_to_wrap)
    def sync_wrapper(*args, **kwargs):
        do some_sync_stuff()
        print('Did some sync stuff')
        return fn_to_wrap(*args, **kwargs)

    # <my_problem>
    if fn_to_wrap belongs_to(SyncClass):
        return sync_wrapper
    else:
        return async_wrapper
    # </my_problem>


class BaseClass:
    @my_wrapper
    def fn_to_wrap(self):
        return 'Finally a return a value'

class SyncClass(BaseClass):
    def fn_to_call(self):
        return self.fn_to_wrap()


class AsyncClass(BaseClass):
    async def fn_to_call(self):
        return await self.fn_to_wrap()

Проблема в том, что метод fn_to_wrap принадлежит BaseClass. От чего наследуются мои классы Sync и Async.

Есть ли способ узнать, принадлежит ли fn_to_wrap экземпляру AsyncClass или SyncClass?

Проще, я хочу, чтобы моя консоль печатала: 

>>> my_sync_class = SyncClass()
>>> print(my_sync_class.fn_to_call())

Done some sync stuff
Finally a return value

и 

>>> my_async_class = AsyncClass()
# not in a coroutine for brevity
>>> print(await my_async_class.fn_to_call())

Done some async stuff
FInally a return value

Итак, как бы вы внедрили </my_problem> для достижения этих результатов?

[EDIT] * * тысячу двадцать-один

Мне известно о существовании inspect.iscoroutinefunction и inspect.iscoroutine. Но это не поможет, потому что упакованный метод всегда синхронен, а оболочка - это то, что выполняет асинхронные задачи.

1 Ответ

0 голосов
/ 01 ноября 2018

Если my_wrapper разрешено знать о AsyncClass и SyncClass (или вы управляете ими и можете добавить атрибут класса, например _is_sync, который сообщает оболочке, с каким классом он имеет дело), ​​вы можете просто осмотреть self.

Это невозможно сделать из местоположения <my_problem>, поскольку self там еще не доступно; код должен возвращать одну оболочку для случаев синхронизации и асинхронности. После вызова оболочка должна обнаружить асинхронный случай и вернуть экземпляр async def, если вам нужно асинхронное поведение. (Функция синхронизации, которая возвращает объект сопрограммы, функционально эквивалентна функции сопрограммы, так же, как обычная функция, оканчивающаяся на return some_generator(), идеально подходит для использования в качестве генератора.)

Вот пример, который использует isinstance для определения, какой вариант вызывается: 

def my_wrapper(fn_to_wrap):
    async def async_wrapper(*args, **kwargs):
        await asyncio.sleep(.1)
        print('Did some async stuff')
        return fn_to_wrap(*args, **kwargs)

    @wraps(fn_to_wrap)
    def uni_wrapper(self, *args, **kwargs):
        # or if self._is_async, etc.
        if isinstance(self, AsyncClass):
            return async_wrapper(self, *args, **kwargs)
        time.sleep(.1)
        print('Did some sync stuff')
        return fn_to_wrap(self, *args, **kwargs)

    return uni_wrapper

Эта реализация приводит к желаемому результату: 

>>> x = SyncClass()
>>> x.fn_to_call()
Did some sync stuff
'Finally a return a value'
>>> async def test():
...     x = AsyncClass()
...     return await x.fn_to_call()
... 
>>> asyncio.get_event_loop().run_until_complete(test())
Did some async stuff
'Finally a return a value'

Приведенное выше решение не будет работать, если оболочка не сможет различить SyncClass и AsyncClass. Есть два ограничения, которые могут помешать этому:

  • isinstance не будет работать, если число подклассов не ограничено;
  • пользовательский атрибут класса не будет работать, если окончательные классы не контролируются автором оболочки.

В этом случае оставшийся вариант - прибегнуть к черной магии, чтобы определить, вызывается ли функция из сопрограммы или из функции синхронизации. Черная магия удобно представлена ​​Дэвидом Бизли в этом разговоре : 

def from_coroutine():
    return sys._getframe(2).f_code.co_flags & 0x380

Используя from_coroutine, uni_wrapper часть my_wrapper будет выглядеть так: 

    @wraps(fn_to_wrap)
    def uni_wrapper(*args, **kwargs):
        if from_coroutine():
            return async_wrapper(*args, **kwargs)
        time.sleep(.1)
        print('Did some sync stuff')
        return fn_to_wrap(*args, **kwargs)

... с тем же результатом.

Конечно, вы должны знать, что черная магия может перестать работать в следующем выпуске Python без какого-либо предупреждения. Но если вы знаете, что делаете, это может оказаться весьма полезным.

...