标签: python-asyncio

我正在尝试,到目前为止,未能使用python asyncio访问串行端口.

我非常感谢在简单的fd上使用新的python异步框架的任何提示.

干杯!

詹姆士

我正在尝试使用我正在使用以下命令开发的Django应用程序使用Gunicorn中的gaiohttp工作程序:

gunicorn -k gaiohttp -b localhost:8080 myproject.wsgi

我的最终目标是能够同时处理请求 - 即让1名gunicorn工作者同时处理多个请求.有I/O绑定操作使这些请求变慢.

我知道当我处理请求时,事件循环已经在运行:

class MyView(View):

    def get(self, request):
        loop = asyncio.get_event_loop()
        loop.is_running() # True
        ...

问题:

1. 如何yield from asyncio.sleep(10)在视图代码中执行操作？

   class MyView(View):
   
       def get(self, request):
           # Raises AssertionError: yield from wasn't used with future
           yield from asyncio.sleep(10)
   

   Run Code Online (Sandbox Code Playgroud)

2. 我可以向事件循环添加任务,但是在处理请求时它们不会阻塞

   @asyncio.coroutine
   def simulate_work():
       yield from asyncio.sleep(10)
   
   class MyView(View):
   
       def get(self, request):
           # This runs simulate_work(), however, it doesn't block 
           # the response is returned before simulate_work() is finished
           loop = asyncio.get_event_loop()
           task = …

   Run Code Online (Sandbox Code Playgroud)

考虑一个调用另一个协程的协程:

async def foo(bar):
     result = await bar()
     return result

如果bar是协程,这可以正常工作.我需要做什么(即我需要将调用包装起来bar),以便这个代码在bar正常函数中做正确的事情？

async def即使它从不做任何异步(即从不使用await),也完全有可能定义一个协同程序.但是,问题是如何bar在代码中包装/修改/调用常规函数foo,bar以便等待.

我有一个REST API包装器,应该在交互式Python会话中运行.HTTP请求既可以通过自动后台线程(使用API​​包装器),也可以由最终用户通过交互式会话手动完成.我试图将所有HTTP请求管理从前一个新的每线程请求方法迁移到asyncio,但由于我无法在主线程中运行asyncio循环(它必须是免费的ad-hoc Python命令/请求),我写了以下内容在后台线程中运行它:

import aiohttp
import asyncio
from concurrent.futures import ThreadPoolExecutor

def start_thread_loop(pool=None):
    """Starts thread with running loop, bounding the loop to the thread"""
    def init_loop(loop):
        asyncio.set_event_loop(loop)  # bound loop to thread
        loop.run_forever()
    _pool = ThreadPoolExecutor() if pool is None else pool
    loop = asyncio.new_event_loop()
    future = _pool.submit(init_loop, loop)
    return future, loop

def send_to_loop(coro, loop):
    """Wraps couroutine in Task object and sends it to given loop"""
    return asyncio.run_coroutine_threadsafe(coro, loop=loop)

实际的API包装器类似于以下内容:

class Foo:
    def __init__(self):
        _, self.loop = start_thread_loop()
        self.session = …

Python3.6 现在异步可迭代。是否有内置方法可以将异步可迭代转换为同步可迭代。

我目前有这个辅助函数，但感觉非常不Pythonic。有一个更好的方法吗？

async def aiter_to_list(aiter):
    l = []
    async for i in aiter:
        l.append(i)
    return l

Flask 2.0 中介绍了 async/await 的用法。（https://flask.palletsprojects.com/en/2.0.x/async-await/）

我正在使用 Flask-RestX，那么是否可以在 RestX 请求处理程序中使用async/ ？await

就像是：

@api.route('/try-async')
class MyResource(Resource):
    @api.expect(some_schema)
    async def get(self):
        result = await async_function()
        return result

不起作用，当我尝试到达此端点时，出现错误：

TypeError: Object of type coroutine is not JSON serializable

有这方面的信息吗？

封装版本：

flask==2.0.1
flask-restx==0.4.0

我也flask[async]按照文档的建议安装了。

我正在尝试以下操作：

class Payload_Session_Generator:
    def __init__(self):
        pass

    async def __anext__(self):
        async for payload in generate_fb_payload():
            if type(payload) != str:
                yield payload
            else:
                StopAsyncIteration

    def __aiter__(self):
        return self

然后将其作为实例传递给不同的函数，并__aiter__显式调用该方法，并且该方法_iter是上述类的对象：

chunk = await self._iter.__anext__()

这会产生以下错误：

类型错误：对象 async_generator 不能在“await”表达式中使用

想象一下一个异步aiohttpWeb 应用程序，它由通过连接的 Postgresql 数据库支持asyncpg，并且不执行其他 I/O。我怎样才能有一个托管应用程序逻辑的中间层，而不是异步的？（我知道我可以简单地使所有内容异步 - 但想象我的应用程序具有大量应用程序逻辑，仅受数据库 I/O 约束，并且我无法触及其中的所有内容）。

伪代码：

async def handler(request):\n    # call into layers over layers of application code, that simply emits SQL\n    ...\n\ndef application_logic():\n    ...\n    # This doesn\'t work, obviously, as await is a syntax\n    # error inside synchronous code.\n    data = await asyncpg_conn.execute("SQL")\n    ...\n    # What I want is this:\n    data = asyncpg_facade.execute("SQL")\n    ...\n

如何asyncpg构建同步 fa\xc3\xa7ade 来允许应用程序逻辑进行数据库调用？async.run()在这种情况下，诸如 using或等浮动的配方asyncio.run_coroutine_threadsafe()不起作用，因为我们来自已经异步的上下文。我认为这不可能是不可能的，因为已经有一个事件循环原则上可以运行协 …

我需要在异步函数中返回一个值。我尝试使用同步形式的返回：

import asyncio

async def main():
    for i in range(10):
        return i
        await asyncio.sleep(1)

print(asyncio.run(main()))

输出：

0 [Finished in 204ms]

但它只是返回第一个循环的值，而不是expexted。于是将代码修改如下：

import asyncio

async def main():
    for i in range(10):
        yield i
        await asyncio.sleep(1)

for _ in main():
    print(_)

输出：

TypeError: 'async_generator' object is not iterable

通过使用异步生成器我面临这个错误。如何为异步函数的每个循环返回一个值？

谢谢

有这个简单的代码：

import asyncio

async def main():
    f = asyncio.Future()
    await f

asyncio.run(main())

协程（此处main）可以等待 Future 对象。它基本上会被阻止，直到f有结果或设置异常，或者直到它们被取消。

出于好奇，我想知道“这种等待是如何发生的”？我具体检查了Task.py的 Python 实现def __step()。

在最简单的形式中，快乐的情况下，当返回的result是 Future 时，它​​是：

.
.
result = coro.send(None)
.
.
blocking = getattr(result, '_asyncio_future_blocking', None)
.
.
if blocking:   # So `result` is a Future with `_asyncio_future_blocking == True`
    result._asyncio_future_blocking = False
    result.add_done_callback(self.__wakeup, context=self._context)
    self._fut_waiter = result
    if self._must_cancel:
        if self._fut_waiter.cancel(msg=self._cancel_message):
            self._must_cancel = False

我从本节中得到了所有其他答案（例如当它result是裸露的yield或CancellationError发生时等等），除了这个！ …