python Fastapi异步休眠()获取所用时间

omjgkv6w  于 2023-03-11  发布在  Python
关注(0)|答案(2)|浏览(218)
async def work():
    asyncio.sleep(3)

@router.get('')
async def test():
    time1 = monotonic()
    ...  # need to call work
    time2 = monotonic()
    return TestResponse(time=time2-time1)

work函数在同一时间只能调用一次。我需要编写test函数,这样如果我在同一时间多次调用端点,我会得到响应:time =与先前呼叫相差至少3秒的时间。

yzuktlbb

yzuktlbb1#

我不知 prop 体的时间安排,但是您可能想看看asyncio.Lock,然后您可以执行以下操作

import asyncio

work_lock = asyncio.Lock()

async def work():
    async with work_lock():
        # Do your actual work

这应该可以保证,如果两个请求调用您的端点,其中一个将等待另一个。

9njqaruj

9njqaruj2#

因此,您需要一个速率限制器-将其限制为每3秒1次调用信号量
你可以试试这个代码:

import asyncio
import time

PARALLEL = 1
INTERVAL = 3

semaphore = asyncio.Semaphore(PARALLEL)


async def work():
    with semaphore:
        start = time.time()
        try:
            # do actual I/O calls here
            ...
            result = ...
        finally:
            ellapsed = time.time() - start()
            # wait to liberate semaphore:
            await asyncio.sleep(INTERVAL - ellapsed) 

    return result

...
@router.get('')
async def test():
    time1 = monotonic()
    ...  # need to call work
    time2 = monotonic()
    return TestResponse(time=time2-time1)

相关问题