标签: python-asyncio

我正在使用aiohttp实现一个 Web API ，并使用启用了 UVloop 的Gunicorn--worker-class aiohttp.GunicornUVLoopWebWorker进行部署。因此，我的代码始终在异步上下文中运行。我的想法是在处理请求时实现并行作业以获得更好的性能。

我不使用是asyncio因为我想要Parallelism，而不是Concurrency。

我知道python 中的多处理和GIL 问题。但加入流程也适用于我的问题。

这是一个例子：

from aiohttp.web import middleware

@middleware
async def context_init(request, handler):
    request.context = {}
    request.context['threads'] = []
    ret = await handler(request)
    for thread in request.context['threads']:
        thread.join()
    return ret

考虑到thread.join()orprocess.join()阻止当前线程，这将阻止事件循环（据我所知）。如何异步加入？我想要的可以形象地表示为：await thread.join()或await process.join()。

更新：

感谢@user4815162342，我能够为我的项目编写正确的代码：

中间件：

from aiohttp.web import middleware
from util.process_session import ProcessSession

@middleware
async def context_init(request, handler): …

根据asyncio 同步原语，有多种同步方法。

• 我很困惑为什么我们需要异步同步？
• 我的意思是，asyncio 是异步的。
• 在异步中添加一些同步有意义吗？

我正在尝试运行Telethon 文档提供的第一个代码片段。但是，在出现多个问题（此处和此处）之后，我最终得到了这个修改后的版本：

import os
import sys
from telethon.sync import TelegramClient, events

# import nest_asyncio
# nest_asyncio.apply()

session_name = "<session_name>"
api_id = <api_id>
api_hash = "<api_hash>"


os.chdir(sys.path[0])

if f"{session_name}.session" in os.listdir():
    os.remove(f"{session_name}.session")

async with TelegramClient(session_name, api_id, api_hash) as client:
   client.send_message('me', 'Hello, myself!')
   print(client.download_profile_photo('me'))

   @client.on(events.NewMessage(pattern='(?i).*Hello'))
   async def handler(event):
      await event.reply('Hey!')

   client.run_until_disconnected()

但是现在我收到这些警告：

usr/local/lib/python3.7/site-packages/ipykernel_launcher.py:23：RuntimeWarning：协程“MessageMethods.send_message”从未等待
RuntimeWarning：启用tracemalloc以获取对象分配回溯
/usr/local/lib/python3.7/site-packages/ipykernel_launcher.py:24：RuntimeWarning：从未等待协程“DownloadMethods.download_profile_photo”
RuntimeWarning：启用tracemalloc以获取对象分配回溯
/usr/local/lib/python3.7/site-packages/ipykernel_launcher.py:30：RuntimeWarning：从未等待协程“UpdateMethods._run_until_disconnected”
RuntimeWarning：启用tracemalloc以获取对象分配回溯

在 Jupyter 上运行代码时。现在我的问题是：

• 这些警告消息意味着什么以及我应该如何解决它们？
• 如果正常工作，这段代码的预期结果是什么？我应该通过 Telegram 或其他方式收到消息吗？因为除了登录码之外我没有收到任何消息。
• @行首的符号是什么@client.on...意思？那条线应该做什么？从这一行开始我就不明白代码了。如果您能帮助我理解它，我将不胜感激。

我们正在运行一个 Fastapi + Uvicorn Web 应用程序，使用 Gunicorn 作为进程管理器，使用 Nginx 作为反向代理服务器。对于大多数 I/O 操作（数据库调用、Rest api），应用程序都以异步模式运行。整个设置在 Ubuntu 16.04 上的 Docker 容器内运行。

该设置在大多数情况下都有效，但有时它根本不处理请求，并且在 Nginx 端超时。我们还尝试将 Nginx 从设置中删除，并观察到在很长一段时间（例如 15 分钟后）后很少有请求得到处理。这是非常随机的，但通常在一小时内发生 2-3 次。

下面是我们使用的gunicorn配置\xe2\x80\x93

host = os.getenv("HOST", "0.0.0.0")\nport = os.getenv("PORT", "80")\n\n# Gunicorn config variables\nworkers = web_concurrency\nbind = f"{host}:{port}"\nkeepalive = 2\n\ntimeout = 60  \ngraceful_timeout = 30\nthreads = 2\nworker_tmp_dir = "/dev/shm"\n# Logging mechanism\ncapture_output = True\nloglevel = os.getenv("LOG_LEVEL", "debug") \n

并用命令调用gunicornexec gunicorn -k uvicorn.workers.UvicornWorker -c "$GUNICORN_CONF" "$APP_MODULE"

我们尝试了一些配置更改，例如 \xe2\x80\x93

\n
1. 改变worker数量、worker超时
\n …

我想使用带有 Gather() 的信号量来限制 api 调用。我想我必须使用 create_task() 但我收到运行时错误：“RuntimeError：await 未与 future 一起使用”。我该如何修复它？

这是代码：

import asyncio
# pip install git+https://github.com/sammchardy/python-binance.git@00dc9a978590e79d4aa02e6c75106a3632990e8d
from binance import AsyncClient


async def catch_up_aggtrades(client, symbols):
    tasks = asyncio.create_task(get_historical_aggtrades(client, symbol) for symbol in symbols)
    sem = asyncio.Semaphore(1)
    async with sem:
        await asyncio.gather(*tasks)


async def get_historical_aggtrades(client, symbol):
    async for trade in client.aggregate_trade_iter(symbol, '1 day ago UTC'):
        print(f"symbol {symbol}")


async def main():
    client = await AsyncClient.create()
    symbols = ['BTCUSDT', 'ETHUSDT', 'BNBUSDT']
    await catch_up_aggtrades(client, symbols)


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

我们已经使用 FastAPI 创建了一个服务。当我们的服务启动时，它会创建一些 Python 对象，然后端点使用这些对象来存储或检索数据。

生产中的 FastAPI 从多个工人开始。我们的问题是每个工人创建自己的对象而不是共享一个。

下面的脚本显示了我们正在做的（简化的）示例，尽管在我们的例子中 Meta() 的使用要复杂得多。

from fastapi import FastAPI, status

class Meta:
   def __init__(self):
      self.count = 0  

app = FastAPI()

meta = Meta()

# increases the count variable in the meta object by 1
@app.get("/increment")
async def increment():
    meta.count += 1
    return status.HTTP_200_OK

# returns a json containing the current count from the meta object
@app.get("/report")
async def report():
    return {'count':meta.count}

# resets the count in the meta object to 0
@app.get("/reset") …

来自 python 的 asyncio 示例：

import asyncio
import time
def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    # Note that time.sleep() can be replaced with any blocking
    # IO-bound operation, such as file operations.
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

async def main():
    print(f"started main at {time.strftime('%X')}")

    await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1))

    print(f"finished main at {time.strftime('%X')}")


asyncio.run(main())

# Expected output:
#
# started main at 19:50:53
# start blocking_io at 19:50:53
# blocking_io complete at 19:50:54
# finished main at 19:50:54

它正在输出下一个错误：

    asyncio.to_thread(blocking_io),
    AttributeError: …

我使用 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 …