标签: python-asyncio

我一直在尝试创建一个服务器进程，该进程从客户端进程异步接收输入文件路径和输出路径。服务器执行一些依赖于数据库的转换，但为了简单起见，我们假设它只是将所有内容都设为大写。这是服务器的一个玩具示例：

import asyncio
import aiofiles as aiof
import logging
import sys


ADDRESS = ("localhost", 10000)

logging.basicConfig(level=logging.DEBUG,
                    format="%(name)s: %(message)s",
                    stream=sys.stderr)

log = logging.getLogger("main")
loop = asyncio.get_event_loop()


async def server(reader, writer):
    log = logging.getLogger("process at {}:{}".format(*ADDRESS))
    paths = await reader.read()
    in_fp, out_fp = paths.splitlines()
    log.debug("connection accepted")
    log.debug("processing file {!r}, writing output to {!r}".format(in_fp, out_fp))
    async with aiof.open(in_fp, loop=loop) as inp, aiof.open(out_fp, "w", loop=loop) as out:
        async for line in inp:
            out.write(line.upper())
        out.flush()
    writer.write(b"done")
    await writer.drain()
    log.debug("closing")
    writer.close()
    return


factory = asyncio.start_server(server, *ADDRESS) …

我试图通过 Flask 方法调用阻塞函数，但需要几秒钟，所以我想我可以做一些异步调用来加快速度，但它没有按预期工作。显然使用 asyncio 我不能只是在后台启动一个协程而不等待执行结束，也许我需要使用线程？或者使用 grequest 因为我的阻塞功能正在使用请求...

到目前为止，这是我的代码：

@app.route("/ressource", methods=["GET"])
def get_ressource():
    do_stuff()
    return make_response("OK",200)  

def do_stuff():
  # Some stuff
  fetch_ressource()


async def fetch_ressource():
    return await blocking_function()


def blocking_function():
  # Take 2-3 seconds
  result = request.get('path/to/remote/ressource')
  put_in_database(result)

我听说过 Celeri，但对于只有一个功能来说似乎有点矫枉过正。

这似乎是一个常见问题，请参见示例：RuntimeError: This event loop is already running in python

但就我而言，我只启动一次事件循环，至少就我所见。此示例也直接遵循此处的说明：

import asyncio

loop = asyncio.get_event_loop()

async def coroutine():
    print("hey")
    await asyncio.sleep(1)
    print("ho")
    return 1


async def main():

    tasks = []
    for i in range(2):
        tasks.append(asyncio.ensure_future(coroutine()))
    await asyncio.gather(*tasks)

results = loop.run_until_complete(main())
loop.close()

这将打印一条错误消息，以及协程中的 print() 调用的输出：

---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
<ipython-input-1-f4a74fbfac46> in <module>
     16     await asyncio.gather(*tasks)
     17 
---> 18 results = loop.run_until_complete(asyncio.gather(*tasks))
     19 loop.close()

~/anaconda3/envs/keras_dev/lib/python3.6/asyncio/base_events.py in run_until_complete(self, future)
    453         future.add_done_callback(_run_until_complete_cb)
    454         try:
--> 455             self.run_forever()
    456 …

我通过 POST 接受该文件。本地保存时，可以使用file.read()读取内容，但是显示通过file.name不正确(16)的名称。当我尝试按此名称查找它时，出现错误。可能是什么问题？

我的代码：

  @router.post(
    path="/po/{id_po}/upload",
    response_model=schema.ContentUploadedResponse,
)
async def upload_file(
        id_po: int,
        background_tasks: BackgroundTasks,
        uploaded_file: UploadFile = File(...)):
    """pass"""
    uploaded_file.file.rollover()
    uploaded_file.file.flush()
    #shutil.copy(uploaded_file.file.name, f'/home/fyzzy/Desktop/api/{uploaded_file.filename}')
    background_tasks.add_task(s3_upload, uploaded_file=fp)
    return schema.ContentUploadedResponse()

我正在尝试异步处理多个文件，并且处理每个文件需要对 SQLite 数据库进行一些读取和写入操作。我一直在寻找一些选项，我在这里找到了 aiosqlite 模块。但是，我在这里阅读 SQLite 文档，它说它支持多线程模式。事实上，默认模式是“序列化”，这意味着它“可以被多个线程安全地使用，没有限制”。

我不明白有什么区别。aiosqlite 文档说：

aiosqlite 允许在主 AsyncIO 事件循环上与 SQLite 数据库交互，而不会在等待查询或数据获取时阻止其他协程的执行。它通过为每个连接使用一个共享线程来实现这一点。

我知道 aiosqlite 和 sqlite 上的“多线程”模式之间存在差异，因为多线程模式每个线程只需要一个连接，而在 aiosqlite 中，您可以跨多个线程重用这个单个连接。但这与序列化模式不一样，它可以“由多个线程无限制地使用”？

编辑：我现在的问题是“我目前的理解是否正确？”：

1. “序列化”模式下的 Sqlite 可以同时被多个线程使用，因此如果我threading在 python 中使用该模块并生成多个线程，则将使用它。在这里，我可以选择为每个线程使用单独的连接或跨多个线程共享连接。
2. aiosqlite 与 asyncio 一起使用。因此，由于 asyncio 有多个协程共享一个线程，因此 aiosqlite 也可以与一个线程一起使用。因此，我创建了一个在所有协程之间共享的连接。
3. 由于 aiosqlite 基本上是 sqlite 的一个包装器，我可以结合 1 和 2 的功能。所以我可以有多个线程，其中每个线程都有一个带有多个协程的 asyncio 事件循环。因此，基本的 sqlite 功能将处理多线程，而 aiosqlite 将处理协程。

我开始使用 AsyncIO 和 AioHTTP，我正在编写一些基本代码来熟悉语法。我尝试了以下应该同时执行 3 个请求的代码：

import time
import logging
import asyncio
import aiohttp
import json
from aiohttp import ClientSession, ClientResponseError
from aiocfscrape import CloudflareScraper

async def nested(url):
    async with CloudflareScraper() as session:
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    URL = "https://www.binance.com/api/v3/exchangeInfo"
    await asyncio.gather(nested(URL), nested(URL), nested(URL))

asyncio.run(main())

这是输出：

raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

我不明白为什么我会收到这个错误，有人可以帮我解决这个问题吗？

我见过两种获取asyncio Lock 的方法：

async def main(lock):
  async with lock:
    async.sleep(100)

和

async def main(lock):
  with await lock:
    async.sleep(100)

它们之间有什么区别？

tl;dr

1. 以下哪个选项是正确的工作流程fastapi？
2. 如何以编程方式测试调用是否真正被阻止（而不是从浏览器手动测试）？是否有对or 的压力测试扩展？uvicornfastapi

我在服务器中有许多端点fastapi（uvicorn目前正在使用），它们对常规同步 Python 代码有长时间的阻塞调用。尽管有文档（https://fastapi.tiangolo.com/async/），我仍然不清楚我是否应该单独使用def或async def混合我的功能。

据我了解，我有三个选择，假设：

def some_long_running_sync_function():
  ...

选项 1 - 始终def仅用于端点

@app.get("route/to/endpoint")
def endpoint_1:
  some_long_running_sync_function()


@app.post("route/to/another/endpoint")
def endpoint_2:
  ...

选项 2 - 始终async def仅使用并在执行器中运行阻塞同步代码

import asyncio


@app.get("route/to/endpoint")
async def endpoint_1:
  loop = asyncio.get_event_loop()
  await loop.run_in_executor(None, some_long_running_sync_function)


@app.post("route/to/another/endpoint")
async def endpoint_2:
  ...

选项 3 - 混合搭配def并async def基于底层调用

import asyncio


@app.get("route/to/endpoint")
def endpoint_1: …

在这个页面上我读到了这样的内容：

asyncio 模块中的协程不受全局解释器锁或 GIL 的限制。

asyncio但是，如果事件循环和threading线程都在带有 GIL 的单个 Python 进程中运行，这怎么可能呢？

据我了解， GIL 对 的影响asyncio不会像对 那样强烈threading，因为在 的情况下threading，解释器会切换到无用的操作，例如time.sleep()。因此，在使用时asyncio，建议使用asyncio.sleep().

据我所知，这些工具的设计目的略有不同，threading更常用于执行 IO 绑定操作的“遗留”阻塞代码，以及asyncio非阻塞代码。

我使用 OpenAI API。我从 PowerPoint 演示文稿中提取了幻灯片文本，并为每张幻灯片编写了提示。现在，我想进行异步 API 调用，以便同时处理所有幻灯片。

这是异步主函数的代码：

for prompt in prompted_slides_text:
    task = asyncio.create_task(api_manager.generate_answer(prompt))
    tasks.append(task)
results = await asyncio.gather(*tasks)

这是generate_answer函数：

@staticmethod
    async def generate_answer(prompt):
        """
        Send a prompt to OpenAI API and get the answer.
        :param prompt: the prompt to send.
        :return: the answer.
        """
        completion = await openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}]

        )
        return completion.choices[0].message.content

问题是：

对象 OpenAIOb​​ject 不能在“await”表达式中使用

我不知道如何等待generate_answer函数中的响应

将不胜感激任何帮助！