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
Height 2 LDraw With python

Height2Ldraw About This project aims to be able to make a full lego 3D model using the ldraw file format (.ldr) from a height and color map, currently

1 Dec 22, 2021
Stop python warnings, no matter what!

SHUTUP - Stop python warnings, no matter what! Sometimes you just can't mute python warnings. Use this library to solve this. Installation pip install

80 Jan 04, 2023
Minimal, super readable string pattern matching for python.

simplematch Minimal, super readable string pattern matching for python. import simplematch simplematch.match("He* {planet}!", "Hello World!") {"p

Thomas Feldmann 147 Dec 01, 2022
InverterApi - This project has been designed to take monitoring data from Voltronic, Axpert, Mppsolar PIP, Voltacon, Effekta

InverterApi - This project has been designed to take monitoring data from Voltronic, Axpert, Mppsolar PIP, Voltacon, Effekta

Josep Escobar 2 Sep 03, 2022
Files for QMC Workshop 2021

QMC Workshop 2021 This repository contains the presented slides and example files for the Quantum Monte Carlo (QMC) Workshop 5 October - 23 November,

QMCPACK 39 Nov 04, 2022
A simple interface to help lazy people like me to shutdown/reboot/sleep their computer remotely.

🦥 Lazy Helper ! A simple interface to help lazy people like me to shut down/reboot/sleep/lock/etc. their computer remotely. - USAGE If you're a lazy

MeHDI Rh 117 Nov 30, 2022
These are my solutions to Advent of Code problems.

Advent of Code These are my solutions to Advent of Code problems. If you want to join my leaderboard, the code is 540750-9589f56d. When I solve for sp

Sumner Evans 5 Dec 19, 2022
Adversarial Robustness with Non-uniform Perturbations

Adversarial Robustness with Non-uniform Perturbations This repository hosts the code to replicate experiments of the paper Adversarial Robustness with

5 May 20, 2022
Purge all transformation orientations addon for Blender 2.8 and newer versions

CTO Purge This add-on adds a new button to Blender's Transformation Orientation panel which empowers the user to purge all of his/her custom transform

MMMrqs 10 Dec 29, 2022
PyCASCLib: CASC interface for Warcraft III

PyCASCLib CASC interface for Warcraft III. This repo provides bindings for JCASC: https://github.com/DrSuperGood/JCASC Installation Jdk is required fo

2 Jun 04, 2022
poetry2nix turns Poetry projects into Nix derivations without the need to actually write Nix expressions

poetry2nix poetry2nix turns Poetry projects into Nix derivations without the need to actually write Nix expressions. It does so by parsing pyproject.t

Nix community projects 405 Dec 29, 2022
Automatically unpin old messages so you can always pin more!

PinRotate Automatically unpin old messages so you can always pin more! Installation You will need to install poetry to run this bot locally for develo

3 Sep 18, 2022
To check my COVID-19 vaccine appointment, I wrote an infinite loop that sends me a Whatsapp message hourly using Twilio and Selenium. It works on my Raspberry Pi computer.

COVID-19_vaccine_appointment To check my COVID-19 vaccine appointment, I wrote an infinite loop that sends me a Whatsapp message hourly using Twilio a

Ayyuce Demirbas 24 Dec 17, 2022
Force you (or your user) annotate Python function type hints.

Must-typing Force you (or your user) annotate function type hints. Notice: It's more like a joke, use it carefully. If you call must_typing in your mo

Konge 13 Feb 19, 2022
Coinloggr - A learning resource and social platform for the coin collecting community

Coinloggr A learning resource and social platform for the coin collecting commun

John Galiszewski 1 Jan 10, 2022
Algo próximo do ARP

ArpPY Algo parecido com o ARP-Scan. Dependencias O script necessita no mínimo ter o Python versão 3.x instalado e ter o sockets instalado. Executando

Feh's 3 Jan 18, 2022
Controller state monitor plugin for EVA ICS

eva-plugin-cmon Controller status monitor plugin for EVA ICS Monitors connected controllers status in SFA and pushes measurements into an external Inf

Altertech 1 Nov 06, 2021
Registro Online (100% Python-Mysql)

Registro elettronico scritto in python, utilizzando database Mysql e Collegando Registro elettronico scritto in PHP

Sergiy Grimoldi 1 Dec 20, 2021
An ultra fast cross-platform multiple screenshots module in pure Python using ctypes.

Python MSS from mss import mss # The simplest use, save a screen shot of the 1st monitor with mss() as sct: sct.shot() An ultra fast cross-platfo

Mickaël Schoentgen 799 Dec 30, 2022
Framework for creating efficient data processing pipelines

Aqueduct Framework for creating efficient data processing pipelines. Contact Feel free to ask questions in telegram t.me/avito-ml Key Features Increas

avito.tech 137 Dec 29, 2022