(app: Starlette)
| 5920 | |
| 5921 | @asynccontextmanager |
| 5922 | async def _lifespan(app: Starlette) -> _LifespanGen[None, None]: |
| 5923 | import asyncio |
| 5924 | await _on_startup() |
| 5925 | nonlocal _rediscovery_task, _benchmark_refresh_task |
| 5926 | if upstream: |
| 5927 | _rediscovery_task = asyncio.create_task(_rediscovery_loop()) |
| 5928 | _benchmark_refresh_task = asyncio.create_task(_benchmark_refresh_loop()) |
| 5929 | try: |
| 5930 | yield |
| 5931 | finally: |
| 5932 | if _rediscovery_task is not None: |
| 5933 | _rediscovery_task.cancel() |
| 5934 | if _benchmark_refresh_task is not None: |
| 5935 | _benchmark_refresh_task.cancel() |
| 5936 | # ─── v2 lifecycle shutdown ─── |
| 5937 | try: |
| 5938 | from uncommon_route.v2_lifecycle import on_shutdown as v2_shutdown |
| 5939 | v2_shutdown() |
| 5940 | except Exception as e: |
| 5941 | logger.warning("v2 lifecycle shutdown failed: %s", e) |
| 5942 | |
| 5943 | routes = [ |
| 5944 | Route("/health", handle_health, methods=["GET"]), |
nothing calls this directly
no test coverage detected