programing

__init__에서 wait와 함께 클래스 속성을 설정하는 방법

topblog 2023. 5. 18. 20:38
반응형

__init__에서 wait와 함께 클래스 속성을 설정하는 방법

클래스를 정의하려면 어떻게 해야 합니까?await생성자 또는 클래스 본문에서?

예를 들어, 내가 원하는 것:

import asyncio

# some code


class Foo(object):

    async def __init__(self, settings):
        self.settings = settings
        self.pool = await create_pool(dsn)

foo = Foo(settings)
# it raises:
# TypeError: __init__() should return None, not 'coroutine'

또는 클래스 본문 속성이 있는 예제:

class Foo(object):

    self.pool = await create_pool(dsn)  # Sure it raises syntax Error

    def __init__(self, settings):
        self.settings = settings

foo = Foo(settings)

나의 해결책 (그러나 나는 좀 더 우아한 방법을 보고 싶습니다)

class Foo(object):

    def __init__(self, settings):
        self.settings = settings

    async def init(self):
        self.pool = await create_pool(dsn)

foo = Foo(settings)
await foo.init()

대부분의 마술적인 방법은async def/await일반적으로, 당신은 오직 사용해야 합니다.await전용 비동기 마법 방법 내부 -__aiter__,__anext__,__aenter__,그리고.__aexit__다른 마술 방법에 사용하는 것은 전혀 효과가 없을 것입니다.__init__(여기에 설명된 다른 답변에 설명된 트릭을 사용하지 않는 한) 또는 비동기 컨텍스트에서 마법 메서드 호출을 트리거하는 모든 항목을 항상 사용하도록 강제합니다.

존재하는asyncio도서관은 이 문제를 다음 두 가지 방법 중 하나로 처리하는 경향이 있습니다.먼저,asyncio-redis 사용된 공장 패턴(, 예를 들어)을 보았습니다.

import asyncio

dsn = "..."

class Foo(object):
    @classmethod
    async def create(cls, settings):
        self = Foo()
        self.settings = settings
        self.pool = await create_pool(dsn)
        return self

async def main(settings):
    settings = "..."
    foo = await Foo.create(settings)

다른 라이브러리들은 팩토리 메서드가 아닌 객체를 만드는 최상위 코루틴 함수를 사용합니다.

import asyncio

dsn = "..."

async def create_foo(settings):
    foo = Foo(settings)
    await foo._init()
    return foo

class Foo(object):
    def __init__(self, settings):
        self.settings = settings

    async def _init(self):
        self.pool = await create_pool(dsn)

async def main():
    settings = "..."
    foo = await create_foo(settings)

그create_pool에서 기능.aiopg당신이 전화하고 싶은 것.__init__실제로는 이 정확한 패턴을 사용하고 있습니다.

이것은 적어도 그것을 해결합니다.__init__이슈. 제가 기억할 수 있는 야생에서 비동기 호출을 하는 클래스 변수를 본 적이 없어서 잘 정립된 패턴이 나타났는지 모르겠습니다.

재미를 위한 또 다른 방법:

class aobject(object):
    """Inheriting this class allows you to define an async __init__.

    So you can create objects by doing something like `await MyClass(params)`
    """
    async def __new__(cls, *a, **kw):
        instance = super().__new__(cls)
        await instance.__init__(*a, **kw)
        return instance

    async def __init__(self):
        pass

#With non async super classes

class A:
    def __init__(self):
        self.a = 1

class B(A):
    def __init__(self):
        self.b = 2
        super().__init__()

class C(B, aobject):
    async def __init__(self):
        super().__init__()
        self.c=3

#With async super classes

class D(aobject):
    async def __init__(self, a):
        self.a = a

class E(D):
    async def __init__(self):
        self.b = 2
        await super().__init__(1)

# Overriding __new__

class F(aobject):
    async def __new__(cls):
        print(cls)
        return await super().__new__(cls)

    async def __init__(self):
        await asyncio.sleep(1)
        self.f = 6

async def main():
    e = await E()
    print(e.b) # 2
    print(e.a) # 1

    c = await C()
    print(c.a) # 1
    print(c.b) # 2
    print(c.c) # 3

    f = await F() # Prints F class
    print(f.f) # 6

import asyncio
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

이와 같은 작업을 수행하는 것이 좋습니다. 매우 쉽습니다.

import asyncio

class Foo:
    def __init__(self, settings):
        self.settings = settings

    async def async_init(self):
        await create_pool(dsn)

    def __await__(self):
        return self.async_init().__await__()

loop = asyncio.get_event_loop()
foo = loop.run_until_complete(Foo(settings))

기본적으로 여기서 일어나는 일은__init__()일반적으로 먼저 호출됩니다.그리고나서__await__()호출되어 대기 중입니다.async_init().

저는 별도의 공장 방식을 추천합니다.안전하고 간단합니다.하지만, 만약 당신이 고집한다면,async의 버전__init__()예를 들어 보겠습니다.

def asyncinit(cls):
    __new__ = cls.__new__

    async def init(obj, *arg, **kwarg):
        await obj.__init__(*arg, **kwarg)
        return obj

    def new(cls, *arg, **kwarg):
        obj = __new__(cls, *arg, **kwarg)
        coro = init(obj, *arg, **kwarg)
        #coro.__init__ = lambda *_1, **_2: None
        return coro

    cls.__new__ = new
    return cls

용도:

@asyncinit
class Foo(object):
    def __new__(cls):
        '''Do nothing. Just for test purpose.'''
        print(cls)
        return super().__new__(cls)

    async def __init__(self):
        self.initialized = True


async def f():
    print((await Foo()).initialized)

loop = asyncio.get_event_loop()
loop.run_until_complete(f())

출력:

<class '__main__.Foo'>
True

설명:

클래스 구성은 다음을 반환해야 합니다.coroutine해당 인스턴스 대신 개체를 지정합니다.

AsyncObj 클래스는__ainit__"skc-proxer":

class AsyncObj:
    def __init__(self, *args, **kwargs):
        """
        Standard constructor used for arguments pass
        Do not override. Use __ainit__ instead
        """
        self.__storedargs = args, kwargs
        self.async_initialized = False

    async def __ainit__(self, *args, **kwargs):
        """ Async constructor, you should implement this """

    async def __initobj(self):
        """ Crutch used for __await__ after spawning """
        assert not self.async_initialized
        self.async_initialized = True
        await self.__ainit__(*self.__storedargs[0], **self.__storedargs[1])  # pass the parameters to __ainit__ that passed to __init__
        return self

    def __await__(self):
        return self.__initobj().__await__()

    def __init_subclass__(cls, **kwargs):
        assert asyncio.iscoroutinefunction(cls.__ainit__)  # __ainit__ must be async

    @property
    def async_state(self):
        if not self.async_initialized:
            return "[initialization pending]"
        return "[initialization done and successful]"

다음은 "비동기화 클래스"의 예입니다.

class MyAsyncObject(AsyncObj):
    async def __ainit__(self, param1, param2=0):
        print("hello!", param1, param2)
        # go something async, e.g. go to db

용도:

async def example():
    my_obj = await MyAsyncObject("test", 123)

[거의] @ojii의 표준적인 대답.

@dataclass
class Foo:
    settings: Settings
    pool: Pool

    @classmethod
    async def create(cls, settings: Settings, dsn):
        return cls(settings, await create_pool(dsn))

나는 코루틴 기반 방법을 훨씬 더 쉽게 시작하는 방법을 보여주고 싶습니다.__init__방법.

import asyncio


class Foo(object):

    def __init__(self, settings):
        self.settings = settings
        loop = asyncio.get_event_loop() 
        self.pool = loop.run_until_complete(create_pool(dsn))

foo = Foo(settings)

주의해야 할 중요한 점은 다음과 같습니다.

  • 그러면 비동기 코드가 동기(차단)로 작동합니다.
  • 이것은 비동기 코드를 실행하는 가장 좋은 방법은 아니지만 동기화 방법을 통한 초기화에만 해당하는 경우에는 다음과 같습니다.__init__그것은 잘 맞을 것 같아요.
  • 시작 후 wait와 함께 개체에서 비동기 메서드를 실행할 수 있습니다.예await foo.pool.get(value)
  • 를통시지마오시십하작해▁▁via를 통해 시작하려고 하지 .await받을 전화RuntimeError: This event loop is already running

비슈누 셰티가르의 대답은 지금까지 가장 간단하지만, 그의 대답은async_init메소드는 객체 자체를 반환하지 않으므로foo이 Foo사례하는 가장 은 OP의목적에대말면자하서해클, IMHO를는한방은법아입니다.

import asyncio

class Foo:
    def __init__(self, settings):
        self.settings = settings

    def __await__(self):
        self.pool = asyncio.create_task(create_pool(dsn))
        yield from self.pool
        self.pool = self.pool.result()
        return self

개체를 초기화하려면 다음을 수행합니다.

def main():
    loop = asyncio.get_event_loop()
    foo = loop.run_until_complete(Foo(settings))

또는

async def main():
    foo = await Foo(settings)

우리는 비동기 코드를 수동으로 실행하여 비동기 호출을 동기 호출로 변환할 수 있습니다.asyncio.run()

class Foo:
    async def __ainit__(self, param):
        self._member = await some_async_func(param)

    def __init__(self, param):
        asyncio.run(self.__ainit__(param))

필에따다사수있도습다니용할을음라요있다를 사용할 수도 .AwaitLoader출처: https://pypi.org/project/async-property/

문서에서:

AwaitLoader전화 기다리겠습니다.instance.load()속성을 로드하기 전에 필요합니다.

모든 사용자가 시도할 수 있습니다. https://pypi.org/project/asyncinit/

  • pip 설치 비동기
from asyncinit import asyncinit

@asyncinit
class MyClass:
    async def __init__(self, param):
        self.val = await self.deferredFn(param)

    async def deferredFn(self, x):
        # ...
        return x + 2

obj = await MyClass(42)
assert obj.val == 44

나는 이 믹스를 다음에 썼습니다:

import asyncio


class AsyncMixin:
    """Adds an async_init method to an object which is called when the
    object is awaited.
    Typically the idiom obj = await Object()
    provides the synchronous __init__() and async async_init() calls"""

    async def async_init(self):
        """If an AsyncMixin object is created in an async context (ie await
        Object() then the __init__ method is non-async as normal but
        the async_init() method is called immediately by the
        __await__() magic method.
        """
        pass

    async def _async_init(self):
        task = asyncio.create_task(self.async_init())
        await task
        return self

    def __await__(self):
        return self._async_init().__await__()
    pass

따라서 OP의 솔루션은 다음과 같습니다.

class Foo(object, AsyncMixin):

    def __init__(self, settings):
        self.settings = settings

    async def async_init(self):
        self.pool = await create_pool(dsn)

foo = await Foo(settings)

제가 생각하기에 꽤 우아합니다.

이것은 Python 3.9에서 작동했습니다.


from aiobotocore.session import AioSession
import asyncio




class SomeClass():

    def __init__(self):
        asyncio.run(self.async_init())
        print(self.s3)

    async def async_init(self):
        self.s3 = await AioSession().create_client('s3').__aenter__()

언급URL : https://stackoverflow.com/questions/33128325/how-to-set-class-attribute-with-await-in-init

반응형

'programing' 카테고리의 다른 글

Windows 명령줄에서 Git Bash를 시작하는 방법  (0) 2023.05.18
module.exports 내의 "local" 함수를 module.exports 내의 다른 함수에서 호출하시겠습니까?  (0) 2023.05.18
목록이 얼마나 정렬되어 있는지 측정할 수 있는 방법이 있습니까?  (0) 2023.05.18
UI 레이블 텍스트 여백  (0) 2023.05.18
배열에서 빈 요소를 제거하려면 어떻게 해야 합니까?  (0) 2023.05.18

'programing'의 다른글

  • 현재글__init__에서 wait와 함께 클래스 속성을 설정하는 방법

관련글

댓글

티스토리툴바