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
Simple web application, which has a single endpoint, dedicated to annotation parsing and convertion.

Simple web application, which has a single endpoint, dedicated to annotation parsing and conversion.

Pavel Paranin 1 Nov 01, 2021
Python package that mirrors the original Nodejs ReplAPI-It.

Python-ReplAPI-It Python package that mirrors the original Nodejs ReplAPI-It. Contributing First fork the repo: $ git clone https://github.com/ReplAPI

The ReplAPI.it Project 10 Jun 05, 2022
Developing a python based app prototype with KivyMD framework for a competition :))

Developing a python based app prototype with KivyMD framework for a competition :))

Jay Desale 1 Jan 10, 2022
March-madness - March Madness results 1985-2021

march-madness Results for all 2,268 NCAA Division I Men's Basketball Tournament games since the modern format was introduced in 1985. Includes years,

Darik Harter 2 Feb 26, 2022
🌲 Um simples criador de arvore de items feito em Python para o Prompt 🐍

Esse projeto foi feito em Python com, intuito de fortificar meu aprendizado de programação. Sobre • Tecnologias • Pré Requisitos • Licença • Autor 📄

Kawan Henrique 1 Aug 02, 2021
Add your recently blog and douban states in your GitHub Profile

Add your recently blog and douban states in your GitHub Profile

Bingjie Yan 4 Dec 12, 2022
Python: Wrangled and unpivoted gaming datasets. Tableau: created dashboards - Market Beacon and Player’s Shopping Guide.

Created two information products for GameStop. Using Python, wrangled and unpivoted datasets, and created Tableau dashboards.

Zinaida Dvoskina 2 Jan 29, 2022
Example teacher bot for deployment to Chai app.

Create and share your own chatbot Here is the code for uploading the popular "Ms Harris (Teacher)" chatbot to the Chai app. You can tweak the config t

Chai 1 Jan 10, 2022
How to create the game Rock, Paper, Scissors in Python

Rock, Paper, Scissors! If you want to learn how to do interactive games using Python, then this is great start for you. In this code, You will learn h

SplendidSpidey 1 Dec 18, 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
A python script that fetches the grades of a student from a WAEC result in pdf format.

About waec-result-analyzer A python script that fetches the grades of a student from a WAEC result in pdf format. Built for federal government college

Oshodi Kolapo 2 Dec 04, 2021
Discover and load entry points from installed packages

Entry points are a way for Python packages to advertise objects with some common interface. The most common examples are console_scripts entry points,

Thomas Kluyver 69 Jul 05, 2022
Odoo modules related to website/webshop

Website Apps related to Odoo it's website/webshop features: webshop_public_prices: allow configuring to hide or show product prices and add to cart bu

Yenthe Van Ginneken 9 Nov 04, 2022
🦠 A simple and fast (< 200ms) API for tracking the global coronavirus (COVID-19, SARS-CoV-2) outbreak.

🦠 A simple and fast ( 200ms) API for tracking the global coronavirus (COVID-19, SARS-CoV-2) outbreak. It's written in python using the 🔥 FastAPI framework. Supports multiple sources!

Marius 1.6k Jan 04, 2023
dragmap-meth: Fast and accurate aligner for bisulfite sequencing reads using dragmap

dragmap_meth (dragmap_meth.py) Alignment of BS-Seq reads using dragmap. Intro This works for single-end reads and for paired-end reads from the direct

Shaojun Xie 3 Jul 14, 2022
Bu repoda python ile CAN-Bus çalışmalarını nasıl gerçekleyeceğiniz anlatılmaktadır.

CAN-Bus-with-Python "CAN Bus 1980'li yıllarda Robert BOSCH tarafından geliştirilmiş bir iletişim protokoldür. Hızlı ve hata oranının çok düşük olması

Yunus Emre Coşkun 16 Aug 29, 2022
Random Turkish name generator with realistic probabilities.

trnames Random Turkish name generator with realistic probabilities. Based on Trey Hunner's names package. Installation The package can be installed us

Kaan Öztürk 20 Jan 02, 2023
A script to automatically update bot status at GitHub as well as in Telegram channel.

A simple & short repository to show your bot's status in your GitHub README.md file as well as in you channel.

Jainam Oswal 55 Dec 13, 2022
Solve various integral equations using numerical methods in Python

Solve Volterra and Fredholm integral equations This Python package estimates Volterra and Fredholm integral equations using known techniques. Installa

Matthew Wildrick Thomas 18 Nov 28, 2022
A simple solution for water overflow problem in Python

Water Overflow problem There is a stack of water glasses in a form of triangle as illustrated. Each glass has a 250ml capacity. When a liquid is poure

Kris 2 Oct 22, 2021