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
A web UI for managing your 351ELEC device ROMs.

351ELEC WebUI A web UI for managing your 351ELEC device ROMs. Requirements Python 3 or Python 2.7 are required. If the ftfy package is installed, it w

Ben Phelps 5 Sep 26, 2022
Pre-1.0 door/chest sound injector for Minecraft

doorjector Pre-1.0 door/chest sound injector for Minecraft. While the game is running, doorjector hotswaps the new sounds for the old right before the

Sam 1 Nov 20, 2021
Extremely unfinished animation toolset for Blender 3.

AbraTools Alpha IMPORTANT: Code is a mess. Be careful using it in production. Bug reports, feature requests and PRs are appreciated. Download AbraTool

Abra 15 Dec 17, 2022
Hook and simulate global keyboard events on Windows and Linux.

keyboard Take full control of your keyboard with this small Python library. Hook global events, register hotkeys, simulate key presses and much more.

BoppreH 3.2k Jan 01, 2023
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
This is a Fava extension to display a grouped portfolio view in Fava for a set of Beancount accounts.

Fava Portfolio Summary This is a Fava extension to display a grouped portfolio view in Fava for a set of Beancount accounts. It can also calculate MWR

18 Dec 26, 2022
DSG - Source code for Digital Scholarship Grant project.

DSG Source code for Dr. Stephanie Tsang's Digital Scholarship Grant project. Work performed by Mr. Wang Minghao while as her Research Assistant. The s

1 Jan 04, 2022
Reference python implementation of Chia pool operations for pool operators

This repository provides a sample server written in python, which is meant to server as a basis for a Chia Pool. While this is a fully functional implementation, it requires some work in scalability

Chia Network 451 Dec 13, 2022
A code base for python programs the goal is to integrate all the useful and essential functions

Base Dev EN This GitHub will be available in French and English FR Ce GitHub sera disponible en français et en anglais Author License Screen EN 🇬🇧 D

Pikatsuto 1 Mar 07, 2022
This is a library for simulate probability theory problems specialy conditional probability

This is a library for simulate probability theory problems specialy conditional probability. It is also useful to create custom single or joint distribution with specific PMF or PDF to get probabilit

Mohamadreza Kariminejad 6 Mar 30, 2022
Ronin - Create Fud Meterpreter Payload To Hack Windows 11

Ronin - Create Fud Meterpreter Payload To Hack Windows 11

Dj4w3d H4mm4di 6 May 09, 2022
How to access and display MyEnergi data

MyEnergi-Python-Example How to access and display MyEnergi data Windows PC Install a version of Python typically 3.10 The Python code here needs addit

G6EJD 8 Nov 28, 2022
This suite consists of two different scripts, made to automate attacks against NoSQL databases.

NoSQL-Attack-Suite This suite consists of two different scripts, made to automate attacks against NoSQL databases. The first one looks for a NoSQL Aut

16 Dec 26, 2022
Simple rofi script to choose player for playerctl to execute its command

rofi-playerctl-switcher simple rofi script to choose player for playerctl to execute its command Usage copy playerSwitch.py and playerctl.sh to ~/.con

2 Jan 03, 2022
B-Pkg is a simple tool in python for installing all basic package in termux

Basic-Pkg 👉🏻 Basic-Pkg 👈🏻 B-Pkg is a simple tool in python for installing all basic package in termux This is my first tool, I hope you will like

Macgaiver 3 Oct 21, 2021
Tutorial on Tempo, Beat and Downbeat estimation

Tempo, Beat and Downbeat Estimation By Matthew E. P. Davies, Sebastian Böck and Magdalena Fuentes Resources and Jupyter Book for the ISMIR 2021 tutori

49 Nov 06, 2022
Animation picker for Audodesk Maya 2017 (or higher)

Dreamwall Picker Animation picker for Audodesk Maya 2017 (or higher) Authors: Lionel Brouyère, Olivier Evers This tool is a fork of Hotbox Designer (L

DreamWall 93 Dec 21, 2022
A python script to search for k-uniform Euclidean tilings.

k-uniform-solver A python script to search for k-uniform Euclidean tilings. This project's aim is to replicate and extend the list of k-uniform Euclid

3 Dec 06, 2022
Automates the fixing of problems reported by yamllint by parsing its output

yamlfixer yamlfixer automates the fixing of problems reported by yamllint by parsing its output. Usage This software automatically fixes some errors a

OPT Nouvelle Caledonie 26 Dec 26, 2022
A beacon generator using Cobalt Strike and a variety of tools.

Beaconator is an aggressor script for Cobalt Strike used to generate either staged or stageless shellcode and packing the generated shellcode using your tool of choice.

Capt. Meelo 441 Dec 17, 2022