python-asyncio 相关问题

如何在fast api中异步运行数据处理任务？

我的快速 api 应用程序中有一条路线（下面的示例代码）。我的根路由调用一个名为 read_root 的异步函数。在这个函数中，我想查询数据库，获取一些结果并上传......

如何模拟aiohttp.client.ClientSession.get异步上下文管理器

我在模拟 aiohttp.client.ClientSession.get 上下文管理器时遇到一些麻烦。我找到了一些文章，这里有一个似乎有效的示例：文章 1 所以我想测试的代码：

Python3.6 AttributeError：模块“asyncio”没有属性“run”

我尝试阅读https://hackernoon.com/asynchronous-python-45df84b82434。 这是关于异步 python 的，我尝试了其中的代码，但出现了一个奇怪的错误。 代码是： ` 导入异步

等待取消异步任务的正确方法是什么？

取消的文档听起来像是您通常应该传播 CancelledError 异常： 因此，与 Future.cancel() 不同，Task.cancel() 并不能保证 Task 会被取消，

在 Google Cloud Python SDK 中使用 AsyncClients 时，future 附加到不同的循环

我对在 Python 中使用 asyncio 还很陌生。本来我有一个同步功能： 从 google.cloud 导入 texttospeech 客户端 = texttospeech.TextToSpeechClient() def 说话(*args): # 省略

在 Quart/asyncio 中调度周期性函数调用

我需要在 python 中安排一个定期函数调用（即每分钟调用一次），而不阻塞事件循环（我使用带有 asyncio 的 Quart 框架）。 本质上需要将工作提交到电子...

属性错误：模块“asyncio”没有属性“create_task”

我正在尝试调用 asyncio.create_task() 但我正在处理一个错误： 这是一个例子： 导入异步 导入时间 async def async_say(延迟, 消息): 等待 asyncio.sleep（延迟） 打印（消息）

Python asyncio - 增加Semaphore的值

我正在我的一个项目中使用aiohttp，并希望限制每秒发出的请求数量。我正在使用 asyncio.Semaphore 来做到这一点。我的挑战是我可能想增加/减少...

在后台创建对象而不使用任务队列系统的最佳方法？

我一直在思考在后台创建记录而不使用 django 信号或 celery 等的最佳方法。 我不想使用信号，因为发送者模型将...

异步和非异步上下文之间通信的架构/设计模式

我正在编写一个基于 TCP 的客户端，它既可以进行请求/代表模式通信，也可以进行推/拉模式通信。我尝试使用 asyncio 及其传输/协议结构来处理低级

asyncio 如何链接协程

我有以下测试代码，我试图将不同的协程链接在一起。 我的想法是，我想要一个下载数据的协程，并且一旦下载了数据，我就想要......

使用 Aiohttp 和代理

我正在尝试使用异步从 URL 列表（由 ids 标识）中获取 HTML。我需要使用代理。 我正在尝试将 aiohttp 与代理一起使用，如下所示： 导入异步 导入 aiohttp 来自 BS4

aiohttp.client_exceptions.ClientConnectorError：无法连接到主机：443 ssl：default [信号量超时期限已过期]

我在代码中使用了 asyncio 和 aiohttp，它在我的工作笔记本电脑上运行得非常好。然而，当我从工作笔记本电脑上在虚拟机上运行它时，我遇到了这个错误“信号量超时

运行 asyncio 循环和 tkinter gui

如何运行 asyncio 循环，同时运行 while 循环以进行 websocket 接收和 tkinter gui？ 我当前的代码：（没有GUI） 我知道如何为 tkinter gui 编写代码，但正在获取...

pytest 夹具中的异步 SQLAlchemy：greenlet_spawn 尚未被调用

所以我们假设我尝试检索所有用户进行 Pytest 测试作为固定装置： @pytest_asyncio.fixture（范围=“函数”） 异步 def test_users(db_session): 与 db_session.begin 异步...

使用 asyncio 库在 python 中实现异步后台任务

我希望在后台运行繁重的计算任务而不阻塞 IO。这里的问题是我的主要功能不依赖于繁重的任务，并且需要在执行之前/同时返回值......

为什么我会收到重复收集的 pytest anyio 测试

上下文 我正在使用 pytest 和 asyncio 为 MongoDB + FastAPI 后端编写测试。我在设置时遇到了许多不同的问题，例如事件循环关闭、任务待处理 背景 我正在使用 pytest 和 asyncio 为 MongoDB + FastAPI 后端编写测试。我在设置时遇到了许多不同的问题，例如Event Loop Close、Task Pending <name=和AttributeError: 'async_generator' object has no attribute 'post'。但我最终得到了这个有效的设置。所以这是一个最小的可重复示例。我很抱歉这么长，但我认为有必要了解整个背景 项目设置 main.py： from fastapi import FastAPI from contextlib import asynccontextmanager from app.database import close_mongo_connection, connect_to_mongo from app.routes.auth_routes import router as auth_router app = FastAPI() @asynccontextmanager async def lifespan(app: FastAPI): print("Connecting to MongoDB") await connect_to_mongo() yield print("Closing connection to MongoDB") await close_mongo_connection() app.router.lifespan_context = lifespan app.include_router(auth_router, prefix="/auth", tags=["auth"]) database.py： import os from motor.motor_asyncio import AsyncIOMotorClient from app.config import DATABASE_NAME, MONGO_DETAILS, TEST_DATABASE_NAME client: AsyncIOMotorClient = None db = None async def connect_to_mongo(): global client, db client = AsyncIOMotorClient(MONGO_DETAILS) if os.getenv("IS_TESTING") == "True": db = client[TEST_DATABASE_NAME] else: db = client[DATABASE_NAME] async def close_mongo_connection(): if os.getenv("IS_TESTING") == "True": await client.drop_database(TEST_DATABASE_NAME) client.close() print("Disconnected from MongoDB") async def get_database(): if db is None: await connect_to_mongo() return db routes/auth_routes： from app.database import get_database from fastapi import APIRouter router = APIRouter() async def create_user(user: UserCreate, db) -> UserModel: users_collection = db["users"] user_dict = user.model_dump() user_dict["hashed_password"] = get_password_hash(user_dict.pop("password")) return await users_collection.insert_one(user_dict) @router.post("/register", response_model=UserModel) async def register_user(user: UserCreate, db=Depends(get_database)): print("GET DB", db) db_user = await create_user(user, db) return db_user （相关）安装的库： anyio 4.4.0 httpcore 1.0.5 httptools 0.6.1 httpx 0.27.0 motor 3.4.0 pip 24.0 pytest 8.2.2 pytest-asyncio 0.23.7 python-dotenv 1.0.1 测试设置 tests/conftest.py： import pytest from app.database import close_mongo_connection, connect_to_mongo @pytest.fixture(scope="module", autouse=True) async def setup_and_teardown(): await connect_to_mongo() print("This function does not run, but the testing setup does not work without it") yield await close_mongo_connection() tests/test_auth.py： import pytest from httpx import AsyncClient, ASGITransport from app.main import app from app.database import connect_to_mongo, close_mongo_connection @pytest.fixture(scope="module", autouse=True) async def setup_and_teardown(): await connect_to_mongo() print("this runs") yield await close_mongo_connection() client = AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") @pytest.mark.anyio async def test_register_user(): user_data = { "email": "[email protected]", "password": "password123", } response = await client.post("/auth/register", json=user_data) 问题 功能良好，测试行为正确。但每当我输入 pytest 命令时，都会收集双倍的测试数量。例如，在我提供的示例设置中，pytest 将收集 2 个测试，并且它们都会执行（可以通过创建 2 个用户来验证）。这适用于多个文件中的所有测试。在我的完整设置中，我有 6 个测试，但它始终收集并执行 12 个测试。为什么会发生这种情况以及如何解决它。 我想这会解决你的问题： @pytest.fixture def anyio_backend(): return 'asyncio' 我从官方文档中获取它， 通过这种方式，AnyIO 仅使用 asyncio 进行测试。

如何使用Python asyncio lib高效处理嵌套异步操作？

我正在从事具有下一个异步操作的Python项目。我想确保每个级别都有良好的错误处理并返回主循环以重复执行。我想设置为，我在那里...

如何通过 Python Telethon 让一个机器人持续工作，而另一个机器人 10 秒后工作？

我有 2 个电报机器人，它们应该在聊天中收到新消息。一开始，当一条消息出现时，两个机器人都必须记住该消息，然后第一个机器人会不断记住该消息......

discord.py（带 cogs）+ FastAPI：errno -3 名称解析暂时失败 [发布时已解决]

我已经完成了在 FastAPI 下启动 Discord.py 的相当基本的尝试。 它只是因名称解析中的 errno -3 临时失败而崩溃 我已经包含了大部分相关的鳕鱼......