标签: python-asyncio

正如这个问题中所述，我们可以使用装饰器asyncio.coroutine将函数转换为协程，如下所示：

def hello():
    print('Hello World!')

async_hello = asyncio.coroutine(hello)

然而，从 python 3.8 开始，这个函数已被弃用（由 替代async def ...）。那么我们如何在 3.8+ 中做到这一点而不需要呢async def ...？

我在 Python 3.7.3 中运行此代码

import asyncio

async def fun(time):
    print(f"will wait for {time}")
    await asyncio.sleep(time)
    print(f"done waiting for {time}")

async def async_cenas():
    t1 = asyncio.create_task(fun(1))
    print("after 1")
    t2 = asyncio.create_task(fun(2))
    print("after 2")

def main():
    t1 = asyncio.run(async_cenas())
    print("ok main")
    print(t1)

if __name__ == '__main__':
    main()
    print("finished __name__")

并得到这个输出：

after 1
after 2
will wait for 1
will wait for 2
ok main
None
finished __name__

我还期待看到：

done waiting for 1
done waiting for 2

即，为什么期望asyncio.run(X)会等待协程完成后再继续。

我正在运行一个 python 程序来监听 azure iot hub。该函数返回一个协程对象而不是 json。我看到，如果我们使用异步函数并将其作为普通函数调用，就会发生这种情况，但我创建了一个循环来获取事件，然后使用 run_until_complete 函数。我在这里缺少什么？

async def main():
    try:
        client = IoTHubModuleClient.create_from_connection_string(connection_string)
        print(client)
        client.connect() 
        while True:
            message = client.receive_message_on_input("input1")   # blocking call
            print("Message received on input1")
            print(type(message))
            print(message)

        
    except KeyboardInterrupt:
        print ( "IoTHubClient sample stopped" )
    except:
        print ( "Unexpected error from IoTHub" )
        return

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

输出- 在 input1 上收到消息 <class 'coroutine'> <coroutine object receive_message_on_input at 0x7f1439fe3a98>

我正在Python中学习asyncio，我想尝试使用asyncio。我的目标是连续读取用户的输入并使用该输入创建需要异步运行的作业/任务。

import asyncio
import os
loop = asyncio.get_event_loop()

async def action():
    inp = int(input('enter: '))
    await asyncio.sleep(inp)
    os.system(f"say '{inp} seconds waited'")

async def main():
    while True:
        await asyncio.ensure_future(action())

try:
    asyncio.run(main())
except Exception as e:
    print(str(e))
finally:
    loop.close()

我搞砸了一些事情，我想知道如何实现它。每次用户输入一个数字时，脚本都需要休眠给定的时间，然后说它已经等待了。整个事情需要同时进行。如果用户输入 100 作为输入，脚本应该启动一个任务休眠 100 秒，但在用户端，它需要在用户输入后立即再次请求输入。

我是 asyncio 的新手，我想利用它来为我的 pyqt 应用程序处理通过 tcp 连接的通信。

我制作了这个简单的演示来学习如何在 asyncio 上下文中处理 QT 循环。我看过与此相关的其他帖子，但它们比我目前想做的要复杂得多。因此，我在一个单独的窗口中启动服务器客户端，以便它进行监听，并尝试通过小部件上的简单按钮单击事件发送消息。正如它所得到的准系统......我的问题是它不起作用。

我希望能够进行一次信息交换以及端口对流保持开放的情况。我认为这些任务在 asyncio 中会很简单，但在这一点上让它与 qt 很好地配合似乎很困难。

现在我得到了

RuntimeWarning: coroutine 'PushButton.sendmessage' was never awaited
  rslt = self.__app.exec_()
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

我不知道从哪里开始解决这个问题。

测试.py

from PyQt5.QtWidgets import QApplication, QWidget, QPushButton
from PyQt5.QtGui import QIcon
from PyQt5.QtCore import Qt, pyqtSlot
import sys
import asyncio
from asyncqt import QEventLoop
from async_client import tcp_echo_client


class PushButton(QWidget):

    loop = None

    def __init__(self,app_loop):
        super(PushButton,self).__init__()
        self.initUI()
        self.loop = loop

    def initUI(self):
        self.setWindowTitle("PushButton") …

我正在尝试在我正在开发的使用 aiogram 的电报机器人中发送加密货币的价格警报。我遇到的问题是，我不确定如何启动一个函数作为后台、非阻塞线程，然后继续启动调度程序。我知道如何使用标准同步电报机器人来做到这一点，但我对我应该用 aiogram 做什么感到困惑。我读到我可以dp.loop.create_task 在这里使用，但这会引发错误Nonetype has no attribute create_task。这是我尝试执行这些线程的代码：

print('Starting watchlist process, this needs to run as a non blocking daemon...')
dp.loop.create_task(wl.start_process())
print('Starting broadcaster, this needs to run as a non blocking daemon ... ')
dp.loop.create_task(broadcaster())
print('Starting the bot ...')
executor.start_polling(dp, skip_updates=True)

我只需要wl.start_process和broadcaster函数在后台运行。我该如何实现这个目标？

这是start_process：

async def start_process(self):
    """
    Start the watchlist process.
    :return:
    """
    threading.Thread(target=self.do_schedule).start()
    await self.loop_check_watchlist()

这是broadcaster：

async def broadcaster():

count = 0
while True:
    uid_alerts …

先问问题，再看上下文。如何使用异步 gRPC python 服务器基于从客户端取消 RPC 来执行某些服务器端操作（例如清理）？

在我的微服务中，我有一个asynciogRPC 服务器，其主要 RPC 是双向流。

在客户端（也使用 asyncio），当我取消某些操作时，它会引发一个asyncio.CancelledError被捕获且不会被grpc核心重新引发的异常：

https://github.com/grpc/grpc/blob/master/src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi#L679

    except asyncio.CancelledError:
        _LOGGER.debug('RPC cancelled for servicer method [%s]', _decode(rpc_state.method()))

asyncio.CancelledError所以我不能依赖于在我自己的代码中捕获，因为它是预先捕获的并且不会重新引发。

共享上下文应该包含有关 RPC 是否在客户端被取消的信息，通过.cancel()从 RPC 调用进行调用并能够通过调用查看它是否被取消.cancelled()：

https://grpc.github.io/grpc/python/grpc_asyncio.html#shared-context

抽象取消()

Cancels the RPC.

Idempotent and has no effect if the RPC has already terminated.

Returns

    A bool indicates if the cancellation is performed or not.
Return type

    bool

Run Code Online (Sandbox Code Playgroud)

摘要取消()

Return True if the RPC is cancelled.

The …

Run Code Online (Sandbox Code Playgroud)

我有一个异步get_forecastweather函数，它为我提供 JSON 天气数据，我知道我们无法在同步中执行异步函数，但是我如何在单独的线程中执行它，需要帮助，提前致谢

def weather_detail(request):
    if request.method == 'GET':
        city_name = 'my_city'
        key = 'mykey'
        result = None
        with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
            response = executor.submit(get_forecastweather,city_name,key)
            result = response.result()
        print('Result from thread ',result)
        return render(request,'weather/weather_detail.html')

我得到的错误是

 RuntimeWarning: coroutine 'get_forecastweather' was never awaited
  response = wrapped_callback(request, *callback_args, **callback_kwargs)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

我制作了一个小应用程序，用于SQLAlchemy处理与 postgresql 数据库的连接。现在我想用asincio重写它。由于某种原因，当我运行它时，出现以下错误：

Traceback (most recent call last):
  File "D:\Space\discord_count_bot\bot\bot\main.py", line 12, in <module>
    dbConnection.init_connection(
  File "D:\Space\discord_count_bot\bot\bot\db_hanler.py", line 78, in init_connection
    engine = create_async_engine(connection_string, future=True, echo=True)
  File "D:\Space\discord_count_bot\bot_env\lib\site-packages\sqlalchemy\ext\asyncio\engine.py", line 40, in create_async_engine
    sync_engine = _create_engine(*arg, **kw)
  File "<string>", line 2, in create_engine
  File "D:\Space\discord_count_bot\bot_env\lib\site-packages\sqlalchemy\util\deprecations.py", line 298, in warned
    return fn(*args, **kwargs)
  File "D:\Space\discord_count_bot\bot_env\lib\site-packages\sqlalchemy\engine\create.py", line 560, in create_engine
    dbapi = dialect_cls.dbapi(**dbapi_args)
  File "D:\Space\discord_count_bot\bot_env\lib\site-packages\sqlalchemy\dialects\postgresql\psycopg2.py", line 782, in dbapi
    import psycopg2
ModuleNotFoundError: No module named 'psycopg2'

如果psycopg2安装了，我得到

Traceback (most …

我有一个异步代码，如下所示：

• 有一个第三方函数可以对字符串执行一些操作并返回修改后的字符串，就这个问题而言，它类似于non_async_func.

• 我有一个async def async_func_single包含non_async_func执行单个操作的函数。

• 然后是另一个async def async_func_batch嵌套环绕的函数async_func_single来对一批数据执行该函数。

代码可以工作，但我想更多地了解为什么/如何工作，我的问题是

• 是否有必要创建async_func_single并async_func_batch环绕它？

• 我可以直接输入一批数据来async_func_batch调用吗non_async_func？

• 我有一个 per_chunk 函数，可以批量输入数据，是否有任何异步操作/函数可以避免使用预批处理我想要发送的数据async_func_batch？

import nest_asyncio
nest_asyncio.apply()

import asyncio
from itertools import zip_longest

from loremipsum import get_sentences

def per_chunk(iterable, n=1, fillvalue=None):
  args = [iter(iterable)] * n
  return zip_longest(*args, fillvalue=fillvalue)

def non_async_func(text):
  return text[::-1]

async def async_func_single(text):
  # Perform some string operation.
  return non_async_func(text)

async def async_func_batch(batch):
  tasks = [async_func_single(text) for …