Coverage for src/lexigram/web/routing/versioning.py: 47%

1""" 

2API Versioning Support for Lexigram Web. 

3 

4Supports multiple versioning strategies: 

5- URI versioning: /v1/users 

6- Header versioning: X-API-Version: 1 

7- Media type versioning: Accept: application/vnd.api.v1+json 

8""" 

9 

10from __future__ import annotations 

11 

12from dataclasses import dataclass, field 

13from enum import StrEnum 

14from typing import TYPE_CHECKING, Any 

15 

16if TYPE_CHECKING: 

17 from collections.abc import Awaitable, Callable 

18 

19 from starlette.requests import Request 

20 from starlette.responses import Response 

21 

22 

23class VersioningStrategy(StrEnum): 

24 """API versioning strategy.""" 

25 

26 HEADER = "header" 

27 URI = "uri" 

28 MEDIA_TYPE = "media_type" 

29 QUERY = "query" 

30 

31 

32@dataclass 

33class VersioningConfig: 

34 """Configuration for API versioning.""" 

35 

36 strategy: VersioningStrategy = VersioningStrategy.URI 

37 default_version: str = "1" 

38 header_name: str = "X-API-Version" 

39 uri_prefix: str = "v" 

40 media_type_prefix: str = "vnd.api" 

41 query_param: str = "api_version" 

42 

43 

44class VersionExtractor: 

45 """Extracts version from requests based on strategy.""" 

46 

47 def __init__(self, config: VersioningConfig): 

48 self.config = config 

49 

50 def extract(self, request: Request) -> str: 

51 """Extract version from request.""" 

52 if self.config.strategy == VersioningStrategy.HEADER: 

53 return self._extract_from_header(request) 

54 if self.config.strategy == VersioningStrategy.URI: 

55 return self._extract_from_uri(request) 

56 if self.config.strategy == VersioningStrategy.MEDIA_TYPE: 

57 return self._extract_from_media_type(request) 

58 # QUERY or any future strategy 

59 return self._extract_from_query(request) 

60 

61 def _extract_from_header(self, request: Request) -> str: 

62 """Extract version from header.""" 

63 version = request.headers.get(self.config.header_name) 

64 return version if version else self.config.default_version 

65 

66 def _extract_from_uri(self, request: Request) -> str: 

67 """Extract version from URI path.""" 

68 path = request.url.path 

69 parts = path.strip("/").split("/") 

70 

71 for part in parts: 

72 if part.startswith(self.config.uri_prefix): 

73 return part[len(self.config.uri_prefix) :] 

74 

75 return self.config.default_version 

76 

77 def _extract_from_media_type(self, request: Request) -> str: 

78 """Extract version from Accept header media type.""" 

79 accept = request.headers.get("Accept", "") 

80 

81 # Parse media type like: application/vnd.api.v1+json 

82 if self.config.media_type_prefix in accept: 

83 parts = accept.split(".") 

84 for part in parts: 

85 if part.startswith("v") and part[1:].replace("+", "").isdigit(): 

86 return part[1:].split("+")[0] 

87 

88 return self.config.default_version 

89 

90 def _extract_from_query(self, request: Request) -> str: 

91 """Extract version from query parameter.""" 

92 version = request.query_params.get(self.config.query_param) 

93 return version if version else self.config.default_version 

94 

95 

96def version(api_version: str) -> Callable[[Any], Any]: 

97 """ 

98 Decorator to specify controller or method version. 

99 

100 Usage:: 

101 

102 @version("1") 

103 class UsersController(Controller): 

104 ... 

105 

106 @version("2") 

107 class UsersV2Controller(Controller): 

108 ... 

109 """ 

110 

111 def decorator(target: Any) -> Any: 

112 # Store version metadata 

113 target.__api_version__ = api_version 

114 return target 

115 

116 return decorator 

117 

118 

119class VersioningMiddleware: 

120 """ 

121 Middleware to handle API versioning. 

122 

123 Extracts version from request and stores it in request state. 

124 """ 

125 

126 def __init__(self, config: VersioningConfig): 

127 self.config = config 

128 self.extractor = VersionExtractor(config) 

129 

130 async def __call__( 

131 self, 

132 request: Request, 

133 call_next: Callable[[Request], Awaitable[Response]], 

134 ) -> Response: 

135 """Process request and extract version.""" 

136 # Extract version and store in request state 

137 version = self.extractor.extract(request) 

138 request.state.api_version = version 

139 

140 # Continue processing 

141 return await call_next(request) 

142 

143 

144def get_version(request: Request) -> str: 

145 """ 

146 Get API version from request state. 

147 

148 Args: 

149 request: The HTTP request 

150 

151 Returns: 

152 API version string 

153 """ 

154 return getattr(request.state, "api_version", "1") 

155 

156 

157# --------------------------------------------------------------------------- 

158# @api_version — richer version decorator with URI prefix + deprecation 

159# --------------------------------------------------------------------------- 

160 

161 

162@dataclass 

163class ApiVersionMetadata: 

164 """Metadata attached to a versioned controller or handler by ``@api_version``.""" 

165 

166 version: int | str 

167 """Version number/string (e.g. ``1``, ``2``, ``"2.1"``).""" 

168 deprecated: bool = False 

169 """When ``True``, add ``Deprecation: true`` and ``Sunset`` response headers.""" 

170 sunset: str | None = None 

171 """ISO-8601 date after which the version will be removed (e.g. ``"2025-12-31"``).""" 

172 prefix: str | None = None 

173 """Explicit URL prefix override (e.g. ``"/api/v1"``). When *None*, the prefix 

174 is derived automatically from the version number as ``"/v{version}"``.""" 

175 _extra: dict[str, Any] = field(default_factory=dict, repr=False) 

176 

177 @property 

178 def url_prefix(self) -> str: 

179 """Return the URL prefix for this version.""" 

180 if self.prefix is not None: 

181 return self.prefix 

182 return f"/v{self.version}" 

183 

184 

185def api_version( 

186 ver: int | str, 

187 *, 

188 deprecated: bool = False, 

189 sunset: str | None = None, 

190 prefix: str | None = None, 

191) -> Callable[[Any], Any]: 

192 """Mark a controller (or individual handler) with an API version. 

193 

194 This extends the simpler :func:`version` decorator with: 

195 

196 * **URL prefix** — the version number is automatically prepended to the 

197 controller's ``prefix`` attribute (e.g. ``prefix = "/users"`` becomes 

198 ``"/v1/users"``). Override with the *prefix* argument. 

199 * **Deprecation support** — setting ``deprecated=True`` causes the 

200 routing layer to inject ``Deprecation: true`` and optionally 

201 ``Sunset: <date>`` response headers for all routes on the controller. 

202 * **Metadata storage** — an :class:`ApiVersionMetadata` instance is 

203 stored on the class/function as ``__api_version_meta__`` and the plain 

204 version string as ``__api_version__`` (compatible with 

205 :class:`VersioningMiddleware`). 

206 

207 Args: 

208 ver: Version number or string (e.g. ``1``, ``"2"``, ``"2.1"``). 

209 deprecated: When ``True``, mark this version as deprecated. 

210 sunset: Optional ISO-8601 date string indicating when support ends. 

211 prefix: Explicit URL prefix to use instead of the auto-derived one. 

212 

213 Returns: 

214 Class/function decorator. 

215 

216 Example:: 

217 

218 @api_version(1) 

219 class UserControllerV1(Controller): 

220 prefix = "/users" # mounted at /v1/users 

221 

222 @api_version(2) 

223 class UserControllerV2(Controller): 

224 prefix = "/users" # mounted at /v2/users 

225 

226 @api_version(1, deprecated=True, sunset="2025-12-31") 

227 class LegacyController(Controller): 

228 prefix = "/legacy" # adds Deprecation + Sunset headers 

229 """ 

230 meta = ApiVersionMetadata( 

231 version=ver, 

232 deprecated=deprecated, 

233 sunset=sunset, 

234 prefix=prefix, 

235 ) 

236 

237 def decorator(target: Any) -> Any: 

238 # Store rich metadata 

239 target.__api_version_meta__ = meta 

240 # Keep backward-compat plain string for VersioningMiddleware 

241 target.__api_version__ = str(ver) 

242 

243 # If this is a class (controller), automatically prepend the version 

244 # prefix to the controller's `prefix` attribute. 

245 if isinstance(target, type): 

246 existing_prefix: str = getattr(target, "prefix", "") or "" 

247 # Avoid double-prefixing if the prefix already starts with "/vN" 

248 version_prefix = meta.url_prefix 

249 if not existing_prefix.startswith(version_prefix): 

250 target.prefix = version_prefix + existing_prefix # type: ignore[attr-defined] 

251 

252 return target 

253 

254 return decorator 

255 

256 

257__all__ = [ 

258 "ApiVersionMetadata", 

259 "VersionExtractor", 

260 "VersioningConfig", 

261 "VersioningMiddleware", 

262 "VersioningStrategy", 

263 "api_version", 

264 "get_version", 

265 "version", 

266]