标签: python-asyncio

我目前有两个无限异步任务正在运行，并且希望在它们之间共享状态。一项任务是读取消息然后发送消息的 WebSocket 连接，另一项任务读取传入的光数据。我想在两个任务之间发送一个布尔值，表明 websocket 连接是否成功。

这是我初始化上下文的方式。

client_connect_var = contextvars.ContextVar('client_connect',default = False)
client_connect_var.set(False)
ctx = contextvars.copy_context()
async def main():
  message = json.dumps({'payload': {
                        'payload'})
  loop = asyncio.get_event_loop()
  start_light = asyncio.create_task(calculate_idle(3))
  await asyncio.gather(init_connection(message), start_light)

ctx.run(asyncio.run(main()))

这是我的 init_connection 中的代码：

async def init_connection(message):
  async with websockets.connect(uri) as websocket:
      #This should set the global context variable to true
      client_connect_var.set(True)
      CLIENT_WS = websocket
      client_connect = client_connect_var.get()
    # send init message
    await websocket.send(message)
    print("Connection is open")
    while client_connect:
        await handleMessages(websocket, message)
    await websocket.close()

这是它试图获取灯光代码中当前状态的地方

async def calculate_idle(t): …

我想从 dukaskopy 下载许多文件。典型的 url 如下所示。

url = 'http://datafeed.dukascopy.com/datafeed/AUDUSD/2014/01/02/00h_ticks.bi5'

我在这里尝试了答案，但大多数文件的大小都是 0。

但是当我简单地使用 wget 循环时（见下文），我得到了完整的文件。

import wget
from urllib.error import HTTPError

pair = 'AUDUSD'
for year in range(2014,2015):
    for month in range(1,13):
        for day in range(1,32):
            for hour in range(24): 
                try:
                    url = 'http://datafeed.dukascopy.com/datafeed/' + pair + '/' + str(year) + '/' + str(month-1).zfill(2) + '/' + str(day).zfill(2) + '/' + str(hour).zfill(2) + 'h_ticks.bi5'
                    filename = pair + '-' + str(year) + '-' + str(month-1).zfill(2) + '-' + str(day).zfill(2) + '-' …

我打算在 PyQt 应用程序中嵌入 aiohttp 服务器，但是当我运行下面的代码时，Qt 窗口无法显示，我知道这是由 引起的web.run_app(app)，我尝试将其移动到线程中，但是然后我明白了RuntimeError: There is no current event loop in thread \'Dummy-1\'，那我该怎么办？我发现asyncqt可能会有所帮助，但我不知道如何使用它来处理 aiohttp 服务器。

from PyQt5.QtCore import *\nfrom PyQt5.QtGui import *\nfrom PyQt5.QtWidgets import *\nfrom aiohttp import web\n\n\nclass ThreadGo(QThread):  # threading.Thread\n    # implementing new slots in a QThread subclass is error-prone and discouraged.\n\n    def __init__(self, parent, func, *args, **kwargs):\n        super().__init__(parent)\n        self.func = func\n        self.args = args\n        self.kwargs = kwargs\n        self.result = 0\n\n        onFinished = self.kwargs.get(\'onFinished\')\n        self.finished.connect(onFinished) if onFinished else None …

我正在开发一个从很多网站解析数据的项目。我的大部分代码都已完成，因此我期待使用 asyncio 来消除 I/O 等待，但我仍然想测试线程如何工作，无论好坏。为此，我编写了一些简单的代码来向 100 个网站发出请求。顺便说一句，我正在使用requests_html库，幸运的是它也支持异步请求。

asyncio代码如下：

import requests
import time

from requests_html import AsyncHTMLSession

aio_session = AsyncHTMLSession()
urls = [...] # 100 urls


async def fetch(url):
    try:
        response = await aio_session.get(url, timeout=5)
        status = 200
    except requests.exceptions.ConnectionError:
        status = 404
    except requests.exceptions.ReadTimeout:
        status = 408
   
    if status == 200:
        return {
            'url': url,
            'status': status,
            'html': response.html
        }

    return {
        'url': url,
        'status': status
    }


    
def extract_html(urls):
    tasks = []

    for url in urls:
        tasks.append(lambda …

如果我使用（或 with ，其行为方式相同）安排回调loop.call_later()loop.call_at()，是否可以取消其执行？

为什么？假设我安排某件事从现在起一分钟后运行。但是，由于某些情况，代码决定中止该执行（因为不再需要它），或者决定重新安排另一个时间。问题是如何使用Python来实现它asyncio？

如果您熟悉 JavaScript，我正在寻找setTimeout()和的等效项clearTimeout()。

我尝试了下面的示例（来自此页面）：

nc = NATS()

await nc.connect(servers=["nats://demo.nats.io:4222"])

future = asyncio.Future()

async def cb(msg):
  nonlocal future
  future.set_result(msg)

await nc.subscribe("updates", cb=cb)
await nc.publish("updates", b'All is Well')
await nc.flush()

# Wait for message to come in
msg = await asyncio.wait_for(future, 1)

但这似乎只对接收一条消息有用。我如何订阅并继续接收消息？

我也看过包 example，但它似乎只是玩双方的对话，然后退出。

我正在用 Python 制作一个 Discord 机器人。我想要做的就是等待两个协程，一旦第一个协程完成，就运行其余的代码，但取消剩余的协程（与这里不同，我只有一个协程并希望在特定时间后取消它） .)

这是最小的可重现示例：

import asyncio
import discord

client = discord.Client()

async def testfunc():
    await asyncio.sleep(10)
    print("10 seconds.")

async def testfunc2():
    await asyncio.sleep(15)
    print("15 seconds.")

@client.event
async def on_ready():
    print("Connected!")
    await asyncio.wait([testfunc(),testfunc2()],return_when=asyncio.FIRST_COMPLETED)

client.run('MY BOT TOKEN')

我打算在这个最小的示例中打印“10 秒”。10 秒后，不再打印任何其他内容。然而，实际发生的情况是它还打印了“15 秒”。15 秒后，我想取消该协程，这样就不会发生这种情况。

在这种情况下，我似乎无法使用 asyncio.wait_for，因为有两个协程而不是一个，两个协程必须同时运行，直到一个协程完成。由于 asyncio.wait 在协程完成后不会取消协程，有什么方法可以取消它们吗？

我试图在 Spyder 中创建代码来启动我的不和谐机器人，但遇到以下错误：

回溯（最近一次调用最后一次）：

文件“”，第 1 行，在 runfile('C:/Users/Nathan/Desktop/HW/Python stuff/my_discord_bot.py', wdir='C:/Users/Nathan/Desktop/HW/Python stuff')

文件“C:\ProgramData\Anaconda3\lib\site-packages\spyder_kernels\customize\spydercustomize.py”，第 827 行，在运行文件 execfile(文件名，命名空间)中

文件“C:\ProgramData\Anaconda3\lib\site-packages\spyder_kernels\customize\spydercustomize.py”，第 110 行，在 execfile exec(compile(f.read()，文件名，'exec')，命名空间)

文件“C:/Users/Nathan/Desktop/HW/Python stuff/my_discord_bot.py”，第 17 行，在 client.run('我的应用程序令牌')

文件“C:\ProgramData\Anaconda3\lib\site-packages\discord\client.py”，第 705 行，在 run _cleanup_loop(loop) 中

文件“C:\ProgramData\Anaconda3\lib\site-packages\discord\client.py”，第 98 行，在 _cleanup_loop Loop.close() 中

文件“C:\ProgramData\Anaconda3\lib\asyncio\selector_events.py”，第 83 行，在 close raise RuntimeError(“无法关闭正在运行的事件循环”)

RuntimeError：无法关闭正在运行的事件循环

我在网上搜索过，有人告诉我使用一个名为“nest_asyncio”的包。我 pip 安装了该包，导入它，并在控制台中运行 Nest_asyncio.apply() ，但再次运行后代码仍然给出相同的错误。

这是我的代码：

import discord
from discord.ext import commands

client = commands.Bot(command_prefix = '.')

@client.event
async def on_ready():
    print ('Bot is ready.')
    
client.run('my application token')

顺便说一句，我用“我的应用程序令牌”替换了我的实际令牌，因为我不想与任何人共享它。

请帮忙。如何消除此错误并使其正常运行？

我有一个永远持续运行的脚本（它检查文件中的更改）。每当制作奇怪的文件时，我都需要发送 Discord 消息。

• 问题是，事件监视函数（def run(self):如下）来自子类，所以我无法将其更改为async def run(self):. 因此我不能使用await channel.send()
• 我的解决方案是像这里解释的那样使用run_coroutine_threadsafe： https: //stackoverflow.com/a/53726266/9283107。效果很好！但问题是，消息被放入队列中，并且在该脚本完成之前永远不会发送（在我的情况下是：永远不会）。我假设发送消息函数被放入该脚本所在的线程中，因此线程永远不会到达它们？

也许我们可以将其放入run_coroutine_threadsafe一个单独的线程或其他东西中？这是我能做的最小的例子，它仍然显示了我的子类问题。

import discord
import os
import asyncio
import time

# CHANNEL_ID = 7659170174????????
client = discord.Client()
channel = None

class Example():
    # Imagine this run comes from a subclass, so you can't add sync to it!
    def run(self):
        # await channel.send('Test') # We can't do this because of the above comment
        asyncio.run_coroutine_threadsafe(channel.send('Test'), _loop)
        print('Message sent')

@client.event
async def …

如果一个或多个工作人员同时调用“同步功能”会发生什么？也许一名或多名工人会被封锁一段时间？

async def worker(queue):
    while True:
        queue_out  = await queue.get()
        file_name = queue_out.file.name
        # Create path + file_name 
        destination_path = create_path(file_name) #<-- SYNC function

        await download_medical(queue_out,destination_path)

async def main():
    queue_in = asyncio.Queue(1)
    workers = [asyncio.create_task(worker(queue_in)) for _ in range(5)]
        
    async for result in get_result(building):
        await queue_in.put(result) 

def create_path(file_name):
    #....#
    #operations related to file and folder on the hdd
    #creates a folder based on file name