Coverage for src/lexigram/web/routing/caching.py: 20%

1"""HTTP caching response decorators. 

2 

3Provides declarative decorators for setting HTTP cache-control semantics 

4and ETag-based conditional request handling. 

5 

6Usage:: 

7 

8 from lexigram.web.routing.caching import cache_control, etag 

9 

10 class ArticleController(Controller): 

11 

12 @get("/{id}") 

13 @cache_control(max_age=3600, public=True) 

14 async def get_article(self, id: str) -> ArticleResponse: 

15 ... 

16 

17 @get("/{id}/preview") 

18 @etag 

19 async def get_preview(self, id: str) -> ArticleResponse: 

20 # Framework auto-generates ETag from response body and 

21 # returns 304 Not Modified if If-None-Match matches. 

22 ... 

23""" 

24 

25from __future__ import annotations 

26 

27from collections.abc import Callable 

28import hashlib 

29from typing import Any, TypeVar 

30 

31from starlette.requests import Request 

32from starlette.responses import Response 

33 

34from lexigram.logging import get_logger 

35 

36logger = get_logger(__name__) 

37 

38F = TypeVar("F", bound=Callable[..., Any]) 

39 

40# Sentinel attribute written by decorators so ResponseSerializer can inspect 

41_CACHE_CONTROL_ATTR = "__http_cache_control__" 

42_ETAG_ATTR = "__http_etag__" 

43 

44 

45# --------------------------------------------------------------------------- 

46# @cache_control 

47# --------------------------------------------------------------------------- 

48 

49 

50def cache_control( 

51 *, 

52 max_age: int | None = None, 

53 s_maxage: int | None = None, 

54 public: bool = False, 

55 private: bool = False, 

56 no_cache: bool = False, 

57 no_store: bool = False, 

58 must_revalidate: bool = False, 

59 immutable: bool = False, 

60 stale_while_revalidate: int | None = None, 

61) -> Callable[[F], F]: 

62 """Declaratively set ``Cache-Control`` headers on a controller handler. 

63 

64 The generated ``Cache-Control`` value is written to the response by the 

65 ``RequestPipeline`` after handler execution. When applied, it wraps 

66 the handler and patches whatever ``Response`` is returned. 

67 

68 Args: 

69 max_age: ``max-age`` in seconds. 

70 s_maxage: ``s-maxage`` in seconds (shared/CDN caches). 

71 public: Mark the response as publicly cacheable. 

72 private: Mark the response as private (per-user). 

73 no_cache: Force revalidation before serving from cache. 

74 no_store: Disallow any caching whatsoever. 

75 must_revalidate: Require revalidation when stale. 

76 immutable: Hint that the response will never change. 

77 stale_while_revalidate: Seconds to serve stale while revalidating. 

78 

79 Example:: 

80 

81 @get("/articles") 

82 @cache_control(max_age=60, public=True) 

83 async def list_articles(self) -> list[ArticleDTO]: 

84 ... 

85 """ 

86 directives: list[str] = [] 

87 if public: 

88 directives.append("public") 

89 if private: 

90 directives.append("private") 

91 if no_store: 

92 directives.append("no-store") 

93 if no_cache: 

94 directives.append("no-cache") 

95 if must_revalidate: 

96 directives.append("must-revalidate") 

97 if immutable: 

98 directives.append("immutable") 

99 if max_age is not None: 

100 directives.append(f"max-age={max_age}") 

101 if s_maxage is not None: 

102 directives.append(f"s-maxage={s_maxage}") 

103 if stale_while_revalidate is not None: 

104 directives.append(f"stale-while-revalidate={stale_while_revalidate}") 

105 

106 header_value = ", ".join(directives) if directives else "no-cache" 

107 

108 def decorator(fn: F) -> F: 

109 import functools 

110 

111 @functools.wraps(fn) 

112 async def wrapper(*args: Any, **kwargs: Any) -> Any: 

113 result = await fn(*args, **kwargs) 

114 if isinstance(result, Response): 

115 result.headers["Cache-Control"] = header_value 

116 return result 

117 

118 wrapper.__http_cache_control__ = header_value # type: ignore[attr-defined] 

119 return wrapper # type: ignore[return-value] 

120 

121 return decorator 

122 

123 

124# --------------------------------------------------------------------------- 

125# @etag 

126# --------------------------------------------------------------------------- 

127 

128 

129def etag(fn: F) -> F: 

130 """Auto-generate and validate ETag headers for a controller handler. 

131 

132 When applied, the response body is hashed (MD5) to produce a weak ETag. 

133 If the client sends ``If-None-Match`` and it matches the computed ETag, 

134 the response is replaced with a ``304 Not Modified`` with no body. 

135 

136 Supports both strong and weak ETags (weak is the default). 

137 

138 Args: 

139 fn: The (async) handler method to decorate. 

140 

141 Example:: 

142 

143 @get("/{id}") 

144 @etag 

145 async def get_article(self, id: str) -> ArticleResponse: 

146 ... 

147 """ 

148 import functools 

149 

150 @functools.wraps(fn) 

151 async def wrapper(*args: Any, **kwargs: Any) -> Any: 

152 result = await fn(*args, **kwargs) 

153 

154 # Only decorate Response objects — if the handler returned raw data 

155 # it will be serialized later; skip ETag injection here 

156 if not isinstance(result, Response): 

157 return result 

158 

159 # Generate weak ETag from response body 

160 body: bytes | bytearray = bytes(result.body) if hasattr(result, "body") else b"" 

161 etag_value = f'W/"{hashlib.md5(body, usedforsecurity=False).hexdigest()}"' 

162 

163 result.headers["ETag"] = etag_value 

164 

165 # Check If-None-Match — requires request in scope 

166 # Starlette injects `request` as the first positional argument when 

167 # the handler is a controller method. We inspect args to locate it. 

168 request = _find_request(args, kwargs) 

169 if request is not None: 

170 client_etag = request.headers.get("If-None-Match", "") 

171 if client_etag and _etag_matches(client_etag, etag_value): 

172 return Response(status_code=304, headers={"ETag": etag_value}) 

173 

174 return result 

175 

176 wrapper.__http_etag__ = True # type: ignore[attr-defined] 

177 return wrapper # type: ignore[return-value] 

178 

179 

180def _find_request(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Request | None: 

181 """Locate the Starlette Request object inside handler arguments.""" 

182 for arg in args: 

183 if isinstance(arg, Request): 

184 return arg 

185 for val in kwargs.values(): 

186 if isinstance(val, Request): 

187 return val 

188 return None 

189 

190 

191def _etag_matches(client_header: str, server_etag: str) -> bool: 

192 """Return True if the client's If-None-Match header matches the server ETag. 

193 

194 Handles ``*`` (match-all) and comma-separated lists of tags. 

195 """ 

196 if client_header.strip() == "*": 

197 return True 

198 client_tags = {tag.strip() for tag in client_header.split(",")} 

199 # Compare both weak and strong form 

200 bare = server_etag.strip('"').lstrip("W/").strip('"') 

201 for tag in client_tags: 

202 tag_bare = tag.strip('"').lstrip("W/").strip('"') 

203 if tag_bare == bare: 

204 return True 

205 return False 

206 

207 

208__all__ = ["cache_control", "etag"]