asgi-lifespan

程序化发送启动/关闭 lifespan 事件到 ASGI 应用程序。当与像 HTTPX 这样的 ASGI 兼容 HTTP 客户端结合使用时，这允许模拟或测试 ASGI 应用程序，而无需启动 ASGI 服务器。

功能

• 使用 LifespanManager 向 ASGI 应用程序发送 lifespan 事件。
• 支持 asyncio 和 trio。
• 完全类型注释。
• 100% 测试覆盖率。

安装

pip install 'asgi-lifespan==2.*'

用法

asgi-lifespan 提供了一个 LifespanManager，用于将 ASGI lifespan 事件程序化发送到 ASGI 应用程序。这可用于程序化启动/关闭 ASGI 应用程序，而无需启动 ASGI 服务器。

LifespanManager 可以在 asyncio 或 trio 上运行，并将自动检测正在使用的异步库。

基本用法

# example.py
from contextlib import asynccontextmanager
from asgi_lifespan import LifespanManager
from starlette.applications import Starlette

# Example lifespan-capable ASGI app. Any ASGI app that supports
# the lifespan protocol will do, e.g. FastAPI, Quart, Responder, ...

@asynccontextmanager
async def lifespan(app):
    print("Starting up!")
    yield
    print("Shutting down!")

app = Starlette(lifespan=lifespan)

async def main():
    async with LifespanManager(app) as manager:
        print("We're in!")

# On asyncio:
import asyncio; asyncio.run(main())

# On trio:
# import trio; trio.run(main)

输出

$ python example.py
Starting up!
We're in!
Shutting down!

发送 lifespan 事件进行测试

以下示例演示了如何结合使用 asgi-lifespan、HTTPX 和 pytest 来向 ASGI 应用程序发送测试请求。

• 安装依赖项

pip install asgi-lifespan httpx starlette pytest pytest-asyncio

• 测试脚本

# test_app.py
from contextlib import asynccontextmanager
import httpx
import pytest
import pytest_asyncio
from asgi_lifespan import LifespanManager
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route


@pytest_asyncio.fixture
async def app():
    @asynccontextmanager
    async def lifespan(app):
        print("Starting up")
        yield
        print("Shutting down")

    async def home(request):
        return PlainTextResponse("Hello, world!")

    app = Starlette(
        routes=[Route("/", home)],
        lifespan=lifespan,
    )

    async with LifespanManager(app) as manager:
        print("We're in!")
        yield manager.app


@pytest_asyncio.fixture
async def client(app):
    async with httpx.AsyncClient(app=app, base_url="http://app.io") as client:
        print("Client is ready")
        yield client


@pytest.mark.asyncio
async def test_home(client):
    print("Testing")
    response = await client.get("/")
    assert response.status_code == 200
    assert response.text == "Hello, world!"
    print("OK")

• 运行测试套件

$ pytest -s test_app.py
======================= test session starts =======================

test_app.py Starting up
We're in!
Client is ready
Testing
OK
.Shutting down

======================= 1 passed in 0.88s =======================

访问状态

LifespanManager 提供了一个 lifespan 状态，用于从生命周期周期中持久化数据，以便在请求/响应处理中使用。

为了使您的应用程序能够感知到它，请确保在上下文管理器内部使用 manager.app 而不是 app 本身。

例如，如果您使用 HTTPX 作为异步测试客户端

async with LifespanManager(app) as manager:
    async with httpx.AsyncClient(app=manager.app) as client:
        ...

API 参考

LifespanManager

def __init__(
    self,
    app: Callable,
    startup_timeout: Optional[float] = 5,
    shutdown_timeout: Optional[float] = 5,
)

一个异步上下文管理器，在进入时启动 ASGI 应用程序，在退出时关闭它。

更确切地说

• 在进入时，在后台向 app 发送一个 lifespan 请求，然后发送 lifespan.startup 事件并等待应用程序发送 lifespan.startup.complete。
• 在退出时，发送 lifespan.shutdown 事件并等待应用程序发送 lifespan.shutdown.complete。
• 如果在启动、关闭或 async with 块体中发生异常，则异常会向上冒泡，并且不会执行关闭操作。

示例

async with LifespanManager(app) as manager:
    # 'app' was started up.
    ...

# 'app' was shut down.

参数

• app (Callable)：一个 ASGI 应用程序。
• startup_timeout (Optional[float]，默认为 5)：等待应用程序启动的最大秒数。使用 None 表示没有超时。
• shutdown_timeout (Optional[float]，默认为 5)：等待应用程序关闭的最大秒数。使用 None 表示没有超时。

生成器

• manager (LifespanManager)：LifespanManager 本身。如果您使用 lifespan 状态，请使用 async with LifespanManager(app) as manager: ...，然后通过访问 manager.app 来获取对状态感知应用程序的引用。

引发

• LifespanNotSupported：如果应用程序似乎不支持 lifespan 协议。基于以下理由，如果应用程序支持 lifespan 协议，则它将成功接收 lifespan.startup ASGI 事件，不支持 lifespan 协议的情况在以下两种情况下检测到
  • 应用程序在第一次调用 receive() 之前调用了 send()。
  • 应用程序在启动期间抛出异常，在第一次调用 receive() 之前。例如，这可能是因为应用程序在类似 assert scope["type"] == "http" 的语句上失败。
• TimeoutError：如果启动或关闭超时。
• Exception：应用程序（在启动、关闭或 async with 块体中）引发的任何异常，这些异常不表明它不支持 lifespan 协议。

许可证

MIT

变更日志

此项目中所有值得注意的更改都将记录在此文件中。

格式基于 Keep a Changelog。

2.1.0 - 2023-03-28

添加

• 添加对 lifespan 状态的支持。（拉取请求 #59）

2.0.0 - 2022-11-11

移除

• 停止对 Python 3.6 的支持。（拉取请求 #55）

添加

• 官方支持 Python 3.11。（拉取请求 #55）
• 官方支持 Python 3.9 和 3.10。（拉取请求 #46 - 感谢 @euri10）

修复

• 确保与 mypy 0.990+ 的兼容性，这使得 no_implicit_optional 成为默认值。（拉取请求 #53 - 感谢 @AllSeeingEyeTolledEweSew）

1.0.1 - 2020-06-08

修复

• 将开发状态更新为 5 - 生产/稳定。（拉取请求 #32）

1.0.0 - 2020-02-02

移除

• 停止对 Lifespan 和 LifespanMiddleware 的支持。请改用 Starlette 内置的 lifespan 功能。（拉取请求 #27）

修复

• 使用 sniffio 来自动检测异步环境。（拉取请求 #28）
• 在 CI 上强制执行 100% 的测试覆盖率。（拉取请求 #29）

更改

• 通过切换到私有内部模块来强制从顶级包导入。（拉取请求 #26）

0.6.0 - 2019-11-29

更改

• 将 Lifespan 移动到 lifespan 模块。（拉取请求 #21）
• 重构 LifespanManager 以在 3.6 中取消对 asynccontextmanager 的依赖。（拉取请求 #20）

0.5.0 - 2019-11-29

• 进入 Beta 开发状态。

移除

• 移除 curio 支持。（拉取请求 #18）

添加

• 与源分布一起提供二进制分布（wheels）。

更改

• 对于 asyncio 和 trio 支持，使用自定义并发后端而不是 anyio。（拉取请求 #18）

0.4.2 - 2019-10-06

修复

• 确保将py.typed打包到软件包中，以便类型检查器可以检测类型注解。（拉取请求#16）

0.4.1 - 2019-09-29

修复

• 改进了LifespanManager中的错误处理（拉取请求#11）
  • 现在，上下文管理器主体中抛出的异常或关闭期间抛出的异常现在会被正确传播。
  • 当应用在至少调用一次receive()之前调用send()时，现在也会检测到不支持的lifespan。

0.4.0 - 2019-09-29

• 进入Alpha开发状态。

0.3.1 - 2019-09-29

添加

• 为LifespanManager添加可配置的超时时间。（拉取请求#10）

0.3.0 - 2019-09-29

添加

• 为将lifespan事件发送到ASGI应用添加了LifespanManager。（拉取请求#5）

0.2.0 - 2019-09-28

添加

• 添加了LifespanMiddleware，这是一个ASGI中间件，用于向ASGI应用添加lifespan支持。（拉取请求#9）

0.1.0 - 2019-09-28

添加

• 添加了Lifespan，这是一个实现lifespan协议并支持事件处理程序注册的ASGI应用。（拉取请求#7）

0.0.2 - 2019-09-28

修复

• 由于缺少MANIFEST.in，从PyPI安装曾经失败。

0.0.1 - 2019-09-28

添加

• 空包。