Coverage for agentos/api/versioning.py: 36%
179 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 10:19 +0800
1"""
2AgentOS API Versioning — Semantic Versioning Middleware & Router
3━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
5Production-grade API versioning with:
6 - URL path versioning (default): /v1/resource, /v2/resource
7 - Header-based versioning: Accept: application/json; version=1
8 - Query parameter versioning: /resource?api_version=1
9 - Semantic version negotiation (closest match)
10 - Deprecation notices with sunset dates
11 - Automatic OpenAPI versioned docs
13Architecture:
14 VersioningMiddleware → extract version from request
15 VersionedRouter → route to correct version handler
16 DeprecationPolicy → warn/block deprecated versions
17"""
19from __future__ import annotations
21import re
22from collections.abc import Callable
23from dataclasses import dataclass
24from datetime import UTC, datetime, timedelta
25from enum import StrEnum
26from typing import Any
28from starlette.requests import Request
29from starlette.responses import JSONResponse, Response
31# ---------------------------------------------------------------------------
32# Semantic Version
33# ---------------------------------------------------------------------------
36@dataclass(order=True, frozen=True)
37class SemVer:
38 """Semantic version (major.minor.patch)."""
40 major: int
41 minor: int = 0
42 patch: int = 0
44 _PARSE_RE = re.compile(r"^v?(\d+)(?:\.(\d+)(?:\.(\d+))?)?")
46 @classmethod
47 def parse(cls, version_str: str) -> SemVer | None:
48 """Parse a version string like 'v1', '2.0', '1.2.3'."""
49 m = cls._PARSE_RE.match(version_str.strip())
50 if not m:
51 return None
52 return cls(
53 major=int(m.group(1)),
54 minor=int(m.group(2) or 0),
55 patch=int(m.group(3) or 0),
56 )
58 def is_compatible(self, other: SemVer) -> bool:
59 """Check if other version is API-compatible (same major)."""
60 return self.major == other.major
62 def __str__(self) -> str:
63 return f"v{self.major}.{self.minor}.{self.patch}"
66# ---------------------------------------------------------------------------
67# Version Extraction Strategy
68# ---------------------------------------------------------------------------
71class VersionStrategy(StrEnum):
72 """How to extract API version from requests."""
74 URL_PATH = "url_path" # /v1/resource
75 HEADER = "header" # Accept: application/json; version=1
76 QUERY_PARAM = "query_param" # /resource?api_version=1
79class VersionExtractor:
80 """Extract API version from incoming requests."""
82 STRATEGIES: list[VersionStrategy] = [
83 VersionStrategy.URL_PATH,
84 VersionStrategy.HEADER,
85 VersionStrategy.QUERY_PARAM,
86 ]
88 def __init__(self, strategies: list[VersionStrategy] | None = None):
89 self._strategies = strategies or self.STRATEGIES
91 def extract(self, request: Request) -> SemVer | None:
92 """Try each strategy in order; return first match."""
93 for strategy in self._strategies:
94 version = self._try_strategy(request, strategy)
95 if version is not None:
96 return version
97 return None
99 def _try_strategy(self, request: Request, strategy: VersionStrategy) -> SemVer | None:
100 if strategy == VersionStrategy.URL_PATH:
101 return self._from_url(request)
102 elif strategy == VersionStrategy.HEADER:
103 return self._from_header(request)
104 elif strategy == VersionStrategy.QUERY_PARAM:
105 return self._from_query(request)
106 return None
108 @staticmethod
109 def _from_url(request: Request) -> SemVer | None:
110 """Extract from /v{major}/... or /v{major}.{minor}/..."""
111 path = request.url.path.lstrip("/")
112 parts = path.split("/")
113 for i, part in enumerate(parts):
114 if part.lower().startswith("v") and part[1:].replace(".", "").isdigit():
115 return SemVer.parse(part)
116 return None
118 @staticmethod
119 def _from_header(request: Request) -> SemVer | None:
120 """Extract from Accept header or custom X-API-Version."""
121 custom = request.headers.get("X-API-Version")
122 if custom:
123 return SemVer.parse(custom)
125 accept = request.headers.get("Accept", "")
126 version_match = re.search(r"version=(\d+(?:\.\d+)*)", accept)
127 if version_match:
128 return SemVer.parse(version_match.group(1))
129 return None
131 @staticmethod
132 def _from_query(request: Request) -> SemVer | None:
133 """Extract from ?api_version=1 or ?v=2.0."""
134 for param in ("api_version", "v", "version"):
135 value = request.query_params.get(param)
136 if value:
137 return SemVer.parse(value)
138 return None
141# ---------------------------------------------------------------------------
142# Deprecation Policy
143# ---------------------------------------------------------------------------
146@dataclass
147class DeprecationInfo:
148 """Information about a deprecated API version."""
150 version: SemVer
151 sunset_date: datetime | None = None
152 migration_guide_url: str | None = None
153 message: str = "This API version is deprecated."
155 @property
156 def is_sunset(self) -> bool:
157 if self.sunset_date is None:
158 return False
159 return datetime.now(UTC) > self.sunset_date
162class DeprecationPolicy:
163 """Manage API version deprecation."""
165 def __init__(self):
166 self._deprecated: dict[SemVer, DeprecationInfo] = {}
168 def deprecate(self, version: str, sunset_days: int = 90, **kwargs) -> None:
169 """Mark a version as deprecated."""
170 semver = SemVer.parse(version)
171 if semver is None:
172 raise ValueError(f"Invalid version: {version}")
173 sunset = datetime.now(UTC) + timedelta(days=sunset_days)
174 self._deprecated[semver] = DeprecationInfo(
175 version=semver,
176 sunset_date=sunset,
177 **kwargs,
178 )
180 def is_deprecated(self, version: SemVer) -> bool:
181 return version in self._deprecated
183 def get_info(self, version: SemVer) -> DeprecationInfo | None:
184 return self._deprecated.get(version)
186 def should_block(self, version: SemVer) -> bool:
187 info = self._deprecated.get(version)
188 return info is not None and info.is_sunset
190 def list_deprecated(self) -> dict[str, dict[str, Any]]:
191 return {
192 str(v): {
193 "sunset_date": d.sunset_date.isoformat() if d.sunset_date else None,
194 "is_sunset": d.is_sunset,
195 "migration_guide": d.migration_guide_url,
196 }
197 for v, d in self._deprecated.items()
198 }
201# ---------------------------------------------------------------------------
202# Versioned Router
203# ---------------------------------------------------------------------------
206class VersionedRouter:
207 """
208 Route requests to version-specific handlers.
210 Supports semantic version negotiation:
211 - Exact match: /v1.0.0 → v1.0.0 handler
212 - Minor fallback: /v1.2.x → v1.2.0 handler
213 - Major fallback: /v1.x.x → latest v1 handler
214 """
216 def __init__(self):
217 self._handlers: dict[SemVer, Callable] = {}
218 self._default_version: SemVer | None = None
220 def register(self, version: str, handler: Callable) -> None:
221 """Register a handler for a specific version."""
222 semver = SemVer.parse(version)
223 if semver is None:
224 raise ValueError(f"Invalid version: {version}")
225 self._handlers[semver] = handler
227 def set_default(self, version: str) -> None:
228 """Set the default version when no version is specified."""
229 self._default_version = SemVer.parse(version)
231 def resolve(self, requested: SemVer) -> tuple[Callable | None, SemVer | None]:
232 """
233 Resolve a version to its handler.
234 Returns (handler, actual_version) or (None, None).
235 """
236 # Exact match
237 if requested in self._handlers:
238 return self._handlers[requested], requested
240 # Minor fallback: within same major, find closest <= requested
241 candidates = [v for v in self._handlers if v.major == requested.major and v <= requested]
242 if candidates:
243 best = max(candidates) # highest compatible version
244 return self._handlers[best], best
246 # No match → default or None
247 if self._default_version:
248 default = self._default_version
249 return self._handlers.get(default), default
251 return None, None
253 def list_versions(self) -> list[str]:
254 return sorted(str(v) for v in self._handlers.keys())
257# ---------------------------------------------------------------------------
258# Starlette Middleware
259# ---------------------------------------------------------------------------
262class APIVersioningMiddleware:
263 """
264 Starlette-compatible API versioning middleware.
266 Usage:
267 app = Starlette()
268 app.add_middleware(
269 APIVersioningMiddleware,
270 supported_versions=["v1", "v2"],
271 default_version="v1",
272 )
273 """
275 def __init__(
276 self,
277 app,
278 supported_versions: list[str] | None = None,
279 default_version: str | None = None,
280 deprecation_policy: DeprecationPolicy | None = None,
281 strategies: list[VersionStrategy] | None = None,
282 ):
283 self.app = app
284 self._supported = {SemVer.parse(v) for v in (supported_versions or []) if SemVer.parse(v)}
285 self._default = SemVer.parse(default_version) if default_version else None
286 self._deprecation = deprecation_policy or DeprecationPolicy()
287 self._extractor = VersionExtractor(strategies)
289 async def __call__(self, scope, receive, send) -> None:
290 if scope["type"] != "http":
291 await self.app(scope, receive, send)
292 return
294 request = Request(scope, receive)
295 version = self._extractor.extract(request) or self._default
297 # No version → pass through
298 if version is None:
299 await self.app(scope, receive, send)
300 return
302 # Check if version is supported
303 if self._supported and version not in self._supported:
304 response = self._version_error_response(version, "unsupported")
305 await response(scope, receive, send)
306 return
308 # Check deprecation → add warning header
309 if self._deprecation.should_block(version):
310 response = self._version_error_response(version, "sunset")
311 await response(scope, receive, send)
312 return
314 # Add version to request state
315 request.state.api_version = version
316 request.state.api_version_str = str(version)
318 # Add deprecation warning header
319 if self._deprecation.is_deprecated(version):
320 info = self._deprecation.get_info(version)
321 if info:
322 response = await self._with_deprecation_warning(scope, receive, send, info)
323 return
325 await self.app(scope, receive, send)
327 async def _with_deprecation_warning(self, scope, receive, send, info: DeprecationInfo) -> None:
328 """Wrap response with deprecation headers."""
330 async def send_wrapper(message):
331 if message["type"] == "http.response.start":
332 headers = dict(message.get("headers", []))
333 headers[b"deprecation"] = b"true"
334 headers[b"sunset"] = (
335 info.sunset_date.isoformat().encode() if info.sunset_date else b"unknown"
336 )
337 if info.migration_guide_url:
338 headers[b"link"] = f'<{info.migration_guide_url}>; rel="deprecation"'.encode()
339 message["headers"] = list(headers.items())
340 await send(message)
342 await self.app(scope, receive, send_wrapper)
344 @staticmethod
345 def _version_error_response(version: SemVer, reason: str) -> Response:
346 if reason == "unsupported":
347 detail = f"API version {version} is not supported."
348 status_code = 400
349 else:
350 detail = f"API version {version} has been sunset."
351 status_code = 410
353 return JSONResponse(
354 {"error": reason, "detail": detail, "version": str(version)},
355 status_code=status_code,
356 )
359# ---------------------------------------------------------------------------
360# Backward Compatibility Aliases
361# ---------------------------------------------------------------------------
363# Old API → New API mapping
364APIVersion = SemVer # SemVer replaces APIVersion
365VersionConfig = DeprecationPolicy # DeprecationPolicy replaces VersionConfig
366VersionNegotiator = VersionedRouter # VersionedRouter replaces VersionNegotiator