标签: python-asyncio

我正在尝试使用 asyncio 读取多个文件（CSV），但我不想在执行此操作时阻止主事件循环。

所以我检查了 AIOfile，它似乎承诺读取不会阻塞。虽然这可能是真的，但以下代码片段需要花费大量时间才能完成，但它与此处的示例基本相同https://github.com/mosquito/aiofile#read-file-line-by-line

import asyncio
from aiofile import AIOFile, LineReader
from pathlib import Path
import time

counter = 0

async def main():
    path = 'test_data'
    global counter
    data_dir = Path(path)
    files_in_basepath = (entry for entry in data_dir.iterdir() if entry.is_file())
    list_of_files = [(path + '/' + file.name, file) for file in files_in_basepath]
    for file in list_of_files:
        line_count = 0
        async with AIOFile(file[0]) as afp:
            await afp.fsync()
            async for line in LineReader(afp):
                #print(line)
                values = ''
                line_values = line.split(',') …

1.我有一个函数var。我想知道通过利用系统拥有的所有处理器、内核、线程和 RAM 内存进行多处理/并行处理来快速运行此函数中的循环的最佳方法。

import numpy
from pysheds.grid import Grid

xs = 82.1206, 72.4542, 65.0431, 83.8056, 35.6744
ys = 25.2111, 17.9458, 13.8844, 10.0833, 24.8306

a = r'/home/test/image1.tif'
b = r'/home/test/image2.tif'

def var(interest):
    
    variable_avg = []
    for (x,y) in zip(xs,ys):
        grid = Grid.from_raster(interest, data_name='map')

        grid.catchment(data='map', x=x, y=y, out_name='catch')

        variable = grid.view('catch', nodata=np.nan)
        variable = numpy.array(variable)
        variablemean = (variable).mean()
        variable_avg.append(variablemean)
    return(variable_avg)

2.var如果我可以针对给定的函数多个参数并行运行函数和循环，那就太好了。var(a)例如：var(b)同时。因为它比单独并行化循环消耗的时间要少得多。

如果没有意义，请忽略 2。

我想threading.Lock()在异步函数中使用，asyncio.Lock()不是线程安全的，所以我不能这样做with await asyncio.Lock():。我需要使用的原因threading.Lock()是因为这个对象可能被多个线程访问，因此它在网络应用程序中使用，并且运行它的服务器可以启动许多线程。这样做的有效方法是什么？到目前为止，我已经尝试了一个使用锁的简单函数：

1）

async def main():
    with await threading.Lock():
        a = 6
    return a

TypeError: object _thread.lock can't be used in 'await' expression

async def main():
    async with threading.Lock():
            a = 1564666546
    return a

AttributeError: __aexit__

我正在使用 AIOHTTP 开发 API 服务，尝试集成一些异步 ORM，第一个候选者是 Tortoise-ORM。在 Django 项目中，我有很多链接模型，其__str__方法如下：

from tortoise.models import Model
from tortoise import fields

class Department(Model):
    id = fields.IntField(pk=True)
    title = fields.TextField()
    upper = fields.ForeignKeyField('models.Department', related_name='children')

    def __str__(self):
        if self.upper is not None:
            return f'{self.id} Department {self.title} of {self.upper.title}'
        else:
            return f'{self.id} Department {self.title}, head'

class Employee(Model):
    id = fields.IntField(pk=True)
    name = fields.TextField()
    dep = fields.ForeignKeyField('models.Department', related_name='employees')

    def __str__(self):
        return f'{self.id}. Employee {self.name} of {self.dep.title}'

以便每个对象在描述中显示其相关模型。但在 Tortoise 中我得到一个错误：

AttributeError：“QuerySet”对象没有属性“title”

我想不可能在__str__方法中等待查询。那么，是否有可能使用相关模型的字段来使用 Tortoise-ORM 创建对象表示？

我有一个现有代码，可以根据资源名称依次查询 AWS 资源，并进行过滤。当前的实现是线性的，从每个资源的一个函数转移到下一个函数，创建相应的客户端并使用客户端查询 aws。这当然要消耗大量的时间。是否可以以异步方式运行每个函数？代码流程类似于下面的代码片段，其中包含更多资源查询。任何意见/建议都会有帮助。

def query_acm():
    client = boto3.client('acm', region_name=region)
    client.get_paginator('list_certificates')
    # filter and write to file
def query_asg():
    client = boto3.client('autoscaling', region_name=region)
    paginator = client.get_paginator('describe_auto_scaling_groups')
    # paginate filter and write to file
def main():
    query_acm()
    query_asg()

我读过很多解释asyncPython 的不同文章。但他们都给出了一个例子asyncio.sleep(x)，就像这样：

import asyncio

async def test1 ():
    await asyncio.sleep(1)
    print(1)

async def test2 ():
    print(2)

async def main ():
    await asyncio.gather(test1(), test2())

asyncio.run(main()) #prints 2, then 1

在这种情况下，一切对我来说都很清楚：函数 test1 中的await 表示在执行 asyncio.sleep 期间我们可以做其他事情，例如执行函数 test2。

我不明白的是，如果我不在代码中使用睡眠，异步怎么会有用呢？在这种情况下如何同时运行函数？例如，如何同时运行下面示例中的函数 test1 和 test2？

import asyncio
import time

async def calculate (a):
    return a**a

async def test1 ():
    x = await calculate(1111111)
    print('done!')

async def test2 ():
    for i in range(100):
        print('.', end='')

async def main ():
    await asyncio.gather(test1(), test2())

asyncio.run(main()) …

我正在 Python 3.8 中尝试 Quart 和 Hypercorn。据我了解，Quart 通常用于单线程应用程序。但我看到 Hypercorn 有一个--workers运行该应用程序的选项。

workers      w, --workers      The number of workers to spawn and use.

单线程应用程序如何从使用多个工作线程中受益？

我创建了一个 dash 应用程序来显示另一个代码正在收集的信息，我想使用 Python 中的 asyncio 模块同时运行它们。

我的代码使用异步函数，而 Dash 应用程序（基于 Flask）在服务时阻止其他任何内容执行。

我不确定这是否需要打开更多线程。

这是我当前的代码，仅运行主协程。

async def main():
    some code here...

    while True:
        try:
            await client.handle_message()
        except ConnectionClosedError as error:
            logger.error(error)
    
        for strategy in strategies:
            await asyncio.create_task(...)
            some code here...

async def run_dashboard():
    app = create_app()
    app.run_server('0.0.0.0', 5000, debug=False)


if __name__ == '__main__':
    some code here...

    # Currently just runs the main coroutine
    asyncio.run(main())

如何同时运行 main 和 run_dashboard？

我有以下内容，但由于某种原因，我没有得到查询结果，而是得到了其他东西

这是我使用的Python包

https://pypi.org/project/databases/

这是文档

https://www.encode.io/databases/database_queries/

from databases import Database
database=Database('postgres://redacted')
await database.connect()
...
...
...
query = "SELECT orders.id AS orders_id, orders.notification_method AS orders_notification_method WHERE shipped=True"
result = await database.fetch_all(query=query))

这是我得到的而不是获取查询结果

打印（结果）

[<databases.backends.postgres.Record object at 0x7fb3f415ac50>, <databases.backends.postgres.Record object at 0x7fb3f415ae30>, <databases.backends.postgres.Record object at 0x7fb3f415a470>, <databases.backends.postgres.Record object at 0x7fb3f415af50>, <databases.backends.postgres.Record object at 0x7fb3f415ad70>, <databases.backends.postgres.Record object at 0x7fb3f415ab30>, <databases.backends.postgres.Record object at 0x7fb3f415a7d0>, <databases.backends.postgres.Record object at 0x7fb3f415ae90>]

type 表示这是一个列表

打印（类型（结果））

<class 'list'>

如何返回 SQL 查询的实际结果（即返回查询中的所有行）？

我真正想做的事

这是sqlalchemy有效的版本，但使用databases包无法正常工作，如上所述

我非常想要实现的是从 sqlalchemy …

我有一个使用 asyncio 和 websockets 的 python 套接字服务器。当 websocket 处于活动状态时，100 多个设备将连接并保持连接等待命令/消息。

有两个线程，第一个线程接受连接并将其详细信息添加到全局变量中，然后等待来自设备的消息：

async def thread1(websocket, path):
    client_address = await websocket.recv()
    CONNECTIONS[client_address] = websocket
    async for message in websocket:
        ... do something with message

start_server = websockets.serve(thread1, host, port)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.ensure_future(thread2())
asyncio.get_event_loop().run_forever()

第二个线程处理一些用户数据，一旦需要发送命令，它就会访问全局变量以获取 websocket 信息：

thread2()
    ...some data processing 
    soc = CONNECTIONS[ipaddress] 
    await soc.send("some message")

我的问题：允许另一个线程发送消息的最佳方式是什么？

我可以使用线程锁定和仅用于处理该数据的函数来保留全局变量safe，但是全局变量并不理想。我无法在线程之间发送信息，因为 thread1 正在stuck等待接收消息。