Coverage for src/data_api/core/upstream.py: 100%

1"""Classify expected failures at live upstream boundaries.""" 

2 

3import json 

4from collections.abc import AsyncIterator, Iterator 

5from contextlib import asynccontextmanager, contextmanager 

6from ssl import SSLContext 

7 

8import httpx 

9from pydantic import ValidationError 

10 

11from data_api.core.exceptions import ( 

12 UpstreamException, 

13 UpstreamResponseException, 

14 UpstreamTimeoutException, 

15) 

16 

17 

18@asynccontextmanager 

19async def live_client( 

20 client: httpx.AsyncClient | None, *, verify: SSLContext | bool = True 

21) -> AsyncIterator[httpx.AsyncClient]: 

22 """Borrow a runtime client, or own one for a standalone service call.""" 

23 if client is not None: 

24 yield client 

25 else: 

26 async with httpx.AsyncClient(verify=verify) as owned: 

27 yield owned 

28 

29 

30@contextmanager 

31def upstream_errors() -> Iterator[None]: 

32 try: 

33 yield 

34 except httpx.TimeoutException as exc: 

35 raise UpstreamTimeoutException("Live upstream request timed out") from exc 

36 except (httpx.HTTPStatusError, httpx.RequestError) as exc: 

37 raise UpstreamException("Live upstream request failed") from exc 

38 except (json.JSONDecodeError, UnicodeDecodeError, ValidationError) as exc: 

39 raise UpstreamResponseException("Live upstream response could not be parsed") from exc