标签: python-asyncio

我的 Python 脚本包含一个循环，用于subprocess在脚本外部运行命令。每个子进程都是独立的。我监听返回的消息，以防出现错误；我不能忽略子进程的结果。这是没有 asyncio 的脚本（我用 替换了计算量大的调用sleep）：

from subprocess import PIPE  # https://docs.python.org/3/library/subprocess.html
import subprocess

def go_do_something(index: int) -> None:
    """
    This function takes a long time
    Nothing is returned
    Each instance is independent
    """
    process = subprocess.run(["sleep","2"],stdout=PIPE,stderr=PIPE,timeout=20)
    stdout = process.stdout.decode("utf-8")
    stderr = process.stderr.decode("utf-8")
    if "error" in stderr:
        print("error for "+str(index))
    return

def my_long_func(val: int) -> None:
    """
    This function contains a loop
    Each iteration of the loop calls a function
    Nothing is returned
    """
    for index …

I need to wait until a certain condition/equation becomes True in a asynchronous function in python. Basically it is a flag variable which would be flagged by a coroutine running in asyncio.create_task(). I want to await until it is flagged in the main loop of asyncio. Here's my current code:

import asyncio
flag = False

async def bg_tsk():
    await asyncio.sleep(10)
    flag = True

async def waiter():
    asyncio.create_task(bg_tsk())
    await asyncio.sleep(0)
    # I find the below part unpythonic
    while not flag:
        await …

我正在尝试在 FastAPI 中连接到 mongodb。我反复收到此异常。

文件 - main.py

app = FastAPI(
    title=config.PROJECT_NAME, docs_url="/api/docs", openapi_url="/api"
)

@app.get("/api/testing")
async def testit():
    user_collection = readernetwork_db.get_collection("user_collection")
    all_users = await user_collection.find_one({"email": "sample_email"})
    print("all users --- ", all_users)
    return all_users

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", reload=True, port=8888)

文件 - session.py

import motor.motor_asyncio
from app.core import config


print("here we go again....")
client = motor.motor_asyncio.AsyncIOMotorClient(
    config.MONGOATLAS_DATABASE_URI)
readernetwork_db = client.get_database("readernetwork")

例外 -：

all_users = await user_collection.find_one({"email": "sample_email"})

RuntimeError: Task <Task pending name='Task-4' coro=<RequestResponseCycle.run_asgi() running at /usr/local/lib/python3.8/site-packages/uvicorn/protocols/http/h11_impl.py:389> cb=[set.discard()]> got Future …

我正在尝试使用 Python asyncio 子进程来启动交互式 SSH 会话并自动输入密码。实际用例并不重要，但它有助于说明我的问题。这是我的代码：

    proc = await asyncio.create_subprocess_exec(
        'ssh', 'user@127.0.0.1',
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
        stdin=asyncio.subprocess.PIPE,
    )

    # This loop could be replaced by async for, I imagine
    while True:
        buf = await proc.stdout.read()
        if not buf:
            break
        print(f'stdout: { buf }')

我希望它能像 asyncio 流一样工作，我可以在其中创建两个任务/子例程/future，一个用于监听StreamReader（在本例中由 给出proc.stdout），另一个用于写入 StreamWriter ( proc.stdin)。

但是，它并没有按预期工作。ssh 命令的前几行输出直接打印到终端，直到出现密码提示（或主机密钥提示，视情况而定）并等待手动输入。我希望能够读取前几行，检查它是否要求输入密码或主机提示，并相应地写入 StreamReader。

它运行该行的唯一一次print(f'stdout: { buf }')是在我按 Enter 键后，当它打印时，显然，“stderr：b'主机密钥验证失败。\r\n'”。

我也尝试了推荐的proc.communicate()，它不像使用 StreamReader/Writer 那么简洁，但它有同样的问题：等待手动输入时执行冻结。

这实际上应该如何运作？如果这不是我想象的那样，为什么不呢？有没有什么方法可以实现这一点，而无需诉诸线程中的某种繁忙循环？

PS：我解释使用 ssh 只是为了清楚起见。我最终使用 plink 来实现我想要的功能，但我想了解如何使用 python 来运行任意命令。

我用 python 和 asyncio 创建了一个简单的 HTTP 服务器。但是，我读到基于 asyncio 的服务器只能利用一个 CPU 核心。我正在尝试解决这个问题，multiprocessing但它不起作用。当我尝试创建进程时，它给我一个错误，指出它无法从 _winapi 创建进程。到目前为止，这是我的代码：

def serve_forever(self, host, port):
    srv, loop = self.init(host, port)
    print()
    if self.name:
        print('* Serving App {}'.format(self.name))
    print('* Serving On http://{host}:{port}'.format(host=host, port=port))
    print('* Press <CTRL-C> To Quit')
    workers = []
    try:
        for i in range(mp.cpu_count()-1):
            p = mp.Process(target=loop.run_forever)
            workers.append(p)
            p.start()
        loop.run_forever()
    except KeyboardInterrupt:
        print()
        for p in workers:
            p.terminate()
        srv.close()
        loop.run_until_complete(srv.wait_closed())
        loop.close()
        pass

顺便说一句，该self.init功能有效。

我有两个 websocket 服务器，分别称为 Main 和 Worker，这是所需的工作流程：

\n
• 客户端向Main发送消息
\n
• Main向Worker发送消息
\n
• Worker 响应 Main
\n
• 主要响应客户端
\n

这可行吗？我在 Channels 中找不到任何 WS 客户端功能。我天真地尝试这样做（在consumers.py）：

import websockets\n\nclass SampleConsumer(AsyncWebsocketConsumer):\n    async def receive(self, text_data):\n        async with websockets.connect(url) as worker_ws:\n            await worker_ws.send(json.dumps({ 'to': 'Worker' }))\n            result = json.loads(await worker_ws.recv())\n        await self.send(text_data=json.dumps({ 'to': 'Client' })\n

然而，似乎该with部分被阻止（Main 似乎不接受任何进一步的消息，直到收到 Worker 的响应）。我怀疑这是因为websockets运行了自己的循环，但我不确定。（编辑：我比较了一下id(asyncio.get_running_loop())，它似乎是相同的循环。我不知道为什么它会阻塞。）

响应{ "to": "Client" }不需要在这里，即使它是在不同的方法中，我也可以，只要它在收到 Worker 的响应时触发即可。

有没有办法做到这一点，还是我找错了树？

如果没有办法做到这一点，我正在考虑有一个线程（或进程？或单独的应用程序？）与 Worker 通信，并用于channel_layer …

我有一个使用 FastAPI 构建的 API，该端点将任务提交给 celery 工作人员，等待工作人员完成其工作并将结果返回给用户。

问题是等待结果的正确方法是什么？

端点代码

from tasks import celery_application, some_task
from celery.result import AsyncResult

@api.post('/submit')
async def submit(data: str):
    task = some_task.apply_async(kwargs={'data': data}, queue='some_queue')
    result = AsyncResult(id=task.task_id, app=celery_application).get()
    return {'task_result': result}

问题在于AsyncResult该get方法会阻塞应用程序，它会同步等待结果，同时 api 会冻结。

我想出的解决方案之一是在循环中检查 n 秒的结果

from tasks import celery_application, some_task
import asyncio
import redis


r = redis.Redis.from_url(REDIS_CONN_URI)


@api.post('/submit')
async def submit(data: str):
    task = some_task.apply_async(kwargs={'data': data}, queue='some_queue')
    result = None
    for _ in range(100):
        if r.exists(task.task_id):
            result = r.get(task.task_id)
            break
        await …

当尝试使用 asyncio 扩展连接到 SQLite 引擎时，从 SQLAlchemy 版本升级1.4.0b2到 会导致以下错误。1.4.0b3

>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> engine = create_async_engine('sqlite:///database.db', echo=True, future=True)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<...>/venv/lib/python3.9/site-packages/sqlalchemy/ext/asyncio/engine.py", line 41, in create_async_engine
    return AsyncEngine(sync_engine)
  File "/.../venv/lib/python3.9/site-packages/sqlalchemy/ext/asyncio/engine.py", line 531, in __init__
    raise exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: The asyncio extension requires an async driver to be used. The loaded 'pysqlite' is not async.

这是由于 SQLite 驱动程序和 asyncio 扩展之间不兼容造成的吗？但是，如果是这种情况，为什么它适用于 version 1.4.0b2？

>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> …

我正在使用discord.py 制作一个Discord 机器人，有时我用来运行它的脚本需要关闭该机器人。当我在不使用信号处理程序的情况下关闭它时，会出现很多关于循环未关闭的错误，因此我添加了一个信号处理程序（使用下面的代码），并且在内部我需要调用client.close()and client.logout()，但问题是这些是异步函数和因此要求我等待它们，但我无法等待这些函数，因为信号处理程序不能是异步函数。

这是代码：

def handler():
    print("Logging out of Discord Bot")
    client.logout()
    client.close()
    sys.exit()

@client.event
async def on_ready():
    print('We have logged in as {0.user}'.format(client))

    for signame in ('SIGINT', 'SIGTERM'):
        client.loop.add_signal_handler(getattr(signal, signame),
                                lambda: asyncio.ensure_future(handler()))

有没有办法使用信号处理程序正确注销，或者至少只是沉默代码内部的警告和错误，这样控制台中就不会打印任何错误。

我试图限制使用信号量同时运行的异步函数的数量，但我无法让它工作。我的代码归结为：

import asyncio


async def send(i):

    print(f"starting {i}")
    await asyncio.sleep(4)
    print(f"ending {i}")


async def helper():
    async with asyncio.Semaphore(value=5):
        await asyncio.gather(*[
            send(1),
            send(2),
            send(3),
            send(4),
            send(5),
            send(6),
            send(7),
            send(8),
            send(9),
            send(10),
        ])


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(helper())
    loop.close()

输出是：

starting 1
starting 2
starting 3
starting 4
starting 5
starting 6
starting 7
starting 8
starting 9
starting 10
ending 1
ending 2
ending 3
ending 4
ending 5
ending 6
ending 7
ending 8
ending 9
ending …