From 20daacd48805d01f68e37adb65d92055979beb98 Mon Sep 17 00:00:00 2001 From: Marcelo Trylesinski Date: Sun, 24 Dec 2023 12:01:37 +0100 Subject: [PATCH] Remove deprecated generator function warnings on lifespan --- starlette/routing.py | 53 -------------------------------------- tests/test_applications.py | 53 -------------------------------------- 2 files changed, 106 deletions(-) diff --git a/starlette/routing.py b/starlette/routing.py index 6c8dde018..bafcb0f2c 100644 --- a/starlette/routing.py +++ b/starlette/routing.py @@ -1,12 +1,8 @@ -import contextlib import functools import inspect import re import traceback -import types import typing -import warnings -from contextlib import asynccontextmanager from enum import Enum from starlette._exception_handler import wrap_app_handling_exceptions @@ -557,36 +553,6 @@ def __repr__(self) -> str: _T = typing.TypeVar("_T") -class _AsyncLiftContextManager(typing.AsyncContextManager[_T]): - def __init__(self, cm: typing.ContextManager[_T]): - self._cm = cm - - async def __aenter__(self) -> _T: - return self._cm.__enter__() - - async def __aexit__( - self, - exc_type: typing.Optional[typing.Type[BaseException]], - exc_value: typing.Optional[BaseException], - traceback: typing.Optional[types.TracebackType], - ) -> typing.Optional[bool]: - return self._cm.__exit__(exc_type, exc_value, traceback) - - -def _wrap_gen_lifespan_context( - lifespan_context: typing.Callable[ - [typing.Any], typing.Generator[typing.Any, typing.Any, typing.Any] - ], -) -> typing.Callable[[typing.Any], typing.AsyncContextManager[typing.Any]]: - cmgr = contextlib.contextmanager(lifespan_context) - - @functools.wraps(cmgr) - def wrapper(app: typing.Any) -> _AsyncLiftContextManager[typing.Any]: - return _AsyncLiftContextManager(cmgr(app)) - - return wrapper - - class _DefaultLifespan: def __init__(self, router: "Router"): self._router = router @@ -619,25 +585,6 @@ def __init__( if lifespan is None: self.lifespan_context: Lifespan[typing.Any] = _DefaultLifespan(self) - - elif inspect.isasyncgenfunction(lifespan): - warnings.warn( - "async generator function lifespans are deprecated, " - "use an @contextlib.asynccontextmanager function instead", - DeprecationWarning, - ) - self.lifespan_context = asynccontextmanager( - lifespan, - ) - elif inspect.isgeneratorfunction(lifespan): - warnings.warn( - "generator function lifespans are deprecated, " - "use an @contextlib.asynccontextmanager function instead", - DeprecationWarning, - ) - self.lifespan_context = _wrap_gen_lifespan_context( - lifespan, - ) else: self.lifespan_context = lifespan diff --git a/tests/test_applications.py b/tests/test_applications.py index 73f2898f9..f225873c3 100644 --- a/tests/test_applications.py +++ b/tests/test_applications.py @@ -355,59 +355,6 @@ async def lifespan(app): assert cleanup_complete -deprecated_lifespan = pytest.mark.filterwarnings( - r"ignore" - r":(async )?generator function lifespans are deprecated, use an " - r"@contextlib\.asynccontextmanager function instead" - r":DeprecationWarning" - r":starlette.routing" -) - - -@deprecated_lifespan -def test_app_async_gen_lifespan(test_client_factory): - startup_complete = False - cleanup_complete = False - - async def lifespan(app): - nonlocal startup_complete, cleanup_complete - startup_complete = True - yield - cleanup_complete = True - - app = Starlette(lifespan=lifespan) - - assert not startup_complete - assert not cleanup_complete - with test_client_factory(app): - assert startup_complete - assert not cleanup_complete - assert startup_complete - assert cleanup_complete - - -@deprecated_lifespan -def test_app_sync_gen_lifespan(test_client_factory): - startup_complete = False - cleanup_complete = False - - def lifespan(app): - nonlocal startup_complete, cleanup_complete - startup_complete = True - yield - cleanup_complete = True - - app = Starlette(lifespan=lifespan) - - assert not startup_complete - assert not cleanup_complete - with test_client_factory(app): - assert startup_complete - assert not cleanup_complete - assert startup_complete - assert cleanup_complete - - def test_middleware_stack_init(test_client_factory: Callable[[ASGIApp], httpx.Client]): class NoOpMiddleware: def __init__(self, app: ASGIApp):