标签: python-asyncio

通常在 IO 操作中我们用来iter()读取哨兵值：

from sys import stdout

with open(r"Z:\github\StackOverFlow\temp.json", "r") as fp:
    for chunk in iter(lambda :fp.read(64), ""):
        stdout.write(chunk)

iter()但是除了for waitable之外还有其他选择吗asyncio.Queue.get()？

from sys import stdout

with open(r"Z:\github\StackOverFlow\temp.json", "r") as fp:
    for chunk in iter(lambda :fp.read(64), ""):
        stdout.write(chunk)

当然，这不会起作用，因为它需要可调用，await不能在非异步函数内调用。

情况不允许queue.get_nowait()，因为队列大部分时间都是空的。

简单的修复方法是使用while循环：

for val in iter(lambda: await queue.get(), sentinel):
    queue.task_done()
    print(val)

但我担心这是否会损害可读性和清晰度。

我有一个使用 fastapi 的异步 http web 服务。我在服务器上的不同端口上运行同一服务的多个实例，并且前面有一个 nginx 服务器，因此我可以利用它们。我有一个特定的资源，需要保护只有一个客户端可以访问它。

@app.get("/do_something")
async def do_something():
     critical_section_here()

我尝试使用如下文件锁来保护这个关键部分：

@app.get("/do_something")
async def do_something():
    with FileLock("dosomething.lock"):
        critical_section()

这样可以防止多个进程同时进入临界区。但我发现这实际上会死锁。考虑以下事件：

1. 客户端1连接8000端口并进入临界区
2. 当客户端 1 仍在使用资源时，客户端 2 被路由到相同的端口 8000，然后它将尝试获取文件锁，但它不能，因此它将继续尝试，这将阻止客户端 1 的执行，而客户端 1 永远不会能够释放文件锁，这意味着不仅此进程被锁定，所有其他服务器实例也将被锁定。

有没有一种方法可以让我协调这些进程，以便只有其中一个进程访问关键部分？我想过为文件锁添加超时，但我真的不想拒绝用户，我只想等到轮到他/她进入临界区。

我正在设计一个应用程序，在其中创建 10000 个任务。

我有一个协程（查询 Web 服务的异步函数），我使用不同的参数在协程上执行 asyncio create_task 10000 次。

可以同时运行的任务是否有上限？

我正在尝试创建一个异步视图来解决 Django 3.1 的端点问题。

我有我的 AsyncView，在这个视图中我有下面的代码行：

try:
    db_result= await sync_to_async(list)(Plan.objects.filter())
    serializer =  MySerializer(db_results, many=True)
    return serializer.data
except Exception:
    # Handle Exception Code

在我的序列化器中，我有一个自定义字段。使用其 obj 属性对数据库进行其他查询来检索其数据。这是我遇到一些问题的地方，因为这是一个异步上下文。

序列化器代码：

MySerializer(serializers.ModelSerializer):

    custom_field = serializers.SerializerMethodField(required=False)

    def get_custom_field(self, obj):
        try:
            # This is failing, is not retrieving data from DB since it's async context
            related_stuff = obj.related_stuff.filter(filter_condition=filter_condition)
            # More code after this
            .
            .
            .
            .
            return custom_field_object
        except Exception:
            return None

我尝试在视图代码中的序列化器之前放置一个等待，并将 async 属性添加到 get_custom_field 方法（使用数据库查询中的sync_to_async 运算符）。什么也没得到。

有人知道如何解决这个具体问题吗？

我有使用 zeep 创建肥皂客户端的代码。我的服务器不返回 wsdl 文件，但我在本地有它。

同步版本的工作原理如下：

import uuid
from os import path
import structlog
import zeep

logger = structlog.get_logger(__name__)



class SyncClient(object):
    def __init__(self, ip_address: str):
        self.ip_address = ip_address
        self.port = 8080
        self.soap_client = None
        self.corrupt_timeseries_files = []
        self.id = uuid.uuid4()

    def connect_soap_client(self):
        this_files_dir = path.dirname(path.realpath(__file__))
        wsdl = 'file://{}'.format(path.join(this_files_dir, 'SOAPInterface.wsdl'))

        transport = zeep.Transport(timeout=5, operation_timeout=3)

        client = zeep.Client(wsdl, transport=transport)
        location = "http://{}:{}".format(self.ip_address, str(self.port))

        self.soap_client = client.create_service("{urn:webservices}SOAPInterface", location)

然后 asyc 客户端看起来像这样：

class AsyncClient(object):
    def __init__(self, ip_address: str):
        self.ip_address = ip_address
        self.port = 8080 …

在 FastAPI 中，当标准函数用作依赖项时，可以将其声明为常规def函数或异步async def函数。FastAPI 声称无论哪种情况它都会做正确的事情。

然而，以这种方式创建的依赖项对于自动完成并不像类依赖项那么友好。此外，类依赖项具有更好的声明语法，只需指定一次依赖项类型，FastAPI 就会找出您所指的依赖项。

def read_item(常见: CommonQueryParam = Depends()):

但是类依赖项需要执行异步操作作为其初始化的一部分。是否可以同时使用类依赖和异步。显然，不能将类__init__函数声明为异步。还有其他方法可以使其发挥作用吗？

我的 pyppeteer 连接意外关闭，导致我的 Python 程序挂起，而不是关闭并正确记录错误。有谁知道如何正确捕获此异常并正确退出Python程序？

这是我的脚本的一部分：

 async def _proxy_browser(self, url, headless = False):

    args = []

    for i in range(5):

        try:
            browser = await launch(headless = headless,
                                   args = args,
                                   defaultViewport = None)
            

            page = await browser.newPage()
            await page.setUserAgent('Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:83.0) Gecko/20100101 Firefox/83.0')
            await page.goto(url, {'waitUntil' : 'domcontentloaded'})
            time.sleep(5)
           
            content = await page.content()

            return content

        except (pyppeteer.errors.PageError, pyppeteer.errors.TimeoutError) as e:
          print('error', e)
          time.sleep(2)
          continue

        finally:
            time.sleep(1)
            await browser.close()

    return 

def proxy_browser_request(self, url, headless = False):

    loop …

我指的是这个 repo将mmaction2 grad-cam demo从短视频离线推理改编为长视频在线推理。脚本如下所示：

注意：为了使该脚本可以轻松重现，我注释掉了一些需要许多依赖项的代码。

import cv2
import numpy as np
import torchvision.transforms as transforms
import sys
from PIL import Image
#from mmaction.apis import init_recognizer
#from utils.gradcam_utils import GradCAM
import torch
import asyncio
from concurrent.futures import ProcessPoolExecutor
from functools import partial
# sys.path.append('./utils')


async def preprocess_img(arr):
    image = Image.fromarray(np.uint8(arr))
    mean = [0.485, 0.456, 0.406]
    std = [0.229, 0.224, 0.225]
    transform = transforms.Compose([
        transforms.Resize((model_input_height, model_input_width)),
        transforms.ToTensor(),
        transforms.Normalize(mean, std, inplace=False),
    ])
    normalized_img = transform(image)
    img_np = normalized_img.numpy()
    return …

我使用 tor 的 http 代理（使用 aiohttp_socks）通过 aiohttp 发送多个请求

完成一些请求后，我收到以下错误：

Traceback (most recent call last):
  File "main.py", line 171, in <module>
    loop.run_until_complete(future)
  File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "main.py", line 95, in get_market_pages
    async with session.get(active_link, headers=headers) as response:
  File "/home/mrlalatg/.local/lib/python3.8/site-packages/aiohttp/client.py", line 1012, in __aenter__
    self._resp = await self._coro
  File "/home/mrlalatg/.local/lib/python3.8/site-packages/aiohttp/client.py", line 504, in _request
    await resp.start(conn)
  File "/home/mrlalatg/.local/lib/python3.8/site-packages/aiohttp/client_reqrep.py", line 847, in start
    message, payload = await self._protocol.read()  # type: ignore  # noqa
  File "/home/mrlalatg/.local/lib/python3.8/site-packages/aiohttp/streams.py", line 591, in …

我有几个异步函数。如果我调用main()，我无法使用第二个check()函数，因为main()它在 while 循环中连续运行。我可以main()“在后台”运行以使其余功能可用吗？

import asyncio

async def main():
    while True:
        # update database

async def check():
    # confirm update

async def process(command):
    if command == "main":
        await main()
    elif command == "check":
        await check()