Coverage for src/lexigram/admin/data/timeout.py: 0%

1from __future__ import annotations 

2 

3import asyncio 

4from collections.abc import Awaitable 

5from typing import TypeVar 

6 

7T = TypeVar("T") 

8 

9 

10class AdminQueryTimeoutError(TimeoutError): 

11 """Admin data operation exceeded the configured timeout.""" 

12 

13 

14async def with_query_timeout( 

15 awaitable: Awaitable[T], 

16 *, 

17 timeout_seconds: float, 

18 operation: str, 

19) -> T: 

20 try: 

21 return await asyncio.wait_for(awaitable, timeout=timeout_seconds) 

22 except TimeoutError as exc: 

23 raise AdminQueryTimeoutError( 

24 f"admin query timed out after {timeout_seconds}s: {operation}" 

25 ) from exc