标签: python-asyncio

第一次使用 Aync 生成器。我正在使用Python 3.9。

这是我的实现：

import asyncio

class SubEventStream():
    def __init__(self) -> None:
        self.queue = asyncio.Queue()
        return

    async def __aiter__(self):
        return self

    async def __anext__(self):
        return await self.pop()

    async def append(self, request):
        return await self.queue.put(request)

    async def pop(self):
        r = await self.queue.get()
        self.queue.task_done()
        return r

def create_append_tasks(ls, q):
    return [
        asyncio.create_task(q.append(i))
        for i in ls
    ]

async def append_tasks(q):
    tasks = create_append_tasks(('a', 'b', 'c', 'd', 'e'), q)
    return await asyncio.gather(*tasks)


async def run():
    q = SubEventStream()
    await append_tasks(q)

    async for …

由于请求查询阻塞了我的所有线程，我正在将普通视图转换为异步视图。到目前为止，除了一个问题之外，我已经解决了大部分问题。如何异步保存模型？

async def dashboardAddChart(request, rowId):
    row = (await sync_to_async(list)(DashboardRow.objects.filter(pk=rowId).select_related('dashboard__site', 'dashboard__theme')))[0]

    chart = DashboardChart(dashboard=row.dashboard, dashboardRow=row)
    
    if row.dashboard.theme is not None:
        dashboardThemes.applyThemeToChart(chart)

    chart.save()

    chartData = await getChartData(chart.pk)

我尝试过很多事情，chart.save()包括：

await sync_to_async(chart.save)

t = asyncio.ensure_future(sync_to_async(chart.save))
await asyncio.gather(t)

但我没搞对。

任何帮助将不胜感激！

我使用 pytest 夹具的测试设置配置有问题：

@pytest.fixture(autouse=True)
async def run_around_tests(elastic_api, catalog):
    es_index_mapping = open('test/resources/es_index_mapping.json')
    es_index_mapping_dict = json.load(es_index_mapping)
    es_cars = open('test/resources/es_cars.json')
    es_cars_dict = json.load(es_cars)
    await elastic_api.create_index(index='catalog_test', payload=es_index_mapping_dict)
    await catalog.upsert_documents(index='catalog_test', payload=es_cars_dict)
    yield
    await elastic_api.delete_index(index='catalog_test')
    await catalog.close()

似乎yield 没有正确执行并且没有等待测试执行。在测试执行期间发生elasticsearch索引的删除，导致测试失败。为什么只有在所有测试完成后才执行此删除？

我试图了解它asyncio.create_task实际上是如何工作的。假设我有以下代码：

import asyncio
import time

async def delayer():
    await asyncio.sleep(1)

async def messenger():
    await asyncio.sleep(1)
    return "A Message"

async def main():
    message = await messenger()
    await delayer()

start_time = time.time()

asyncio.run(main())

end_time = time.time() - start_time

print(end_time)

该代码大约需要 2 秒。但是如果我对主体进行一些更改，main如下所示：

import asyncio
import time

async def delayer():
    await asyncio.sleep(1)

async def messenger():
    await asyncio.sleep(1)
    return "A Message"

async def main():
    task1 = asyncio.create_task(delayer())
    task2 = asyncio.create_task(delayer())

    await task1
    await task2

start_time = time.time()

asyncio.run(main())

end_time = …

我陷入困境，无法找到使用 Python 异步连接到 Azure SQL 数据库的可行方法。

我已经尝试过asyncio，pyodbc但asyncpg没有成功。

我想这已经很接近了...

import asyncio
import pyodbc

async def query_azure_sql_db():
  connection_string = 'Driver={ODBC Driver 17 for SQL Server};Server=tcp:<mySERVER>.database.windows.net,1433;Database=sqldbstockdata;Uid=<myUN>;Pwd=<myPW>;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;'

  async with pyodbc.connect(connection_string) as conn:
    async with conn.cursor() as cursor:
      query = 'SELECT * FROM dbo.<myTABLE>'
      await cursor.execute(query)
      results = cursor.fetchall()
      return results

loop = asyncio.get_event_loop()
results = loop.run_until_complete(query_azure_sql_db())
print(results)

但会导致这个神秘的错误：AttributeError: __aenter__

我对其他图书馆持开放态度。

任何帮助表示赞赏。

这是我的代码:

import asyncio, socket

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('', 1234))
sock.setblocking(False)

queue = asyncio.Queue()

def sock_reader():
    print(sock.recv(1024))
    # x = yield from queue

def test_sock_reader():
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.sendto(b'HELLO', ('127.0.0.1', 1234))

loop = asyncio.get_event_loop()
loop.add_reader(sock, sock_reader)
loop.call_later(0.5, test_sock_reader)
loop.run_forever()
loop.close()

这是输出:

b'HELLO'

# x = yield from queue取消注释该行时,程序不再打印b'Hello'.

为什么yield from影响应该已经执行的命令？

我在Python 3.6中为asyncio尝试了以下代码:示例1:

import asyncio
import time

async def hello():

    print('hello')
    await asyncio.sleep(1)
    print('hello again')

tasks=[hello(),hello()]    
loop=asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

输出符合预期:

hello
hello
hello again
hello again

然后我想将asyncio.sleep更改为另一个def:

async def sleep():
    time.sleep(1)

async def hello():

    print('hello')
    await sleep()
    print('hello again')


tasks=[hello(),hello()]    
loop=asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

输出:

hello
hello again
hello
hello again

它似乎不是以异步模式运行,而是以正常同步模式运行.

问题是:为什么它不是以异步模式运行？如何将旧的同步模块更改为"异步"模块？

我有一个应用程序,它将协同程序添加到已经运行的事件循环中.这些协同程序的参数依赖于I/O,并且在我最初启动事件循环时不可用 - 使用loop.run_forever(),所以我稍后添加任务.为了演示这种现象,这里有一些示例代码:

import asyncio
from threading import Thread
from time import sleep

loop = asyncio.new_event_loop()

def foo():
    loop.run_forever()

async def bar(s):
    while True:
        await asyncio.sleep(1)
        print("s")

#loop.create_task(bar("A task created before thread created & before loop started"))
t = Thread(target=foo)
t.start()
sleep(1)
loop.create_task(bar("secondary task"))

奇怪的行为是,当调用loop.run_forever()时,循环中至少有一个任务时,一切都按预期工作.即当注释行没有被注释掉时.

但是当它被注释掉时,如上所示,没有打印任何内容,看起来我无法向event_loop添加任务.我应该避免在不添加单个任务的情况下调用run_forever()吗？我不明白为什么这应该是一个问题.在运行后将任务添加到event_loop是标准的,为什么空案例会成为问题？

我是python的新手,但是有一个使用uvloop的异步python应用程序,它使用我创建的C api模块,它也需要访问异步循环.

1)asyncio还没有这个c-api？任何在C中使用事件循环的hack？这是在哪里讨论？

2)uvloop使用我在C中熟悉的libuv.如果我可以抓取uv_loop_t指针,我可以挂钩循环.我想我可以:

A)使用PyObject*到uvloop的循环计算uv_loop_t*的偏移并使用它？假设我知道PyObject_HEAD的长度？

libuv_loop = (uv_loop_t*)((void*)(loop)+0x8);

struct __pyx_obj_6uvloop_4loop_Loop {
    PyObject_HEAD
    uv_loop_t *uvloop;

B)或非hacky修改uvloop以暴露循环指针.因为我从来没有看过cython代码,所以我完全无能为力.我可以在循环上创建一个python函数,从我的C代码调用它并获取C指针吗？喜欢:

(uv_loop_t*)PyObject_CallFunctionObjArgs( getLoop, NULL )

通过在这里添加getLoop:

https://github.com/MagicStack/uvloop/blob/master/uvloop/loop.pyx

cdef uv.uv_loop_t* _getLoop(self):
    return self.uvloop

我在用aiohttp / asyncio做错了。当我尝试run_my_job()一次循环运行另一个文件时，如果我一次性运行它，则下面的代码可以正常工作：

main.py
========================================
 count = 0
    batch_count = math.ceil((abc.get_count()/100))
    print("there are {} batches to complete.".format(batch_count))
    while count < batch_count:
        print("starting batch {}...".format(count))
        abc.run_my_job()
        print("batch {} completed...".format(count))
        count += 1


abc.py
===============================
def run_my_job(self):
    self.queue_manager(self.do_stuff(all_the_tasks))

def queue_manager(self, method):
    print('starting event queue')
    loop = asyncio.get_event_loop()
    future = asyncio.ensure_future(method)
    loop.run_until_complete(future)
    loop.close()

async def async_post(self, resource, session, data):
    async with session.post(self.api_attr.api_endpoint + resource, headers=self.headers, data=data) as response:
        resp = await response.read()
    return resp

async def do_stuff(self, data):
    print('queueing tasks')

    tasks …