标签: python-asyncio

我正在从烧瓶转移到 aiohttp，我需要在不支持异步的 Oracle 数据库中执行一些查询。所以我想知道如何在 aiohttp 中做到这一点？

这个怎么样？

http://pastebin.com/nbWABbvK

或者还有其他（正确的）方法来做到这一点？

提前致谢！

在 asyncio 中工作时，我正在尝试使用列表理解来构建我的任务列表。该函数的基本形式如下：

import asyncio
import urllib.request as req
@asyncio.coroutine
def coro(term):
    print(term)
    google = "https://www.google.com/search?q=" + term.replace(" ", "+") + "&num=100&start=0"
    request = req.Request(google, None, headers) 
    (some beautiful soup stuff)

我的目标是使用术语列表来创建我的任务列表：

terms = ["pie", "chicken" ,"things" ,"stuff"]    
tasks=[
    coro("pie"),
    coro("chicken"),
    coro("things"),
    coro("stuff")]

我最初的想法是：

loop = asyncio.get_event_loop()
tasks = [my_coroutine(term) for term in terms]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

这不会创建在列表理解期间运行该函数的任务列表。有没有办法使用快捷方式来创建任务列表而无需编写每个任务？

使用async/await?执行在 Python 中并行运行的两个异步循环的好方法是什么？

我也想过类似下面的代码，但不能换我围绕着如何使用头async/ await/EventLoop在这种特殊情况下。

import asyncio

my_list = []

def notify():
    length = len(my_list)
    print("List has changed!", length)

async def append_task():
    while True:
        time.sleep(1)
        await my_list.append(random.random())
        notify()

async def pop_task():
    while True:
        time.sleep(1.8)
        await my_list.pop()
        notify()

loop = asyncio.get_event_loop()
loop.create_task(append_task())
loop.create_task(pop_task())
loop.run_forever()

预期输出：

$ python prog.py
List has changed! 1 # after 1sec
List has changed! 0 # after 1.8sec
List has changed! 1 # after 2sec
List has changed! 2 # …

我正在尝试编写一个简单的子进程程序，它将调用一个长时间运行的 shell 命令，允许其他进程运行，然后在完成后执行一些清理任务。

不幸的是，我遇到了错误，只是让 shell 命令在 asyncio 事件循环中正确执行。行为是它看起来 python 从不等待 shell 脚本完成运行。我知道 shell 脚本可以工作，因为我可以从提示符手动运行它。我正在运行的 shell 脚本应该在大约 3-5 分钟内执行。

这是我的示例程序：

import asyncio 
from asyncio.subprocess import PIPE, STDOUT 
import subprocess 
import signal


def signal_handler(signal, frame):
    loop.stop()
    client.close()
    sys.exit(0)

async def run_async(loop = ''):
    cmd = 'sudo long_running_cmd --opt1=AAAA --opt2=BBBB'

    print ("[INFO] Starting script...")
    await asyncio.create_subprocess_shell(cmd1, stdin = PIPE, stdout = PIPE, stderr = STDOUT)
    print("[INFO] Script is complete.")


loop = asyncio.get_event_loop() 
signal.signal(signal.SIGINT, signal_handler) 
tasks = [loop.create_task(run_async())] 
wait_tasks = asyncio.wait(tasks) 
loop.run_until_complete(wait_tasks)

loop.close()

该程序几乎立即运行并失败。此代码生成的错误是： …

我使用aiohttp. 有没有办法在每个请求完成后立即获得结果？

也许使用类似的东西async for？还是 Python 3.6 异步生成器？

目前我await asyncio.gather(*requests)并在所有这些完成后处理它们。

我有一个 Python C 扩展，它从套接字获取数据并处理它，在处理过程中它会释放 GIL。目前我使用 2 个 Python 线程，由于 GIL 释放，它们很好地交织在一起，导致 2 个 CPU 内核上的负载为 90%。

我将如何在 Python 3 asyncio 中实现相同的目标？我找不到正确的 Python-C-API 命令来告诉事件循环它可以去做其他事情。

或者换句话说：如果我的扩展释放了 GIL，这是否自动意味着它不会阻止事件循环中可用的下一项的执行？例如，当我的 C 扩展正在处理第一个套接字的数据时，会读取下一个套接字吗？这是我在任何地方都找不到的。根据我的理解，我可以从多个套接字获取数据，并让更多的 CPU 内核工作。

我正在尝试创建一个简单的监控系统，它会定期检查事物并记录它们。这是我尝试使用的逻辑的简化示例，但我不断收到RuntimeWarning: coroutine 'foo' was never awaited错误消息。

我应该如何从自身重新安排异步方法？

test.py 中的代码：

import asyncio
from datetime import datetime

async def collect_data():
    await asyncio.sleep(1)
    return {"some_data": 1,}

async def foo(loop):
    results = await collect_data()
    # Log the results
    print("{}: {}".format(datetime.now(), results))
    # schedule to run again in X seconds
    loop.call_later(5, foo, loop)

if __name__ == '__main__':

    loop = asyncio.get_event_loop()
    loop.create_task(foo(loop))
    loop.run_forever()
    loop.close()

错误：

pi@raspberrypi [0] $ python test.py 
2018-01-03 01:59:22.924871: {'some_data': 1}
/usr/lib/python3.5/asyncio/events.py:126: RuntimeWarning: coroutine 'foo' was never awaited
  self._callback(*self._args)

我想实现一个 awaitable 并注意到__await__“需要”是一个生成器。

从PEP-492：

具有__await__返回迭代器的方法的对象。

...

__await__在本 PEP 的其余部分中，带有方法的对象称为 Future-like 对象。

如果__await__返回除迭代器以外的任何内容，则为 TypeError 。

根据我的经验，beforeawait是一个语句，yield from与作为生成器实现的协程一起使用。如今，python（我使用的是 3.5）具有使用async def语法的异步方法。因此，我认为该yield from语法是旧的/已弃用的。

所以我打破了解释器，看看这是否有效：

>>> class A:
...     def __await__(self):
...         yield from (asyncio.sleep(1).__await__())
...         return 'spam'
... 
>>> a = A()
>>> loop = asyncio.get_event_loop()
>>> loop.run_until_complete(a)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib64/python3.5/asyncio/base_events.py", line 467, in run_until_complete
    return future.result() …

我已经在 python 应用程序中苦苦挣扎了几天，我希望在一个文件夹中查找一个或多个文件，并遍历每个文件和其中的每个记录，并创建要保留在 Janusgraph 数据库上的对象。我正在使用的特定 OGM 要求使用 asyncio 以异步方式完成与数据库的事务。我已经阅读了很多关于 asyncio 的博客和帖子，我想我理解 async、await、tasks 等的概念......在我的应用程序中，我定义了几个处理不同部分处理的函数：

• 检索所有可用文件的列表
• 选择一个文件进行处理
• 遍历选定的文件并读取一行/记录进行处理
• 接收记录，确定解析 from in 并调用其他几个函数，这些函数负责在将 Model 对象持久化到数据库之前创建它们。例如，我创建了不同的函数：用户、会话、浏览器、设备使用、服务器等...

我理解（我可能是错的）使用 asyncio 的最大优势是在调用函数通常会因 I/O、数据库事务、网络延迟等而阻塞的情况下......

所以我的问题是我是否需要将所有函数转换为协程并安排运行事件循环，或者只是那些会阻塞的函数，例如将事务提交到数据库。我一开始就尝试过这种方法，但遇到了各种各样的问题。

我正在使用 Python3.5 和 Tkinter 创建一个加密货币交换 API 客户端。我有几个显示，我想每 10 秒异步更新一次。我可以Tk.after()在这个例子中使用像每 10 秒更新一次显示

def updateLoans():
    offers = dd.loanOffers()
    demands = dd.loanDemands()
    w.LoanOfferView.delete(1.0, END)
    w.LoanDemandView.delete(1.0, END)
    w.LoanOfferView.insert(END, offers)
    w.LoanDemandView.insert(END, demands)
    print('loans refreshed')

    root.after(10000, updateLoans)

为了使该after方法每 10 秒持续更新一次，该函数updateLoans()需要作为可调用对象传递到after()函数内部。

现在难倒我的部分，当我使这个函数与 python 的新 async 和 await 关键字异步时

async def updateLoans():
    offers = await dd.loanOffers()
    demands = await dd.loanDemands()
    w.LoanOfferView.delete(1.0, END)
    w.LoanDemandView.delete(1.0, END)
    w.LoanOfferView.insert(END, offers)
    w.LoanDemandView.insert(END, demands)
    print('loans refreshed')

    root.after(10000, updateLoans)

这里的问题是我不能在after方法的参数中等待可调用对象。所以我收到运行时警告。RuntimeWarning: coroutine 'updateLoans' was never …