asgi-background

适用于任何ASGI框架的后台任务。

示例（Starlette）

from asgi_background import BackgroundTaskMiddleware, BackgroundTasks
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route


async def task(num: int) -> None:
    await anyio.sleep(1)
    print(num)


async def endpoint(request: Request) -> Response:
    tasks = BackgroundTasks(request.scope)
    await tasks.add_task(task, 1)
    return Response()


app = Starlette(
    routes=[Route("/", endpoint)],
    middleware=[Middleware(BackgroundTaskMiddleware)]
)

执行

与Starlette不同，我们不在ASGI请求/响应周期内执行后台任务。相反，我们将它们安排在绑定到应用程序生命周期的 TaskGroup 中。我们唯一保证的是，后台任务不会阻塞（在异步意义上，而不是全局解释器锁（GIL）意义上）发送响应，当应用程序关闭时，我们会（尝试）等待它们完成。就像Starlette的后台任务一样，您应该只为短生命周期的任务使用这些，它们不是像Redis、Celery等持久队列机制。为了了解背景，Kubernetes中默认的应用程序关闭宽限期为30秒，所以30秒可能是您允许任务运行的最长时间。