Add lifespan documentation (#1759)

Contributes towards https://github.com/spec-first/connexion/issues/1531
This commit is contained in:
Robbe Sneyders
2023-10-29 09:40:16 +01:00
committed by GitHub
parent b9ba13cde5
commit 77a63a07ef
4 changed files with 113 additions and 2 deletions

View File

@@ -112,6 +112,4 @@ class ExceptionMiddleware(StarletteExceptionMiddleware):
)
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
# Needs to be set so starlette router throws exceptions instead of returning error responses
scope["app"] = "connexion"
await super().__call__(scope, receive, send)

View File

@@ -477,4 +477,7 @@ class ConnexionMiddleware:
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if self.middleware_stack is None:
self.app, self.middleware_stack = self._build_middleware_stack()
# Set so starlette router throws exceptions instead of returning error responses
# This instance is also passed to any lifespan handler
scope["app"] = self
await self.app(scope, receive, send)

View File

@@ -71,6 +71,7 @@ Documentation
validation
security
context
lifespan
cookbook
exceptions
cli

109
docs/lifespan.rst Normal file
View File

@@ -0,0 +1,109 @@
Lifespan
========
You can register lifespan handlers to run code before the app starts, or after it shuts down.
This ideal for setting up and tearing down database connections or machine learning models for
instance.
.. tab-set::
.. tab-item:: AsyncApp
:sync: AsyncApp
.. code-block:: python
import contextlib
import typing
from connexion import AsyncApp, ConnexionMiddleware, request
@contextlib.asynccontextmanager
def lifespan_handler(app: ConnexionMiddleware) -> typing.AsyncIterator:
"""Called at startup and shutdown, can yield state which will be available on the
request."""
client = Client()
yield {"client": client}
client.close()
def route():
"""Endpoint function called when receiving a request, you can access the state
on the request here."""
client = request.state.client
client.call()
app = AsyncApp(__name__, lifespan=lifespan_handler)
.. tab-item:: FlaskApp
:sync: FlaskApp
.. code-block:: python
import contextlib
import typing
from connexion import FlaskApp, ConnexionMiddleware, request
@contextlib.asynccontextmanager
def lifespan_handler(app: ConnexionMiddleware) -> typing.AsyncIterator:
"""Called at startup and shutdown, can yield state which will be available on the
request."""
client = Client()
yield {"client": client}
client.close()
def route():
"""Endpoint function called when receiving a request, you can access the state
on the request here."""
client = request.state.client
client.call()
app = FlaskApp(__name__, lifespan=lifespan_handler)
.. tab-item:: ConnexionMiddleware
:sync: ConnexionMiddleware
.. code-block:: python
import contextlib
import typing
from asgi_framework import App
from connexion import ConnexionMiddleware, request
@contextlib.asynccontextmanager
def lifespan_handler(app: ConnexionMiddleware) -> typing.AsyncIterator:
"""Called at startup and shutdown, can yield state which will be available on the
request."""
client = Client()
yield {"client": client}
client.close()
def endpoint():
"""Endpoint function called when receiving a request, you can access the state
on the request here."""
client = request.state.client
client.call()
app = App(__name__)
app = ConnexionMiddleware(app, lifespan=lifespan_handler)
Running lifespan in tests
-------------------------
If you want lifespan handlers to be called during tests, you can use the ``test_client`` as a
context manager.
.. code-block:: python
def test_homepage():
app = ... # Set up app
with app.test_client() as client:
# Application's lifespan is called on entering the block.
response = client.get("/")
assert response.status_code == 200
# And the lifespan's teardown is run when exiting the block.
For more information, please refer to the `Starlette documentation`_.
.. _Starlette documentation: https://www.starlette.io/lifespan/