标签: python-asyncio

我有一个基于Python 3.5+异步框架(apistar,sanic等)构建的Web应用程序.该应用程序进行各种IO调用 - 数据库,Redis等 - 也是异步的.

有些文档建议使用其他事件循环:

import asyncio
import peewee
from peewee_async import Manager, PostgresqlDatabase

loop = asyncio.new_event_loop() # Note: custom loop!
database = PostgresqlDatabase('test')
objects = Manager(database, loop=loop)

我的理解是await语句允许事件循环在上传IO时进行上下文切换,因此额外的事件循环似乎完全没必要.

使用额外的事件循环有什么好处,何时应该使用其他循环？

我有一个test.py文件和一个AsyncioCurl.py文件。
我已经使用session而不仅仅是aiohttp.request
但它也给了我这个错误：

Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x000001FAEFEA7DA0>
Unclosed connector
connections: ['[(<aiohttp.client_proto.ResponseHandler object at 0x000001FAF10AC648>, 119890.906)]']
connector: <aiohttp.connector.TCPConnector object at 0x000001FAF0F702B0>

测试文件

import asyncio
from AsyncioCurl import AsyncioCurl
async def a():
    payload = {}
    url = "https://awebsiteisthere.com"
    data = await AsyncioCurl().get(url,payload)
    print(data)

task = [
    a()
]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(task))

AsyncioCurl.py

import asyncio
import aiohttp
from Log import Log
from Base import sign
from config import config


class AsyncioCurl: …

任何 asyncio 任务在内存和速度方面的开销是多少？在不需要并发运行的情况下最小化任务数量是否值得？

测试以下代码时

@pytest.mark.asynico
async def test_handle_DATA(mocker):
    handle_mock = mocker.MagicMock()
    envelope_mock = mocker.MagicMock(mail_from="Test@From", rcpt_tos=["Test@To"], content=b"TestContent")

    result = SendToDictHandler.handle_DATA(handle_mock, "TestServer", "TestSession", envelope_mock)

    assert result == "250 Message accepted for delivery"
    assert email_core.testing_emails_dict == {
        "Test@To": {
            "from": "Test@From",
            "to": ["Test@To"],
            "msg": "TestContent",
        }
    }

pytest -vvv在项目环境中运行时收到的警告：

PytestWarning: Coroutine functions are not natively supported and have been skipped.
You need to install a suitable plugin for your async framework, for example:
 - pytest-asyncio
 - pytest-trio
 - pytest-tornasync
warnings.warn(PytestWarning(msg.format(pyfuncitem.nodeid)))

我确实pytest-asyncio安装了。我通过pytest --trace-config …

我想使用 asyncio (/curio/trio) 和 pytest 同时运行多个测试，但我找不到任何相关信息。我需要自己安排吗？如果我这样做，有没有办法有一个很好的输出来分离（子）测试用例？

这是我正在尝试的一个小玩具示例：

import pytest
import time
import asyncio

pytestmark = pytest.mark.asyncio
expected_duration = 1
accepted_error = 0.1

async def test_sleep():
    start = time.time()
    time.sleep(expected_duration)
    duration = time.time() - start
    assert abs(duration-expected_duration) < accepted_error

async def test_async_sleep():
    start = time.time()
    await asyncio.sleep(expected_duration)
    duration = time.time() - start
    assert abs(duration-expected_duration) < accepted_error

I'm trying to make an asynchronous web scraper using beautifulsoup and aiohttp.This is my initial code to start things.I'm getting a [TypeError: An asyncio.Future, a coroutine or an awaitable is required] and having a hard time figuring out what is wrong with my code.I am new to python and would appreciate any help regarding this.

import bs4
import asyncio
import aiohttp


async def parse(page):
    soup=bs4.BeautifulSoup(page,'html.parser')
    soup.prettify()
    print(soup.title)



async def request():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://google.com") as resp: …

I don't understand the difference between websocket (websocket client) and websockets. I would like to understand the difference to know which one uses to be optimal.

I have a code that seems to do the same thing.

import websockets
import asyncio
import json

async def capture_data():
    uri = "wss://www.bitmex.com/realtime?subscribe=instrument:XBTUSD"
    #uri = "wss://www.bitmex.com/realtime"
    async with websockets.connect(uri) as websocket:
        while True:
            data = await websocket.recv()
            data = json.loads(data)
            try :
                #print('\n', data['data'][0]["lastPrice"])
                print('\n', data)
            except KeyError:
                continue


asyncio.get_event_loop().run_until_complete(capture_data())

import websocket
import json …

我使用 aiohttp、pytest 和 pytest-cov 来覆盖我的代码。我想要获得更好的覆盖率<但现在我有点卡住了，因为事件简单代码没有显示 100% cov。比如这段代码：

@session_decorator()
async def healthcheck(session, request):
    await session.scalar("select pg_is_in_recovery();")
    return web.json_response({"status": "ok"})

在覆盖率报告中，它显示返回的行未被覆盖。

我已阅读此链接异步方法的代码覆盖率。但这是针对 C# 的，我无法理解：

如果您正在等待的操作在等待之前已完成，则最常发生这种情况。

我的运行测试的代码：

python3.9 -m pytest -vv --cov --cov-report xml --cov-report term-missing

我对上面代码的测试

async def test_healthz_page(test_client):
    response = await test_client.get("/healthz")
    assert response.status == HTTPStatus.OK

asyncio.run 的文档指出：

该函数总是创建一个新的事件循环并在最后关闭它。它应该用作 asyncio 程序的主要入口点，并且最好只调用一次。

但它没有说为什么。我有一个非异步程序需要调用异步内容。我可以asyncio.run在每次到达异步部分时使用，还是这是不安全/错误的？

就我而言，我有几个async想要的协程gather并并行运行以完成。当它们全部完成后，我想继续我的同步代码。

async my_task(url):
    # request some urls or whatever

integration_tasks = [my_task(url1), my_task(url2)]

async def gather_tasks(*integration_tasks):
    return await asyncio.gather(*integration_tasks)


def complete_integrations(*integration_tasks):
    return asyncio.run(gather_tasks(*integration_tasks))

print(complete_integrations(*integration_tasks))

我最近将使用 FastApi 编码的 REST API 迁移到新的 SQLAlchemy 1.4+ 异步版本。我的应用程序编译正确，数据库似乎设置得很好。当我尝试执行多个请求时，会出现问题，出现错误，似乎表明同一会话正在用于我的所有请求。我把错误信息放在最后

这是我的代码，我基于 SQLAlchemy 异步文档和此示例

应用程序引擎初始化

from typing import AsyncIterator
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

from .notification import Notification
from .devices import Device



from sqlalchemy import MetaData
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import create_engine

from app.core.config import Config
from app.core.constants import logger
import asyncio

engine = create_async_engine(Config.RDS_DB_URL)
metadata = MetaData(engine)

#if not engine.dialect.has_table(engine, C.NOTIFICATIONS):
#Base.metadata.create_all(engine)

async def init_connection():
    async with engine.begin() as conn:
        await …