标签: python-asyncio

我想在 heroku 中创建一个应用程序，它将一些数据作为 websocket 服务器并将其作为不和谐机器人发送。

这就是我尝试实现这个想法的方式：

# -*- coding: utf-8 -*-  
import asyncio, websockets, discord   

client = discord.Client() 
channel = client.get_channel(...)
print(channel)

async def response(websocket, path): 
    global channel 
    r = await websocket.recv()
    print(r)  
    await channel.send(str(r))

PORT = os.environ.get('PORT')
start_server = websockets.serve(response, '0.0.0.0', os.environ.get('PORT'))
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

但我不知道如何在这样的代码中启动一个不和谐的机器人，以便它与 websocket 服务器并行工作。请告诉我该怎么做。

我想知道如何在以下 get 调用中传递标头

headers = {
'User-Agent': 'Mozilla'
}
async def fetch(url, session):
async with session.get(url) as response:
    resp = await response.read()
    return resp

我尝试了以下但没有得到任何回应。

headers = {
'User-Agent': 'Mozilla'
}
async def fetch(url, session):
async with session.get(url, headers=headers) as response:
    resp = await response.read()
    return resp

目的是以异步模式调用不同的 url。需要知道是否还有其他替代方法，但无论如何，都需要传递标头才能获得正确的响应。

我正在使用 uvicorn，我需要使用现有的事件循环。我正在使用以下命令：

loop = asyncio.get_event_loop()

但是当我使用这一行时，代码就会卡住。但如果我使用new_event_loop，我会收到以下错误：

RuntimeError: Task <Task pending coro=<open_connection() running at /usr/lib/python3.6/asyncio/streams.py:81> cb=[_release_waiter(<Future pendi...sk._wakeup()]>)() at /usr/lib/python3.6/asyncio/tasks.py:316]> got Future <Future pending> attached to a different loop

请帮助我了解如何使用 uvicorn 创建的现有循环。

因此，考虑到一些复杂的设置，它用于生成要半并行运行的查询列表（使用信号量不要同时运行太多查询，以免对服务器进行 DDoS）。

我有一个（本身是异步的）函数，它创建许多查询：

async def run_query(self, url):
    async with self.semaphore:
        return await some_http_lib(url)

async def create_queries(self, base_url):
   # ...gathering logic is ofc a bit more complex in the real setting
   urls = await some_http_lib(base_url).json()
 
   coros = [self.run_query(url) for url in urls]  # note: not executed just yet
   return coros

async def execute_queries(self):
   queries = await self.create_queries('/seom/url')
   _logger.info(f'prepared {len(queries)} queries')
   results = []
   done = 0

   # note: ofc, in this simple example call these would not actually be asynchronously executed. …

在 python 中使用 asyncio 时，我们如何确定 read() 的最佳参数？12字节？100 字节？

async with self._session.get(url, headers=headers) as response:
    chunk_size = 12
    result = ''

    while True:
       chunk = await response.content.read(chunk_size)
          if not chunk:
              break
          elif isinstance(chunk, (bytes, bytearray)):
              data = chunk.decode('utf8')
               result += data

我需要异步运行 20 个任务（每个任务运行相同的函数，但具有不同的参数）。每个任务都使用Python 的yfinanceAPI 模块。这是我当前的方法：

1. 定义一个包含 20 个元素的列表args；每个元素都是要传递给相应任务的参数。
2. 定义一个异步函数get_data，我将运行该函数 20 次，每次使用不同的参数。
3. 定义一个异步函数main，用于asyncio.gather异步运行 20 个任务。

这是（伪）代码：

import asyncio

stocks = []
args = ['arg1', 'arg2', ... , 'arg20']


async def get_data(arg):
    stock = Stock(arg)
    # do some yfinance calls
    return stock


async def main():
    global stocks
    tasks = [asyncio.ensure_future(get_data(arg)) for arg in args]
    stocks = await asyncio.gather(*tasks)


asyncio.run(main())

print(stocks)  # should be a list of 20 return values from the 20 …

所以我了解如何通过列表创建异步任务集合并使用 asyncio.gather() 来执行它们，但我不知道如何对字典执行相同操作：

    from fastapi import FastAPI
    import asyncio
    import httpx

    app = FastAPI()

    urls = [
        "http://www.google.com",
        "https://www.example.org",
        "https://stackoverflow.com/",
        "https://www.wikipedio.org"
    ]

    async def async_request_return_dict(url: str):
        async with httpx.AsyncClient() as client:
            r = await client.get(url)
        return {url : r.status_code}

    # List of async coroutines
    @app.get("/async_list")
    async def async_list():

        tasks = []
        for url in urls:
            tasks.append(async_request_return_dict(url=url))
        response = await asyncio.gather(*tasks)

        # Convert list of dict -> dict
        dict_response = dict()
        for url in response:
            dict_response.update(url)

        return dict_response

    async def async_request_return_status(url: …

我设置了以下测试程序（Python 3.9.5，numpy 1.20.2）：

import asyncio
from datetime import datetime
import numpy as np

async def calculate():
    print("=== unique")
    await asyncio.to_thread(lambda: np.unique(np.ones((2000, 50000)), axis=0))
    print("=== sort")
    await asyncio.to_thread(lambda: np.sort(np.ones((2000, 50000)), axis=0))
    print("=== cumsum")
    await asyncio.to_thread(lambda: np.cumsum(np.ones((2000, 100000)), axis=0))

async def ping():
    while True:
        print("async", datetime.utcnow())
        await asyncio.sleep(0.2)

async def main():
    p1 = asyncio.create_task(ping())
    c = asyncio.create_task(calculate())
    await asyncio.wait([p1, c], return_when=asyncio.FIRST_COMPLETED)
    p1.cancel()

asyncio.run(main())

输出如下：

async 2021-05-21 13:20:16.308948
=== unique
async 2021-05-21 13:20:16.531135
async 2021-05-21 13:20:40.142323
=== sort
async 2021-05-21 13:20:40.343306
async 2021-05-21 …

我对异步编程非常陌生，我正在尝试使用 httpx。我有以下代码，我确信我做错了什么 - 只是不知道它是什么。有两种方法，一种是同步，另一种是异步。它们都来自谷歌金融。在我的系统上，我看到花费的时间如下：

异步：5.015218734741211
同步：5.173618316650391

这是代码：

import httpx
import asyncio
import time



#
#--------------------------------------------------------------------
#
#--------------------------------------------------------------------
#
def sync_pull(url):
  r = httpx.get(url)
  print(r.status_code)


#
#--------------------------------------------------------------------
#
#--------------------------------------------------------------------
#
async def async_pull(url):
  async with httpx.AsyncClient() as client:
    r = await client.get(url)
    print(r.status_code)


#
#--------------------------------------------------------------------
#
#--------------------------------------------------------------------
#
if __name__ == "__main__":

  goog_fin_nyse_url = 'https://www.google.com/finance/quote/'
  tickers = ['F', 'TWTR', 'CVX', 'VZ', 'GME', 'GM', 'PG', 'AAL', 
             'MARK', 'AAP', 'THO', 'NGD', 'ZSAN', 'SEAC',
             ]  

  print("Running asynchronously...")
  async_start = time.time()
  for ticker in tickers: …

我正在摆弄 sqlalchemy ORM 功能

我能够使其在我的主应用程序上运行，但是当我创建一个单独的文件test.py来测试某些内容时，我不断收到事件循环关闭错误：

Exception ignored in: <function Connection.__del__ at 0x7f7041c07310>
Traceback (most recent call last):
  File "/home/krypt/Documents/Projects/app/env/lib/python3.9/site-packages/aiomysql/connection.py", line 1072, in __del__
  File "/home/krypt/Documents/Projects/app/env/lib/python3.9/site-packages/aiomysql/connection.py", line 298, in close
  File "/usr/lib/python3.9/asyncio/selector_events.py", line 700, in close
  File "/usr/lib/python3.9/asyncio/base_events.py", line 746, in call_soon
  File "/usr/lib/python3.9/asyncio/base_events.py", line 510, in _check_closed
RuntimeError: Event loop is closed

这是代码test.py：

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import Table, Column, Integer, String …