标签: python-asyncio

我正在（龙卷风）网络服务器上运行一个长时间的 CPU 密集型任务。我不想将此任务卸载到线程（目前）。在这个长时间运行的任务中，如何正确地“yield”控制事件循环，以便网络请求仍然得到服务（注意：我在这里以合作调度的方式使用“yield”，而不是在生成器的意义上使用“yield”） ，或 python 关键字yield）。

我的建议是做一个await asyncio.sleep(0)，但是是否可以保证这不只是作为 NOP 实现？我一直在 asyncio 库中寻找专用函数来执行此操作，但到目前为止尚未找到。

示例（python 3.5 async/await 风格）：

async def long_task():
    for i in range(LARGE_NUMBER):
        do_something(i)
        if i % 100 == 0:
            await asyncio.sleep(0)

我正在编写一个程序，它通过 AMQP 接受 RPC 请求以执行网络请求 (CoAP)。处理 RPC 请求时，aioamqp 回调会生成负责网络 IO 的任务。这些任务可以被视为后台任务，它们将无限期地运行以通过 AMQP 流式传输网络响应（在这种情况下，一个 RPC 请求会触发 RPC 响应和数据流）。

我注意到，在我的原始代码中，网络任务将在看似随机的时间间隔后（在完成之前）被销毁，asyncio 然后会打印以下警告“任务已被销毁但正在挂起”。此问题与此处描述的问题类似：https://bugs.python.org/issue21163。

现在，我通过在模块级列表中存储硬引用来规避这个问题，这可以防止 GC 破坏任务对象。但是，我想知道是否有更好的解决方法？理想情况下，我想在 RPC 回调中调用等待任务，但我注意到这会阻止任何进一步的 AMQP 操作完成 -> 例如，创建新的 amqp 通道会停止，并且通过 amqp 接收 rpc 请求也会停止。然而，我不确定是什么导致了这种停顿（因为回调本身就是一个协程，我希望等待不会停顿整个 aioamqp 库）。

我在下面发布了 RPC 客户端和服务器的源代码，两者都基于 aioamqp/aiocoap 示例。在服务器中，on_rpc_request是 amqp rpc 回调，而send_coap_obs_request是网络协程，当删除“ obs_tasks.append(task) ”语句时，该协程会被销毁。

客户端.py：

"""
    CoAP RPC client, based on aioamqp implementation of RPC examples from RabbitMQ tutorial
"""

import base64
import json
import uuid

import asyncio
import aioamqp


class CoAPRpcClient(object): …

示例代码：

async def test():
  """
  >>> await test()
  hello world
  """
  print("hello world")

尝试使用 doctests 运行它会导致SyntaxError: 'await' outside function.

我有多个测试文件，每个文件都有一个异步夹具，如下所示：

@pytest.fixture(scope="module")
def event_loop(request):
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="module")
async def some_fixture():
    return await make_fixture()

我正在使用 xdist 进行并行化。另外我有这个装饰器：

@toolz.curry
def throttle(limit, f):
    semaphore = asyncio.Semaphore(limit)

    @functools.wraps(f)
    async def wrapped(*args, **kwargs):
        async with semaphore:
            return await f(*args, **kwargs)

    return wrapped

我有一个函数使用它：

@throttle(10)
def f():
    ...

现在f正在从多个测试文件调用，并且我收到一个异常，告诉我无法使用不同事件循环中的信号量。

我尝试转向会话级事件循环装置：

@pytest.fixture(scope="session", autouse=True)
def event_loop(request):
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

但这只给了我：

ScopeMismatch：您尝试使用“模块”范围请求对象访问“函数”范围固定装置“event_loop”，涉及工厂

是否有可能让 xdist + 异步装置 + 信号量一起工作？

我试图实现表之间的多对多关系。当我使用反向填充时，特定用户的所有标签都必须位于标签字段中。

表已成功创建。

添加用户和标签。链接表也行。

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.util import await_only, greenlet_spawn

from sqlalchemy import Column, Table, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy.dialects.postgresql import VARCHAR, INTEGER

Base = declarative_base()

user_tag = Table('user_tag', Base.metadata,
                 Column('user_id', INTEGER, ForeignKey('users.id')),
                 Column('tag_id', INTEGER, ForeignKey('tags.id'))
                 )


class User(Base):
    __tablename__ = 'users'
    id = Column(INTEGER, primary_key=True)
    name = Column(VARCHAR(32), nullable=False, unique=True)
    tags = relationship("Tag",
                        secondary=user_tag,
                        back_populates="users")


class Tag(Base):
    __tablename__ = 'tags'
    id = Column(INTEGER, primary_key=True)
    tag = Column(VARCHAR(255), nullable=False, unique=True)
    users = …

我正在尝试在应用程序中使用 SQLAlchemy 的异步版本asyncio。但是，当尝试使用创建表时Metadata().create_all()，会出现以下错误

AttributeError：“AsyncEngine”对象没有属性“_run_ddl_visitor”

我们如何解决这个问题？谢谢！

import asyncio
from sqlalchemy import Column, String, 
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.sql.schema import MetaData
from sqlalchemy.ext.asyncio import create_async_engine


connection_url = f"postgresql+asyncpg://{user}:{password}@{host}:{port}/{dbname}"
engine = create_async_engine(connection_url)
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db_session = scoped_session(Session)

Base = declarative_base()
Base.query = db_session.query_property()

class Foo(Base):
    __tablename__ = "foo"    
    id = Column(String, primary_key=True)
    name = Column(String)

class Store:
    def __init__(self):
        super().__init__()
        self.connection = None

    async def connect(self):
        self.connection = await engine.begin() …

我正在使用 sqlalchemy + asyncpg 和“选择”急切加载。

我有与朋友有一对多关系的个人项目。

我将一个人插入到我的数据库中，但没有相关的朋友条目。如果在同一个会话中我尝试从数据库中获取该人，我可以很好地访问他们的静态（非关系）列，但无法访问关系friends。

我认为尝试访问person.friends会触发延迟加载，尽管它之前是作为selectin加载强制执行的。为什么是这样？我怎样才能避免它？

# Create the ORM model
class Person(Base):
    __tablename__ = 'items'
    id_ = Column(POSTGRES_UUID(as_uuid=True), primary_key=True)
    name = Column(String(32))
    friends = relationship('Friend', lazy='selectin')

# Create an instance
person_id = uuid4()
person = Person(id_=person_id, name='Alice') # Note that this Person's friends are not set

# Add to database
async with AsyncSession(engine, expire_on_commit=False) as session:
    try:
        session.begin()
        session.add(person)
        await session.commit()
    except:
        await session.rollback()
        raise
    # Get the added person from …

我尝试从 Binance Websocket 获取数据。使用 python 3.9 作为解释器它运行良好，但使用 3.10 它给了我错误:(

这是代码：

import asyncio
from binance import AsyncClient, BinanceSocketManager


async def main():
    client = await AsyncClient.create()
    bm = BinanceSocketManager(client)
    # start any sockets here, i.e a trade socket
    ts = bm.trade_socket('BNBBTC')
    # then start receiving messages
    async with ts as tscm:
        while True:
            res = await tscm.recv()
            print(res)


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

我收到此错误：

DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()

我使用 PyCharm 作为 IDE。

请问有人可以帮助我吗？

我有一堂有各种方法的课。我在该类中有一个类似的方法：

 class MyClass:

    async def master_method(self):
      tasks = [self.sub_method() for _ in range(10)]
      results = await asyncio.gather(*tasks)

    async def sub_method(self):
      subtasks = [self.my_task() for _ in range(10)]
      results = await asyncio.gather(*subtasks)

    async def my_task(self):
      return "task done"  

所以这里的问题是：

1. 使用asyncio.gather()从另一个程序调用的内部协同例程是否存在任何问题、优点/缺点asyncio.gather()？有性能问题吗？

2. 循环中所有级别的所有任务是否都以相同的优先级处理asyncio？这是否会提供与我使用单个 asyncio.gather() 调用所有协同例程相同的性能master_method？

我想从网站下载/抓取 5000 万条日志记录。我没有一次性下载 5000 万个，而是尝试使用以下代码一次下载 1000 万个，但它一次只能处理 20,000 个（超过这个数量会引发错误），因此它变得非常耗时下载那么多数据。目前下载20000条记录的速度需要3-4分钟，100%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88| 20000/20000 [03:48<00:00, 87.41it/s]那么如何加速呢？

import asyncio\nimport aiohttp\nimport time\nimport tqdm\nimport nest_asyncio\n\nnest_asyncio.apply()\n\n\nasync def make_numbers(numbers, _numbers):\n    for i in range(numbers, _numbers):\n        yield i\n\n\nn = 0\nq = 10000000\n\n\nasync def fetch():\n    # example\n    url = "https://httpbin.org/anything/log?id="\n\n    async with aiohttp.ClientSession() as session:\n        post_tasks = []\n        # prepare the coroutines that poat\n        async for x in make_numbers(n, q):\n            post_tasks.append(do_get(session, url, x))\n        # now execute them all at once\n\n        responses = [await f for f in tqdm.tqdm(asyncio.as_completed(post_tasks), …