• Stars
    star
    157
  • Rank 238,399 (Top 5 %)
  • Language
    Python
  • License
    MIT License
  • Created about 5 years ago
  • Updated over 1 year ago

Reviews

There are no reviews yet. Be the first to send feedback to the community and the maintainers!

Repository Details

Programmatic startup/shutdown of ASGI apps.

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==2.*'

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 contextlib import asynccontextmanager
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, ...

@asynccontextmanager
async def lifespan(app):
    print("Starting up!")
    yield
    print("Shutting down!")

app = Starlette(lifespan=lifespan)

async def main():
    async with LifespanManager(app) as manager:
        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
from contextlib import asynccontextmanager
import httpx
import pytest
import pytest_asyncio
from asgi_lifespan import LifespanManager
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route


@pytest_asyncio.fixture
async def app():
    @asynccontextmanager
    async def lifespan(app):
        print("Starting up")
        yield
        print("Shutting down")

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

    app = Starlette(
        routes=[Route("/", home)],
        lifespan=lifespan,
    )

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


@pytest_asyncio.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 =======================

Accessing state

LifespanManager provisions a lifespan state which persists data from the lifespan cycle for use in request/response handling.

For your app to be aware of it, be sure to use manager.app instead of the app itself when inside the context manager.

For example if using HTTPX as an async test client:

async with LifespanManager(app) as manager:
    async with httpx.AsyncClient(app=manager.app) as client:
        ...

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) as manager:
    # '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.

Yields

  • manager (LifespanManager): the LifespanManager itself. In case you use lifespan state, use async with LifespanManager(app) as manager: ... then access manager.app to get a reference to the state-aware app.

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

More Repositories

1

awesome-asgi

A curated list of awesome ASGI servers, frameworks, apps, libraries, and other resources
Python
1,303
star
2

djangorestframework-api-key

πŸ” API key permissions for Django REST Framework
Python
578
star
3

aiometer

A Python concurrency scheduling library, compatible with asyncio and trio.
Python
252
star
4

msgpack-asgi

Drop-in MessagePack support for ASGI applications and frameworks
Python
137
star
5

arel

Lightweight browser hot reload for Python ASGI web apps
Python
116
star
6

kafka-fraud-detector

🚨 Simple, self-contained fraud detection system built with Apache Kafka and Python
Python
77
star
7

httpx-sse

Consume Server-Sent Event (SSE) messages with HTTPX
Python
46
star
8

asgi-htmx

HTMX integration for ASGI applications
Python
39
star
9

starlette-auth-toolkit

Authentication backends and helpers for Starlette-based ASGI apps and frameworks
Python
32
star
10

www

Code for https://florimond.dev
Python
30
star
11

ddtrace-asgi

Unofficial Datadog tracing integration for ASGI apps and frameworks
Python
30
star
12

proxyx

[No maintenance intended] Proof of concept lightweight HTTP/1.1 proxy service built with ASGI and HTTPX.
Python
29
star
13

asgi-sitemaps

Sitemap generation for Python ASGI web apps
Python
20
star
14

asgi-caches

Server-side HTTP caching for ASGI applications, inspired by Django's cache framework
Python
18
star
15

pyboids

A boids flocking behaviour algorithm implementation in Python and Pygame
Python
14
star
16

python-in-browser

🐍πŸ›₯🌟 Running Python in the browser with Batavia and Starlette
Python
12
star
17

dataclasses-properties

🐍🀝 Reconciling Python's dataclasses and properties
Python
11
star
18

ng-courses

Fetch and display a list of courses by implementing the Model-Adapter Pattern
TypeScript
10
star
19

httpxprof

A tool for profiling HTTPX using cProfile and SnakeViz
Python
9
star
20

all-my-repos

Apply changes across all my repos using https://github.com/asottile/all-repos.
Python
7
star
21

personal

πŸ–₯ Personal blog frontend app
TypeScript
7
star
22

cs-ir

Implementation of an Information Retrieval System (IRS)
Jupyter Notebook
5
star
23

limier

Smart conversion and validation toolkit powered by type annotations
Python
4
star
24

paperiano

Paper + Piano (+ Computer Vision + Deep Learning) = Paperiano
Jupyter Notebook
4
star
25

personal-vue

β›° Vue remake of my personal blog frontend app. Built for learning purposes.
Vue
4
star
26

subscriptions-transport-ws-python

Pure Python, asynchronous, event-loop-agnostic implementation of the subscriptions-transport-ws protocol
Python
4
star
27

azure-pipelines-templates

Azure Pipelines templates for my repos
3
star
28

azp-python-example

Opinionated example Python package Azure Pipelines setup w/ TestPyPI publish on tags
Shell
2
star
29

roller-coaster-loops

Calculus applied to roller coaster physics
Python
2
star
30

datacenters-drought

Data visualization of datacenters location and water resources vs US West 2021 extreme droughts
Python
2
star
31

personal-api

⚑️ Personal blog API
Python
2
star
32

httpx-unixsocket-poc

POC third-party package for adding Unix Domain Socket support to HTTPX
Python
2
star
33

fetch-metadata-asgi

PoC ASGI middleware implementation of the Fetch Metadata specification
Python
2
star
34

pygame-assets

Lightweight asset management for Pygame
Python
2
star
35

ansible-learn

Learning Ansible by using it to configure a Nginx + Python + Uvicorn + Gunicorn + Supervisor deployment in a Vagrant VM.
Makefile
2
star
36

django-clean-media

Unused media files cleaning plugin for Django
Python
1
star
37

aragon

Rule-based streaming data cleansing application written in Python
Python
1
star
38

dots

Small game made with Love2D
Lua
1
star
39

pytest-unasync

Pytest plugin for async-to-sync test generation
Python
1
star
40

python-package-template

My opinionated cookiecutter template for Python packages.
Python
1
star
41

uptv

πŸ“Ί TV show alerting web app
Python
1
star
42

solitaire-rs

A Rust terminal UI (TUI) implementation of Solitaire
Rust
1
star
43

ddtrace-examples

Playing around with Datadog APM
Python
1
star
44

simulations

Fun mathematical simulations, using Python
Python
1
star
45

realchat

A web-based real-time chat room application made with Go, Svelte, and SocketIO.
Svelte
1
star
46

tower-defense

A cool Python game about defending a tower. All your base are belong to us.
Python
1
star
47

bocadillo-ws-test

Debugging WebSocket connection closures with Bocadillo and Uvicorn
Python
1
star