Programmatic startup/shutdown of ASGI apps.

Overview

asgi-lifespan

Build Status Coverage Package version

Programmatically send startup/shutdown lifespan events into ASGI applications. When used in combination with an ASGI-capable HTTP client such as HTTPX, this allows mocking or testing ASGI applications without having to spin up an ASGI server.

Features

  • Send lifespan events to an ASGI app using LifespanManager.
  • Support for asyncio and trio.
  • Fully type-annotated.
  • 100% test coverage.

Installation

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

Usage

asgi-lifespan provides a LifespanManager to programmatically send ASGI lifespan events into an ASGI app. This can be used to programmatically startup/shutdown an ASGI app without having to spin up an ASGI server.

LifespanManager can run on either asyncio or trio, and will auto-detect the async library in use.

Basic usage

# example.py
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, ...
app = Starlette(
    on_startup=[lambda: print("Starting up!")],
    on_shutdown=[lambda: print("Shutting down!")],
)

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

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

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

Output:

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

Sending lifespan events for testing

The example below demonstrates how to use asgi-lifespan in conjunction with HTTPX and pytest in order to send test requests into an ASGI app.

  • Install dependencies:
pip install asgi-lifespan httpx starlette pytest pytest-asyncio
  • Test script:
# test_app.py
import httpx
import pytest
from asgi_lifespan import LifespanManager
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route


@pytest.fixture
async def app():
    async def startup():
        print("Starting up")

    async def shutdown():
        print("Shutting down")

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

    app = Starlette(
        routes=[Route("/", home)],
        on_startup=[startup],
        on_shutdown=[shutdown]
    )

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


@pytest.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")
  • Run the test suite:
$ 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 =======================

API Reference

LifespanManager

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

An asynchronous context manager that starts up an ASGI app on enter and shuts it down on exit.

More precisely:

  • On enter, start a lifespan request to app in the background, then send the lifespan.startup event and wait for the application to send lifespan.startup.complete.
  • On exit, send the lifespan.shutdown event and wait for the application to send lifespan.shutdown.complete.
  • If an exception occurs during startup, shutdown, or in the body of the async with block, it bubbles up and no shutdown is performed.

Example

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

# 'app' was shut down.

Parameters

  • app (Callable): an ASGI application.
  • startup_timeout (Optional[float], defaults to 5): maximum number of seconds to wait for the application to startup. Use None for no timeout.
  • shutdown_timeout (Optional[float], defaults to 5): maximum number of seconds to wait for the application to shutdown. Use None for no timeout.

Raises

  • LifespanNotSupported: if the application does not seem to support the lifespan protocol. Based on the rationale that if the app supported the lifespan protocol then it would successfully receive the lifespan.startup ASGI event, unsupported lifespan protocol is detected in two situations:
    • The application called send() before calling receive() for the first time.
    • The application raised an exception during startup before making its first call to receive(). For example, this may be because the application failed on a statement such as assert scope["type"] == "http".
  • TimeoutError: if startup or shutdown timed out.
  • Exception: any exception raised by the application (during startup, shutdown, or within the async with body) that does not indicate it does not support the lifespan protocol.

License

MIT

Comments
  • Should we get rid of `Lifespan` and `LifespanMiddleware`?

    Should we get rid of `Lifespan` and `LifespanMiddleware`?

    Currently this package provides three components…

    • Lifespan: an ASGI app that acts as a container of startup/shutdown event handlers.
    • LifespanMiddleware: routes lifespan requests to a Lifespan app, used to add lifespan support to any ASGI app.
    • LifespanManager: sends lifespan events into an ASGI app.

    LifespanManager is obviously useful, as it's addressing a use case that's not addressed elsewhere — sending lifespan events into an app, with async support. (Starlette does this with its TestClient, but it's sync only, and HTTPX does not send lifespan events when calling into an ASGI app.)

    I'm less confident that Lifespan and LifespanMiddleware are worth keeping in here, though. Most ASGI frameworks provide a lifespan implementation anyway…

    Eg. Starlette essentially has the exact same code than our own Lifespan built into its router:

    https://github.com/encode/starlette/blob/6a65461c6e13d0a55243c19ce5580b1b5a65a770/starlette/routing.py#L508-L527

    The super-light routing provided by our LifespanMiddleware is built into the router as well:

    https://github.com/encode/starlette/blob/6a65461c6e13d0a55243c19ce5580b1b5a65a770/starlette/routing.py#L529-L540

    So I'm wondering if using these two components on their own own is a use case that people encounter in the wild. I think it's not, and that maybe we should drop them and point people at using Starlette for handling lifespan on the application side.

    opened by florimondmanca 8
  • run no_implicit_optional

    run no_implicit_optional

    mypy has recently disabled implicit Optional by default (so def f(x: int = None): is no longer allowed)

    They created a tool, no_implicit_optional, to auto-modify code.

    However, I found that since my app's tests use asgi-lifespan, I cannot just run the no_implicit_optional tool on my own code. Since mypy now defaults to implicit_optional = False, this results in wrong typing for asgi_lifespan.LifespanManager. mypy then complains about my own correct use of LifespanManager.

    For now, I'm working around my use of LifespanManager with # type: ignore, but it would be nice to make asgi-lifespan use strict Optional.

    Before:

    $ mypy .
    src/asgi_lifespan/_concurrency/trio.py:54: error: Incompatible return value type (got "Background", expected "AbstractAsyncContextManager[Any]")  [return-value]
    src/asgi_lifespan/_concurrency/trio.py:54: note: Following member(s) of "Background" have conflicts:
    src/asgi_lifespan/_concurrency/trio.py:54: note:     Expected:
    src/asgi_lifespan/_concurrency/trio.py:54: note:         def __aexit__(self, Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType], /) -> Coroutine[Any, Any, Optional[bool]]
    src/asgi_lifespan/_concurrency/trio.py:54: note:     Got:
    src/asgi_lifespan/_concurrency/trio.py:54: note:         def __aexit__(self, exc_type: Optional[Type[BaseException]] = ..., exc_value: BaseException = ..., traceback: TracebackType = ...) -> Coroutine[Any, Any, None]
    src/asgi_lifespan/_concurrency/trio.py:69: error: Incompatible default for argument "exc_value" (default has type "None", argument has type "BaseException")  [assignment]
    src/asgi_lifespan/_concurrency/trio.py:69: note: PEP 484 prohibits implicit Optional. Accordingly, mypy has changed its default to no_implicit_optional=True
    src/asgi_lifespan/_concurrency/trio.py:69: note: Use https://github.com/hauntsaninja/no_implicit_optional to automatically upgrade your codebase
    src/asgi_lifespan/_concurrency/trio.py:70: error: Incompatible default for argument "traceback" (default has type "None", argument has type "TracebackType")  [assignment]
    src/asgi_lifespan/_concurrency/trio.py:70: note: PEP 484 prohibits implicit Optional. Accordingly, mypy has changed its default to no_implicit_optional=True
    src/asgi_lifespan/_concurrency/trio.py:70: note: Use https://github.com/hauntsaninja/no_implicit_optional to automatically upgrade your codebase
    src/asgi_lifespan/_concurrency/asyncio.py:51: error: Incompatible return value type (got "Background", expected "AbstractAsyncContextManager[Any]")  [return-value]
    src/asgi_lifespan/_concurrency/asyncio.py:51: note: Following member(s) of "Background" have conflicts:
    src/asgi_lifespan/_concurrency/asyncio.py:51: note:     Expected:
    src/asgi_lifespan/_concurrency/asyncio.py:51: note:         def __aexit__(self, Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType], /) -> Coroutine[Any, Any, Optional[bool]]
    src/asgi_lifespan/_concurrency/asyncio.py:51: note:     Got:
    src/asgi_lifespan/_concurrency/asyncio.py:51: note:         def __aexit__(self, exc_type: Optional[Type[BaseException]] = ..., exc_value: BaseException = ..., traceback: TracebackType = ...) -> Coroutine[Any, Any, None]
    src/asgi_lifespan/_concurrency/asyncio.py:71: error: Incompatible default for argument "exc_value" (default has type "None", argument has type "BaseException")  [assignment]
    src/asgi_lifespan/_concurrency/asyncio.py:71: note: PEP 484 prohibits implicit Optional. Accordingly, mypy has changed its default to no_implicit_optional=True
    src/asgi_lifespan/_concurrency/asyncio.py:71: note: Use https://github.com/hauntsaninja/no_implicit_optional to automatically upgrade your codebase
    src/asgi_lifespan/_concurrency/asyncio.py:72: error: Incompatible default for argument "traceback" (default has type "None", argument has type "TracebackType")  [assignment]
    src/asgi_lifespan/_concurrency/asyncio.py:72: note: PEP 484 prohibits implicit Optional. Accordingly, mypy has changed its default to no_implicit_optional=True
    src/asgi_lifespan/_concurrency/asyncio.py:72: note: Use https://github.com/hauntsaninja/no_implicit_optional to automatically upgrade your codebase
    src/asgi_lifespan/_manager.py:97: error: Incompatible default for argument "exc_type" (default has type "None", argument has type "Type[BaseException]")  [assignment]
    src/asgi_lifespan/_manager.py:97: note: PEP 484 prohibits implicit Optional. Accordingly, mypy has changed its default to no_implicit_optional=True
    src/asgi_lifespan/_manager.py:97: note: Use https://github.com/hauntsaninja/no_implicit_optional to automatically upgrade your codebase
    src/asgi_lifespan/_manager.py:98: error: Incompatible default for argument "exc_value" (default has type "None", argument has type "BaseException")  [assignment]
    src/asgi_lifespan/_manager.py:98: note: PEP 484 prohibits implicit Optional. Accordingly, mypy has changed its default to no_implicit_optional=True
    src/asgi_lifespan/_manager.py:98: note: Use https://github.com/hauntsaninja/no_implicit_optional to automatically upgrade your codebase
    src/asgi_lifespan/_manager.py:99: error: Incompatible default for argument "traceback" (default has type "None", argument has type "TracebackType")  [assignment]
    src/asgi_lifespan/_manager.py:99: note: PEP 484 prohibits implicit Optional. Accordingly, mypy has changed its default to no_implicit_optional=True
    src/asgi_lifespan/_manager.py:99: note: Use https://github.com/hauntsaninja/no_implicit_optional to automatically upgrade your codebase
    

    I ran:

    $ no_implicit_optional .
    Calculating full-repo metadata...
    Executing codemod...
    Finished codemodding 15 files!
     - Transformed 15 files successfully.
     - Skipped 0 files.
     - Failed to codemod 0 files.
     - 0 warnings were generated.
    $ isort  src/ tests/
    Fixing /home/amnesia/Persistent/checkout/asgi-lifespan/src/asgi_lifespan/_manager.py
    Fixing /home/amnesia/Persistent/checkout/asgi-lifespan/src/asgi_lifespan/_concurrency/trio.py
    Fixing /home/amnesia/Persistent/checkout/asgi-lifespan/src/asgi_lifespan/_concurrency/asyncio.py
    

    After:

    $ mypy .
    Success: no issues found in 15 source files
    
    opened by AllSeeingEyeTolledEweSew 4
  • Rationale for strong dependency on starlette 0.13?

    Rationale for strong dependency on starlette 0.13?

    Hello,

    I think this is an awesome project. However, the strong dependency on starlette 0.13 makes my test hang forever (I haven't upgrade my code to 0.13 yet so I guess that's that...). Using starlette 0.12 doesn't cause an issue as far as I can tell with this plugin. So, I wondered about the rationale for such strict dependency?

    As it stands, I had to drop the plugin when I wanted to migrate all my tests to it :(

    Thanks,

    • Sylvain
    question 
    opened by Lawouach 4
  • Error reproducing an example from the REAME.md

    Error reproducing an example from the REAME.md

    I reproduced the app example given in the README.txt and I got the following error:

    ❯ pytest -s test_app.py
    ========================================== test session starts ===========================================
    platform linux -- Python 3.10.7, pytest-7.1.3, pluggy-1.0.0
    rootdir: /home/daniel/coding/python/test-asgi-lifespan
    plugins: anyio-3.6.1, asyncio-0.19.0
    asyncio: mode=strict
    collected 1 item
    
    test_app.py Testing
    F
    
    ================================================ FAILURES ================================================
    _______________________________________________ test_home ________________________________________________
    
    client = <async_generator object client at 0x7f65b037a9c0>
    
        @pytest.mark.asyncio
        async def test_home(client):
            print("Testing")
    >       response = await client.get("/")
    E       AttributeError: 'async_generator' object has no attribute 'get'
    
    test_app.py:43: AttributeError
    ======================================== short test summary info =========================================
    FAILED test_app.py::test_home - AttributeError: 'async_generator' object has no attribute 'get'
    =========================================== 1 failed in 0.06s ============================================
    

    I run pytest -s test_app.py on a new environment with the installed packages explained in the README. Here is my pip list:

    Package        Version
    -------------- ---------
    anyio          3.6.1
    asgi-lifespan  1.0.1
    attrs          22.1.0
    certifi        2022.9.14
    h11            0.12.0
    httpcore       0.15.0
    httpx          0.23.0
    idna           3.4
    iniconfig      1.1.1
    packaging      21.3
    pip            22.2.2
    pluggy         1.0.0
    py             1.11.0
    pyparsing      3.0.9
    pytest         7.1.3
    pytest-asyncio 0.19.0
    rfc3986        1.5.0
    setuptools     63.2.0
    sniffio        1.3.0
    starlette      0.20.4
    tomli          2.0.1
    

    Sounds like pytest fixture is returning an async iterable instead the value in the yield. Could it be caused by some bug in some version of some package, or is a problem of the asgi-lifespan? Or I mess up with my setting?

    opened by netsmash 3
  • Testing use case guide

    Testing use case guide

    This package was originally prompted by https://github.com/encode/httpx/issues/350.

    The use case was to test a Starlette app by making calls within a lifespan context, e.g.:

    # asgi.py
    from starlette.applications import Starlette
    from starlette.responses import PlainTextResponse
    
    app = Starlette()
    
    @app.route("/")
    async def home(request):
        return PlainTextResponse("Hello, world!")
    
    # tests.py
    import httpx
    import pytest
    from asgi_lifespan import LifespanManager
    
    from .asgi import app
    
    @pytest.fixture
    async def app():
        async with LifespanManager(app):
            yield
    
    @pytest.fixture
    async def client(app):
        async with httpx.AsyncClient(app=app) as client:
            yield client
    
    @pytest.mark.asyncio
    async def test_app(client):
        r = await client.get("http://testserver")
        assert r.status_code == 200
        assert r.text == "Hello, world!"
    

    We should add a section to the documentation on how to use asgi-lifespan in this use case.

    documentation good first issue 
    opened by florimondmanca 3
  • Can't use LifespanManager with pytest-asyncio

    Can't use LifespanManager with pytest-asyncio

    Related to https://github.com/agronholm/anyio/issues/74

    from asgi_lifespan import LifespanManager, Lifespan
    import pytest
    
    
    @pytest.fixture
    async def app():
        app = Lifespan()
        async with LifespanManager(app):
            yield app
    
    
    @pytest.mark.asyncio
    async def test_something(app):
        pass
    

    Running pytest tests.py fails with a KeyError.

    This is an issue with anyio that is being resolved — tracking here for clarity.

    bug 
    opened by florimondmanca 2
  • Release 2.0.0

    Release 2.0.0

    2.0.0 (November 11, 2022)

    Removed

    • Drop support for Python 3.6. (Pull #55)

    Added

    • Add official support for Python 3.11. (Pull #55)
    • Add official support for Python 3.9 and 3.10. (Pull #46 - Thanks euri10)

    Fixed

    • Ensure compatibility with mypy 0.980+, which made no_implicit_optional the default. (Pull #53 - Thanks AllSeeingEyeTolledEweSew)
    opened by florimondmanca 1
  • Update package setup

    Update package setup

    • Drop Python 3.6 support.
    • Add official Python 3.11 support.
    • Bump dev dependencies.
    • Switch to pyproject.toml.
    • Switch to Makefile scripts.
    • Use newest CI templates.
    • Update tests.
    opened by florimondmanca 1
  • Test fail on FreeBSD

    Test fail on FreeBSD

    I'm having trouble with tests. I attached the log of "make test" command. Could it be that pytest-4 is too old and it requires pytest-6 (upgrade is in progress on FreeBSD bugzilla)? asgi-lifespan.log

    opened by mekanix 1
  • Handling `lifespan.shutdown.failed` ASGI messages

    Handling `lifespan.shutdown.failed` ASGI messages

    Prompted by https://gitter.im/encode/community?at=5f37fc0eba27767ee5f1d4ed

    As per the ASGI spec, when an application shutdown fails, the app should send the lifespan.shutdown.failed ASGI message.

    Currently, LifespanManager does not listen for this message on shutdown, but rather catches any exception raised during startup.

    This works fine for Starlette apps (since Starlette does not obey the spec, and lets any shutdown exception bubble through instead of sending lifespan.shutdown.failed), but for example I'd expect we have a small interop issue with Quart here (see source).

    It's possible we might need to wait for this to be resolved in Starlette before fixing things, but not clear we can't support both use cases at the same time either.

    Related issue in Uvicorn: https://github.com/encode/uvicorn/pull/755

    enhancement 
    opened by florimondmanca 1
  • Improve auto-detection of async environment

    Improve auto-detection of async environment

    ~Async auto-detection was brittle, as it was basically using trio if sniffio was installed, and asyncio otherwise. (But users might have sniffio installed and be running in an asyncio environment…).~

    ~So, use the same ideas than sniffio. (Not decided on using it yet, for the sake of staying no-dependencies.)~

    Switch to sniffio for detecting the concurrency backend.

    refactor 
    opened by florimondmanca 1
  • [feature request] support ContextVars

    [feature request] support ContextVars

    It seem that setting context variables in on_startup is not supported.

    My naïve understanding is that there needs to be a new (copied) context in which all of on_startup, asgi request and then on_shutdown are ran.

    Otherwise, the value that's been set is not accessible in the request handlers and/or the context var cannot be reset in the shutdown function using the token that was generated in the startup function.

    question 
    opened by dimaqq 8
Releases(2.0.0)
Owner
Florimond Manca
Pythonista, open source developer, casual tech blogger. Idealist on a journey, and it’s good fun!
Florimond Manca
【幼盾】个性化图片徽章服务!

【幼盾】个性化图片徽章服务! 你对方形的徽章感到无聊了吗?想要定制属于自己的开源项目徽章了吗? 快来使用unv-shield吧! unv-shield提供包含自定义图片的徽章服务,可以让你的项目主页更加个性化!

黄巍 130 Dec 23, 2022
Dyson Sphere Program Blueprint Toolkit

dspbptk This is dspbptk, the Dyson Sphere Program Blueprint toolkit. Dyson Sphere Program is an amazing factory-building game by the incredibly talent

Johannes Bauer 22 Nov 15, 2022
Cloud Native sample microservices showcasing Full Stack Observability using AppDynamics and ThousandEyes

Cloud Native Sample Bookinfo App Observability Bookinfo is a sample application composed of four Microservices written in different languages.

Cisco DevNet 13 Jul 21, 2022
Add any Program in any language you like or add a hello world Program ❣️ if you like give us :star:

Welcome to the Hacktoberfest 2018 Hello-world 📋 This Project aims to help you to get started with using Github. You can find a tutorial here What is

Aniket Sharma 1.5k Nov 16, 2022
Convert Roman numerals to modern numerals and vice-versa

Roman Numeral Conversion Utilities This is a utility module for converting from and to Roman numerals. It supports numbers upto 3,999,999, using the v

Fictive Kin 1 Dec 17, 2021
The code behind sqlfmt.com, a web UI for sqlfmt

The code behind sqlfmt.com, a web UI for sqlfmt

Ted Conbeer 2 Dec 14, 2022
Fiber implements an proof-of-concept Python decorator that rewrites a function

Fiber implements an proof-of-concept Python decorator that rewrites a function so that it can be paused and resumed (by moving stack variables to a heap frame and adding if statements to simulate jum

Tyler Hou 225 Dec 13, 2022
A bot to use in a pump & dump event

A bot to use in a pump & dump event on Binance.com. Please note the bot is in heavy devleopment currently so be aware of errors. If you experience err

Freddie Jonas 189 Dec 24, 2022
Python-Course-V1 - This Repo contains a series of Python Jupyter Notebooks and assignments

This Repo contains a series of Python Jupyter Notebooks and assignments. The assignments are taken from Python Crash Course book by Eric Matthes.

2 Nov 15, 2022
Supply Chain will be a SAAS platfom to provide e-logistic facilites with most optimal

Shipp It Welcome To Supply Chain App [ Shipp It ] In "Shipp It" we are creating a full solution[web+app] for a entire supply chain from receiving orde

SAIKAT_CLAW 25 Dec 26, 2022
This interactive script demonstrates the Menezes-Vanstone-EC-Cryptosystem

Menezes-Vanstone-EC-Cryptosystem This interactive script demonstrates the Meneze

Nishaant Goswamy 1 Jan 02, 2022
Google Foobar challenge solutions from my experience and other's on the web.

Google Foobar challenge Google Foobar challenge solutions from my experience and other's on the web. Note: Problems indicated with "Mine" are tested a

Islam Ayman 6 Jan 20, 2022
Model Quantization Benchmark

MQBench Update V0.0.2 Fix academic prepare setting. More deployable prepare process. Fix setup.py. Fix deploy on SNPE. Fix convert_deploy bug. Add Qua

500 Jan 06, 2023
Developing and Comparing Vision-based Algorithms for Vision-based Agile Flight

DodgeDrone: Vision-based Agile Drone Flight (ICRA 2022 Competition) Would you like to push the boundaries of drone navigation? Then participate in the

Robotics and Perception Group 115 Dec 10, 2022
A collection of design patterns and idioms in Python (With tests!).

Python Patterns Help the project financially: Donate: https://smartlegion.github.io/donate/ Yandex Money: https://yoomoney.ru/to/4100115206129186 PayP

5 Sep 12, 2021
Goddard A collection of small, simple strategies for Freqtrade

Goddard A collection of small, simple strategies for Freqtrade. Simply add the strategy you choose in your strategies folder and run. ⚠️ General Crypt

Shane Jones 118 Dec 14, 2022
Automatização completa do site https://blaze.com

PyBlaze Pyblaze possibilita o acesso a api do site blaze utilizando python, retornando os últimos resultados de crashs e doubles. Agora também é possí

Cleiton Leonel 24 Dec 30, 2022
🤖️ Plugin for Sentry which allows sending notification via DingTalk robot.

Sentry DingTalk Sentry 集成钉钉机器人通知 Requirments sentry = 21.5.1 特性 发送异常通知到钉钉 支持钉钉机器人webhook设置关键字 配置环境变量 DINGTALK_WEBHOOK: Optional(string) DINGTALK_CUST

1 Nov 04, 2021
Xbps-install wrapper written in Python that doesn't care about case sensitiveness and package versions

xbi Xbps-install wrapper written in Python that doesn't care about case sensitiveness and package versions. Description This Python script can be easi

Emanuele Sabato 5 Apr 11, 2022
A Modern Fetch Tool for Linux!

Ufetch A Modern Fetch Tool for Linux! Programming Language: Python IDE: Visual Studio Code Developed by Avishek Dutta If you get any kind of problem,

Avishek Dutta 7 Dec 12, 2021