Coverage for src/pipeline_analytics/errors.py: 100%

1"""Typed errors for the pipeline-analytics API. 

2 

3Every 4xx/5xx response pipeline-analytics returns shares the same 

4``{code, message}`` body (see ``openapi/openapi.yaml``'s ``Error`` schema). 

5:func:`decode_error` turns that, plus the status code and any 

6``X-Request-Id`` header, into a single :class:`APIError` -- never a bare 

7string or a raw ``httpx.Response`` (rules/sdk-generation.md's "Client 

8shape"). 

9""" 

10 

11from __future__ import annotations 

12 

13import json 

14from collections.abc import Mapping 

15from typing import Protocol 

16 

17_REQUEST_ID_HEADER = "X-Request-Id" 

18 

19 

20class _ErrorResponse(Protocol): 

21 """What :func:`decode_error` needs from a response. 

22 

23 Both ``httpx.Response`` and the generated per-operation 

24 ``pipeline_analytics._generated.types.Response`` satisfy this 

25 structurally -- decoding works the same whether you call a wrapper 

26 method or drop down to a generated ``sync_detailed``/``asyncio_detailed`` 

27 call directly. Declared as read-only properties (rather than plain 

28 attributes) so a subtype's more specific attribute types -- 

29 ``http.HTTPStatus`` for ``status_code``, a ``MutableMapping`` for 

30 ``headers`` -- still satisfy the protocol; Protocol attribute matching 

31 is otherwise invariant. 

32 """ 

33 

34 @property 

35 def status_code(self) -> int: ... 

36 

37 @property 

38 def content(self) -> bytes: ... 

39 

40 @property 

41 def headers(self) -> Mapping[str, str]: ... 

42 

43 

44class APIError(Exception): 

45 """Raised for any pipeline-analytics response carrying an error body. 

46 

47 Attributes: 

48 status_code: The HTTP status code. 

49 code: The API's own machine-readable error code (empty string if 

50 the body wasn't the documented ``{code, message}`` shape). 

51 message: The API's human-readable error message. 

52 request_id: The ``X-Request-Id`` response header, or ``None`` if 

53 absent. 

54 """ 

55 

56 def __init__( 

57 self, 

58 status_code: int, 

59 code: str, 

60 message: str, 

61 request_id: str | None = None, 

62 ) -> None: 

63 self.status_code = status_code 

64 self.code = code 

65 self.message = message 

66 self.request_id = request_id 

67 super().__init__(str(self)) 

68 

69 def __str__(self) -> str: 

70 base = f"pipeline-analytics: {self.status_code} {self.code}: {self.message}" 

71 if self.request_id: 

72 return f"{base} (request {self.request_id})" 

73 return base 

74 

75 

76def decode_error(response: _ErrorResponse) -> APIError | None: 

77 """Build an :class:`APIError` from *response*, or return ``None`` if 

78 its status code isn't an error. 

79 

80 Decodes ``response.content`` directly rather than trusting the 

81 generated per-operation ``.parsed`` field, which is only populated for 

82 status codes the spec documents on that specific operation -- a 500 or 

83 429 that survived every retry attempt still needs an ``APIError``, and 

84 the API returns the same ``{code, message}`` shape for those too. 

85 

86 Usage:: 

87 

88 response = sync_detailed(client=client.raw, ...) 

89 if error := decode_error(response): 

90 raise error 

91 """ 

92 if response.status_code < 400: 

93 return None 

94 

95 code = "" 

96 message = "" 

97 try: 

98 body = json.loads(response.content) 

99 if isinstance(body, dict): 

100 code = str(body.get("code", "")) 

101 message = str(body.get("message", "")) 

102 except ValueError: 

103 pass 

104 

105 request_id = response.headers.get(_REQUEST_ID_HEADER) or None 

106 

107 return APIError(int(response.status_code), code, message, request_id)