Coverage for src/lexigram/web/middleware/compression.py: 17%

1"""Compression Middleware.""" 

2 

3from __future__ import annotations 

4 

5import gzip 

6from typing import Any 

7 

8from starlette.types import ASGIApp, Receive, Scope, Send 

9 

10from lexigram.logging import get_logger 

11 

12logger = get_logger(__name__) 

13 

14 

15class CompressionMiddleware: 

16 """Response compression middleware""" 

17 

18 def __init__( 

19 self, 

20 app: ASGIApp, 

21 minimum_size: int = 1024, 

22 compress_types: list[str] | None = None, 

23 ): 

24 self.app = app 

25 self.minimum_size = minimum_size 

26 self.compress_types = compress_types or [ 

27 "text/*", 

28 "application/json", 

29 "application/javascript", 

30 ] 

31 

32 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

33 """Process request and compress response if appropriate.""" 

34 

35 if scope["type"] != "http": 

36 await self.app(scope, receive, send) 

37 return 

38 

39 # Check if client supports compression 

40 headers = dict(scope.get("headers", [])) 

41 accept_encoding_bytes = headers.get(b"accept-encoding") 

42 accept_encoding = ( 

43 accept_encoding_bytes.decode("latin-1") if accept_encoding_bytes else "" 

44 ) 

45 client_supports_gzip = "gzip" in accept_encoding 

46 

47 if not client_supports_gzip: 

48 await self.app(scope, receive, send) 

49 return 

50 

51 # Track response for compression 

52 response_started = False 

53 response_status = 200 

54 response_headers = [] 

55 response_body = b"" 

56 content_type = "" 

57 

58 async def send_compressed(message: Any) -> None: 

59 nonlocal \ 

60 response_started, \ 

61 response_status, \ 

62 response_headers, \ 

63 response_body, \ 

64 content_type 

65 

66 if message["type"] == "http.response.start": 

67 response_started = True 

68 response_status = message.get("status", 200) 

69 response_headers = list(message.get("headers", [])) 

70 

71 # Get content type 

72 for key, value in response_headers: 

73 if key == b"content-type": 

74 content_type = value.decode("latin-1") 

75 break 

76 

77 elif message["type"] == "http.response.body": 

78 if not response_started: 

79 logger.warning( 

80 "compression_body_before_start", 

81 detail="http.response.body received before http.response.start — forwarding uncompressed", 

82 ) 

83 await send(message) 

84 return 

85 

86 response_body += message.get("body", b"") 

87 

88 # If this is the final message (no more body expected) 

89 if not message.get("more_body", False): 

90 await self._compress_and_send( 

91 send, 

92 response_status, 

93 response_headers, 

94 response_body, 

95 content_type, 

96 ) 

97 # Don't send yet if more body is coming 

98 else: 

99 await send(message) 

100 

101 await self.app(scope, receive, send_compressed) 

102 

103 async def _compress_and_send( 

104 self, 

105 send: Send, 

106 status: int, 

107 headers: list, 

108 body: bytes, 

109 content_type: str, 

110 ) -> None: 

111 """Compress response body and send.""" 

112 

113 # Check content type 

114 should_compress = any( 

115 pattern in content_type or pattern.replace("*", "") in content_type 

116 for pattern in self.compress_types 

117 ) 

118 

119 if not should_compress or len(body) < self.minimum_size: 

120 # Send uncompressed 

121 await send( 

122 { 

123 "type": "http.response.start", 

124 "status": status, 

125 "headers": headers, 

126 }, 

127 ) 

128 await send( 

129 { 

130 "type": "http.response.body", 

131 "body": body, 

132 }, 

133 ) 

134 return 

135 

136 # Compress body 

137 compressed = gzip.compress(body) 

138 

139 # Update headers 

140 headers = list(filter(lambda kv: kv[0] != b"content-length", headers)) 

141 headers.append((b"content-encoding", b"gzip")) 

142 headers.append((b"content-length", str(len(compressed)).encode())) 

143 

144 # Send compressed response 

145 await send( 

146 { 

147 "type": "http.response.start", 

148 "status": status, 

149 "headers": headers, 

150 }, 

151 ) 

152 await send( 

153 { 

154 "type": "http.response.body", 

155 "body": compressed, 

156 }, 

157 )