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
Custom python interface to xstan (a modified (cmd)stan)

Custom python interface to xstan (a modified (cmd)stan) Use at your own risk, currently everything is very brittle and will probably be changed in the

2 Dec 16, 2021
Tools for analyzing Java JVM gc log files

gc_log This package consists of two separate utilities useful for : gc_log_visualizer.py regionsize.py GC Log Visualizer This was updated to run under

Brad Schoening 0 Jan 04, 2022
This is a a CSMA/CA simulator written in Python based on simulator of the same type

This is a a CSMA/CA simulator written in Python based on simulator of the same type found the link https://github.com/StevenSLXie/CSMA-Simulator with

M. Ismail 4 Nov 22, 2022
Retrying is an Apache 2.0 licensed general-purpose retrying library, written in Python, to simplify the task of adding retry behavior to just about anything.

Retrying Retrying is an Apache 2.0 licensed general-purpose retrying library, written in Python, to simplify the task of adding retry behavior to just

Ray Holder 1.9k Dec 29, 2022
Py4J enables Python programs to dynamically access arbitrary Java objects

Py4J Py4J enables Python programs running in a Python interpreter to dynamically access Java objects in a Java Virtual Machine. Methods are called as

Barthelemy Dagenais 1k Jan 02, 2023
A python script for combining multiple native SU2 format meshes into one mesh file for multi-zone simulations.

A python script for combining multiple native SU2 format meshes into one mesh file for multi-zone simulations.

MKursatUzuner 1 Jan 20, 2022
🟥This is an overview of how to set up and use DataStore3 in your Roblox experiences

Welcome to DataStore3 👋 This is an overview of how to set up and use DataStore3 in your Roblox experiences What is it? 🤔 DataStore3 is a service tha

Reece Harris 7 Aug 19, 2022
An end-to-end encrypted chat

An end-to-end encrypted chat, that allows users to anonymously talk without ip logs, personal info, or need for registration.

Privalise 1 Nov 27, 2021
A Python package that provides physical constants.

PhysConsts A Python package that provides physical constants. The code is being developed by Marc van der Sluys of the department of Astrophysics at t

Marc van der Sluys 1 Jan 05, 2022
A random cat fact python module

A random cat fact python module

Fayas Noushad 4 Nov 28, 2021
Sublime Text 2/3 style auto completion for ST4

Hippie Autocompletion Sublime Text 2/3 style auto completion for ST4: cycle through words, do not show popup. Simply hit Tab to insert completion, hit

Alexander Schepanovski 20 May 19, 2022
Whatsapp Messenger master

Whatsapp Messenger master

Swarup Kharul 5 Nov 21, 2021
Repository for DNN training, theory to practice, part of the Large Scale Machine Learning class at Mines Paritech

DNN Training, from theory to practice This repository is complementary to the deep learning training lesson given to les Mines ParisTech on the 11th o

Alexandre Défossez 6 Nov 14, 2022
Unofficial Python Library to communicate with SESAME 3 series products from CANDY HOUSE, Inc.

pysesame3 Unofficial Python Library to communicate with SESAME 3 series products from CANDY HOUSE, Inc. This project aims to control SESAME 3 series d

Masaki Tagawa 18 Dec 12, 2022
In the works, creating a new Chess Board and way to Play...

sWJz4Chess date started on github.com 11-13-2021 In the works, creating a new Chess Board and way to Play... starting to write this in Pygame, any ind

Shawn 2 Nov 18, 2021
The-White-Noise-Project - The project creates noise intentionally

The-White-Noise-Project High quality audio matters everywhere, even in noise. Be

Ali Hakim Taşkıran 1 Jan 02, 2022
A basic python project which replicates the functionalities on an 8 Ball.

Magic-8-Ball To the people who wish to make decisions using a Magic 8 Ball but can't get one? I gotchu. This is a basic python project which replicate

3 Jun 24, 2021
Notifies server owners of mod updates, also notifies of player deaths and player joins through Discord.

ProjectZomboid-ServerAssistant Notifies server owners of mod updates, also notifies of player deaths and player joins through Discord. A Python based

3 Sep 30, 2022
Fastest Semantle solver this side of the Mississippi

semantle Fastest Semantle solver this side of the Mississippi. Roughly 3 average turns to win Measured against (part of) the word2vec-google-news-300

Frank Odom 8 Dec 26, 2022
Helps compare between New and Old Tax Regime.

Income-Tax-Calculator Helps compare between New and Old Tax Regime. Sample Console Input/Output

2 Jan 10, 2022