Coverage for src / lexigram / contracts / web / http_models.py: 78%

1"""Framework-owned HTTP response model. 

2 

3Decouples consumers from ``aiohttp`` (or any other HTTP library) by providing 

4a transport-agnostic response object. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from typing import Any 

11 

12 

13@dataclass(frozen=True) 

14class HttpResponse: 

15 """Framework-owned representation of an HTTP response. 

16 

17 Wraps the raw response produced by the underlying HTTP library (e.g. 

18 ``aiohttp``) so that consumers never need to import transport-level 

19 objects. 

20 

21 Attributes: 

22 status: HTTP status code (e.g. 200, 404). 

23 headers: Response headers as a plain ``dict[str, str]``. 

24 body: Raw response body bytes. 

25 text: Response body decoded as UTF-8 text; empty string when absent. 

26 json: Parsed JSON payload; ``None`` when the response is not JSON. 

27 url: Final (possibly redirected) URL as a string. 

28 method: HTTP method of the originating request (upper-cased). 

29 content_length: Parsed ``Content-Length`` response header value, or 

30 ``None`` when absent or non-numeric. 

31 """ 

32 

33 status: int 

34 headers: dict[str, str] = field(default_factory=dict) 

35 body: bytes = b"" 

36 text: str = "" 

37 json: Any = None 

38 url: str = "" 

39 method: str = "" 

40 content_length: int | None = None 

41 

42 def raise_for_status(self) -> None: 

43 """Raise :class:`HttpStatusError` when the status code indicates failure. 

44 

45 Raises: 

46 HttpStatusError: For any 4xx or 5xx status code. 

47 """ 

48 if self.status >= 400: 

49 raise HttpStatusError( 

50 f"HTTP {self.status} for {self.method} {self.url}", 

51 status=self.status, 

52 response=self, 

53 ) 

54 

55 @property 

56 def Ok(self) -> bool: 

57 """``True`` when the status code is below 400.""" 

58 return self.status < 400 

59 

60 

61from lexigram.contracts.exceptions.infra import InfrastructureError 

62 

63 

64class HttpStatusError(InfrastructureError): 

65 """Raised by :meth:`HttpResponse.raise_for_status` for 4xx/5xx responses. 

66 

67 Attributes: 

68 status: HTTP status code that triggered this error. 

69 response: The originating :class:`HttpResponse` instance. 

70 """ 

71 

72 _code: str = "LEX_ERR_WEB_002" 

73 

74 def __init__( 

75 self, 

76 message: str, 

77 *, 

78 status: int, 

79 response: HttpResponse, 

80 **kwargs: Any, 

81 ) -> None: 

82 super().__init__( 

83 message=message, 

84 code=f"HTTP_{status}", 

85 details={"status": status, "url": response.url, "method": response.method}, 

86 **kwargs, 

87 ) 

88 self.status = status 

89 self.response = response 

90 

91 

92__all__ = ["HttpResponse", "HttpStatusError"]