标签: python-asyncio

我正在使用 asyncio 构建一个应用程序。我将使用 sqlalchemy 作为 orm。据我了解，scoped_session 将会话与线程关联起来，这样它们就不会相互操作。现在，由于 asyncio 在单线程上工作，我相信scoped_session 将无法正常工作，这会导致问题。将 sqlalchemy 会话与 asyncio 结合使用的正确方法是什么？

我试图了解它是如何asyncio工作的。在我的场景中，客户端与服务器建立tcp连接，发送登录字符串（如果经过身份验证）-接收字符流。最后将字符串KeyboardInterrupt发送logoff到服务器并愉快地断开连接。

目前，我陷入了最后一部分，因为我的注销方法/任务在有机会完成之前就被破坏了。

^CTask was destroyed but it is pending!
source_traceback: Object created at (most recent call last):
  File "tst.py", line 101, in <module>
    client.login()
  File "tst.py", line 29, in login
    logoff_tsk = self.loop.create_task(self.logoff())
task: <Task pending coro=<logoff() running at tst.py:49> cb=[myClass._shutdown()] created at tst.py:29>

下面是产生此错误的代码：

from functools import partial
import asyncio

class myClass:
    def __init__(self, *a, **kw):
        self.transport = None
        self.protocol = None
        self.usr = str(kw.get("usr", ""))
        self.pwd = str(kw.get("pwd", ""))
        self.host …

我用这种逻辑编写了一个脚本，以便在生成许多记录时将其插入到 PostgreSQL 表中。

#!/usr/bin/env python3
import asyncio
from concurrent.futures import ProcessPoolExecutor as pool
from functools import partial

import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base


metadata = sa.MetaData(schema='stackoverflow')
Base = declarative_base(metadata=metadata)


class Example(Base):
    __tablename__ = 'example'
    pk = sa.Column(sa.Integer, primary_key=True)
    text = sa.Column(sa.Text)


sa.event.listen(Base.metadata, 'before_create',
    sa.DDL('CREATE SCHEMA IF NOT EXISTS stackoverflow'))

engine = sa.create_engine(
    'postgresql+psycopg2://postgres:password@localhost:5432/stackoverflow'
)
Base.metadata.create_all(engine)
session = sa.orm.sessionmaker(bind=engine, autocommit=True)()


def task(value):
    engine.dispose()
    with session.begin():
        session.add(Example(text=value))


async def infinite_task(loop):
    spawn_task = partial(loop.run_in_executor, None, task)
    while True:
        await asyncio.wait([spawn_task(value) for …

如何让协程因超时而停止？

我不明白为什么 asyncio.wait_for() 对我不起作用。我有这样一段代码（计划实现 telnet 客户端）：

def expect(self, pattern, timeout=20): 
    if type(pattern) == str:
        pattern = pattern.encode('ascii', 'ignore')        
    return self.loop.run_until_complete(asyncio.wait_for(self.asyncxpect(pattern), timeout))

async def asyncxpect(self, pattern): #receives data in a cumulative way until match is found
    regexp = re.compile(b'(?P<payload>[\s\S]*)(?P<pattern>%s)' %pattern)
    self.buffer = b''
    while True:
        # add timeout
        # add exception handling for unexpectedly closed connections
        data = await self.loop.sock_recv(self.sock, 10000) 
        self.buffer += data
        m = re.match(regexp, self.buffer)
        if m:
            payload = m.group('payload')
            match = m.group('pattern')
            return payload, match 

正如我所认为的，这段代码在某个时刻（在等待语句中）将控制权返回给事件循环。我认为当没有更多数据可以接收时应该发生这种情况。如果事件循环有控制权，它可以超时停止。

但是，如果服务器没有发送任何有用的（匹配的）我的代码就会在这个循环中绊倒，就在等待点。 …

我正在尝试对异步套接字服务器进行单元测试，并使用pytest-asynciopytest 与异步代码库兼容。服务器一旦启动，总是会通过 while 循环发送回复，并且可能花费大部分时间等待 中的传入消息client_loop()。问题是在单元测试框架终止事件循环并发出此警告之前，无法取消此任务：

任务已被销毁，但正在等待处理！

任务： < 任务待处理 coro=< Server.new_client() 已完成，在 /[...path...]/server.py:16 处定义> wait_for=< 未来待处理 cb=[< 位于 0x106d7cbe8 的 TaskWakeupMethWrapper 对象>() ]>>

我有权访问的唯一任务是由 创建的任务asyncio.create_task()，这似乎不是同一个任务。该任务如下所示：

任务： < 任务挂起 coro=< start_server() 运行在 /usr/local/Cellar/python/[...不同路径...]/streams.py:86>>

所以调用task.cancel(); await task.wait_cancelled()这个任务是没有效果的。

如何编写这个单元测试以干净地启动并为每个测试启动服务器而不切断可能仍在运行的任务？

这是例子：

测试服务器.py

import pytest
import asyncio

@pytest.fixture
async def server(event_loop):
    from server import Server
    the_server = Server()
    await the_server.start()
    yield the_server
    the_server.stop()

@pytest.mark.asyncio
async def test_connect(server):
    loop = asyncio.get_event_loop()
    reader, writer = await asyncio.open_connection('0.0.0.0', 8888, loop …

例如

shared = {}

async def coro1():
    # do r/w stuff with shared 

async def coro2():
    # do r/w stuff with shared 

asyncio.create_task(coro1())
asyncio.create_task(coro2())

如果coro1和coro2两个接入一个字典/变量，读取和写入，这需要某种形式的互斥/锁？或者它会很好，因为 asyncio 的东西只发生在 1 个线程上？

我正在尝试测试我是否正确使用 asyncio 库（aiofiles）来查看它是否被阻塞。

我怎样才能做到这一点？我的想法是让 asyncio.sleep 无限循环检查每个调用之间的延迟，同时我尝试在两者之间插入一个阻塞调用

这是我目前的尝试，但从blocking未被调用，我假设问题是无限循环实际上也在阻塞。

from asyncio import sleep, get_event_loop
import time


async def test_blocking():
  while True:
    timea = time.time()
    await sleep(1)
    timeb = time.time() - timea
    delay = timeb - 1.00
    if delay >= 0.01:
      print(f'Function blocked by {delay}s')

async def blocking():
  print('Running blocking function')
  time.sleep(5)


loop = get_event_loop()
loop.run_until_complete(test_blocking())
loop.run_until_complete(blocking())

我正在尝试使用 asyncio 和 Unix 域套接字在 Python 中构建一个玩具内存 Redis 服务器。

我的最小示例只返回baz每个请求的值：

import asyncio


class RedisServer:
    def __init__(self):
        self.server_address = "/tmp/redis.sock"

    async def handle_req(self, reader, writer):
        await reader.readline()
        writer.write(b"$3\r\nbaz\r\n")
        await writer.drain()
        writer.close()
        await writer.wait_closed()

    async def main(self):
        server = await asyncio.start_unix_server(self.handle_req, self.server_address)
        async with server:
            await server.serve_forever()

    def run(self):
        asyncio.run(self.main())


RedisServer().run()

当我测试有两个连续的客户端请求的redis客户端库与下面的脚本，它的工作原理：

import time
import redis


r = redis.Redis(unix_socket_path="/tmp/redis.sock")

r.get("foo")
time.sleep(1)
r.get("bar")

但是，如果我删除time.sleep(1)，有时它会起作用，有时第二个请求会失败，并且会出现以下任一情况：

Traceback (most recent call last):
  File "/tmp/env/lib/python3.8/site-packages/redis/connection.py", line 706, in send_packed_command …

我有以下两个异步功能

from tornado.httpclient import AsyncHTTPClient

async def get_categories(): # return a list of str
    # ....
    http = AsyncHTTPClient()
    resp = await http.fetch(....)
    return [....]

async def get_details(category): # return a list of dict
    # ....
    http = AsyncHTTPClient()
    resp = await http.fetch(....)
    return [....]

现在我需要创建一个函数来获取所有类别的详细信息（同时运行 http fetch）并将它们组合在一起。

async def get_all_details():
    categories = await get_categories()
    tasks = list(map(lambda x: {'category': x, 'task':get_details(x)}, categories))
    r = await asyncio.gather(*tasks) # error

# need to return [
#   {'category':'aaa', 'detail':'aaa detail 1'}, 
# …

我收到此错误：

D:\pythonstuff\demo.py:28: DeprecationWarning: 自 Python 3.8 起不推荐将协程对象显式传递给 asyncio.wait()，并计划在 Python 3.11 中删除。等待 asyncio.wait([

等了1秒！
等了5秒！
经过的时间：0小时：0分：5秒

进程以退出代码 0 结束

当我运行代码时：

import asyncio
import time

class class1():
    async def function_inside_class(self):
        await asyncio.sleep(1)
        print("Waited 1 second!")

    async def function_inside_class2(self):
        await asyncio.sleep(5)
        print("Waited 5 second!")

def tic():
    global _start_time
    _start_time = time.time()

def tac():
    t_sec = round(time.time() - _start_time)
    (t_min, t_sec) = divmod(t_sec,60)
    (t_hour,t_min) = divmod(t_min,60)
    print('Time passed: {}hour:{}min:{}sec'.format(t_hour,t_min,t_sec))

object = class1()


async def main():
    tic()
    await asyncio.wait([
        object.function_inside_class(),
        object.function_inside_class2()
        ])
    tac() …