Async class method python

Создание классов с asyncio#

В целом асинхронные классы пишутся так же, как синхронные, но есть несколько отличий:

• особенности работы __init__
• некоторые специальные методы имеют отдельные версии для асинхронных классов, например, __aenter__ , __aexit__ , __anext__ , __aiter__

Методы класса могут быть сопрограммами или обычными функциями.

Метод __init__#

В синхронном варианте, если создавать класс который подключается по SSH, то само подключение обычно будет выполняться в методе __init__ . В асинхронных классах так сделать не получится, так как для выполнения подключения в __init__ , надо сделать init сопрограммой. Так как init должен возвращать None, если сделать init сопрограммой, возникнет ошибка:

In [3]: class ConnectSSH: . : async def __init__(self): . : await asyncio.sleep(0.1) . : In [5]: r1 = ConnectSSH() --------------------------------------------------------------------------- TypeError Traceback (most recent call last) ipython-input-5-ceb9b33fd87d> in module> ----> 1 r1 = ConnectSSH() TypeError: __init__() should return None, not 'coroutine' 

Распространенный вариант решения проблемы — создание отдельного метода для подключения.

Источник

Saved searches

Use saved searches to filter your results more quickly

You signed in with another tab or window. Reload to refresh your session. You signed out in another tab or window. Reload to refresh your session. You switched accounts on another tab or window. Reload to refresh your session.

Write classes with async def __ainit__

License

mosquito/async-class

This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

Name already in use

A tag already exists with the provided branch name. Many Git commands accept both tag and branch names, so creating this branch may cause unexpected behavior. Are you sure you want to create this branch?

Sign In Required

Please sign in to use Codespaces.

Launching GitHub Desktop

If nothing happens, download GitHub Desktop and try again.

Launching GitHub Desktop

If nothing happens, download GitHub Desktop and try again.

Launching Xcode

If nothing happens, download Xcode and try again.

Launching Visual Studio Code

Your codespace will open once ready.

There was a problem preparing your codespace, please try again.

Latest commit

Git stats

Files

Failed to load latest commit information.

README.rst

Adding abillity to write classes with awaitable initialization function.

import asyncio from async_class import AsyncClass, AsyncObject, task, link class MyAsyncClass(AsyncClass): async def __ainit__(self): # Do async staff here pass class MainClass(AsyncObject): async def __ainit__(self): # Do async staff here pass async def __adel__(self): """ This method will be called when object will be closed """ pass class RelatedClass(AsyncObject): async def __ainit__(self, parent: MainClass): link(self, parent) async def main(): instance = await MyAsyncClass() print(instance) main_instance = await MainClass() related_instance = await RelatedClass(main_instance) assert not main_instance.is_closed assert not related_instance.is_closed await main_instance.close() assert main_instance.is_closed # will be closed because linked to closed main_instance assert related_instance.is_closed asyncio.run(main())

Async objects might be created when no one event loop has been running. self.loop property is lazily evaluated.

Module provides useful abstractions for writing async code.

Objects inherited from AsyncClass might have their own __init__ method, but it strictly not recommend.

Is a base wrapper with metaclass has no TaskStore instance and additional methods like self.create_task and self.create_future .

This class just solves the initialization problem:

import asyncio from async_class import AsyncClass class MyAsyncClass(AsyncClass): async def __ainit__(self): future = self.loop.create_future() self.loop.call_soon(future.set_result, True) await future async def main(): instance = await MyAsyncClass() print(instance) asyncio.run(main())

Base class with task store instance and helpers for simple task management.

import asyncio from async_class import AsyncObject class MyClass(AsyncObject): async def __ainit__(self): self.task = self.create_task(asyncio.sleep(3600)) async def main(): obj = await MyClass() assert not obj.task.done() await obj.close() assert obj.task.done() asyncio.run(main())

TaskStore is a task management helper. One instance has create_task() and create_future() methods and all created entities will be destroyed when TaskStore will be closed via close() method.

Also, a task store might create a linked copy of the self, which will be closed when the parent instance will be closed.

import asyncio from async_class import TaskStore async def main(): store = TaskStore(asyncio.get_event_loop()) task1 = store.create_task(asyncio.sleep(3600)) child_store = store.get_child() task2 = child_store.create_task(asyncio.sleep(3600)) await store.close() assert task1.done() and task2.done() asyncio.run(main())

About

Write classes with async def __ainit__

Источник