From d25651cb52f98bb3160d896a516066d216dc8c01 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
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 6013ea452..541ea7679 100644
--- a/starlette/routing.py
+++ b/starlette/routing.py
@@ -1,14 +1,10 @@
from __future__ import annotations
-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: type[BaseException] | None,
- exc_value: BaseException | None,
- traceback: types.TracebackType | None,
- ) -> bool | None:
- 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 5ccf4ca61..879f6e69c 100644
--- a/tests/test_applications.py
+++ b/tests/test_applications.py
@@ -359,59 +359,6 @@ async def lifespan(app: ASGIApp) -> AsyncGenerator[None, None]:
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: TestClientFactory) -> None:
- startup_complete = False
- cleanup_complete = False
-
- async def lifespan(app: ASGIApp) -> AsyncGenerator[None, None]:
- nonlocal startup_complete, cleanup_complete
- startup_complete = True
- yield
- cleanup_complete = True
-
- app = Starlette(lifespan=lifespan) # type: ignore
-
- 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: TestClientFactory) -> None:
- startup_complete = False
- cleanup_complete = False
-
- def lifespan(app: ASGIApp) -> Generator[None, None, None]:
- nonlocal startup_complete, cleanup_complete
- startup_complete = True
- yield
- cleanup_complete = True
-
- app = Starlette(lifespan=lifespan) # type: ignore
-
- 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: TestClientFactory) -> None:
class NoOpMiddleware:
def __init__(self, app: ASGIApp):