Coverage for agentos/api/versioning.py: 35%
178 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +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 dataclasses import dataclass
23from datetime import datetime, timedelta, timezone
24from enum import Enum
25from typing import Any, Callable, Dict, List, Optional, Tuple
27from starlette.requests import Request
28from starlette.responses import JSONResponse, Response
31# ---------------------------------------------------------------------------
32# Semantic Version
33# ---------------------------------------------------------------------------
36@dataclass(order=True, frozen=True)
37class SemVer:
38 """Semantic version (major.minor.patch)."""
39 major: int
40 minor: int = 0
41 patch: int = 0
43 _PARSE_RE = re.compile(r"^v?(\d+)(?:\.(\d+)(?:\.(\d+))?)?")
45 @classmethod
46 def parse(cls, version_str: str) -> Optional["SemVer"]:
47 """Parse a version string like 'v1', '2.0', '1.2.3'."""
48 m = cls._PARSE_RE.match(version_str.strip())
49 if not m:
50 return None
51 return cls(
52 major=int(m.group(1)),
53 minor=int(m.group(2) or 0),
54 patch=int(m.group(3) or 0),
55 )
57 def is_compatible(self, other: "SemVer") -> bool:
58 """Check if other version is API-compatible (same major)."""
59 return self.major == other.major
61 def __str__(self) -> str:
62 return f"v{self.major}.{self.minor}.{self.patch}"
65# ---------------------------------------------------------------------------
66# Version Extraction Strategy
67# ---------------------------------------------------------------------------
70class VersionStrategy(str, Enum):
71 """How to extract API version from requests."""
72 URL_PATH = "url_path" # /v1/resource
73 HEADER = "header" # Accept: application/json; version=1
74 QUERY_PARAM = "query_param" # /resource?api_version=1
77class VersionExtractor:
78 """Extract API version from incoming requests."""
80 STRATEGIES: List[VersionStrategy] = [
81 VersionStrategy.URL_PATH,
82 VersionStrategy.HEADER,
83 VersionStrategy.QUERY_PARAM,
84 ]
86 def __init__(self, strategies: Optional[List[VersionStrategy]] = None):
87 self._strategies = strategies or self.STRATEGIES
89 def extract(self, request: Request) -> Optional[SemVer]:
90 """Try each strategy in order; return first match."""
91 for strategy in self._strategies:
92 version = self._try_strategy(request, strategy)
93 if version is not None:
94 return version
95 return None
97 def _try_strategy(self, request: Request, strategy: VersionStrategy) -> Optional[SemVer]:
98 if strategy == VersionStrategy.URL_PATH:
99 return self._from_url(request)
100 elif strategy == VersionStrategy.HEADER:
101 return self._from_header(request)
102 elif strategy == VersionStrategy.QUERY_PARAM:
103 return self._from_query(request)
104 return None
106 @staticmethod
107 def _from_url(request: Request) -> Optional[SemVer]:
108 """Extract from /v{major}/... or /v{major}.{minor}/..."""
109 path = request.url.path.lstrip("/")
110 parts = path.split("/")
111 for i, part in enumerate(parts):
112 if part.lower().startswith("v") and part[1:].replace(".", "").isdigit():
113 return SemVer.parse(part)
114 return None
116 @staticmethod
117 def _from_header(request: Request) -> Optional[SemVer]:
118 """Extract from Accept header or custom X-API-Version."""
119 custom = request.headers.get("X-API-Version")
120 if custom:
121 return SemVer.parse(custom)
123 accept = request.headers.get("Accept", "")
124 version_match = re.search(r"version=(\d+(?:\.\d+)*)", accept)
125 if version_match:
126 return SemVer.parse(version_match.group(1))
127 return None
129 @staticmethod
130 def _from_query(request: Request) -> Optional[SemVer]:
131 """Extract from ?api_version=1 or ?v=2.0."""
132 for param in ("api_version", "v", "version"):
133 value = request.query_params.get(param)
134 if value:
135 return SemVer.parse(value)
136 return None
139# ---------------------------------------------------------------------------
140# Deprecation Policy
141# ---------------------------------------------------------------------------
144@dataclass
145class DeprecationInfo:
146 """Information about a deprecated API version."""
147 version: SemVer
148 sunset_date: Optional[datetime] = None
149 migration_guide_url: Optional[str] = None
150 message: str = "This API version is deprecated."
152 @property
153 def is_sunset(self) -> bool:
154 if self.sunset_date is None:
155 return False
156 return datetime.now(timezone.utc) > self.sunset_date
159class DeprecationPolicy:
160 """Manage API version deprecation."""
162 def __init__(self):
163 self._deprecated: Dict[SemVer, DeprecationInfo] = {}
165 def deprecate(self, version: str, sunset_days: int = 90, **kwargs) -> None:
166 """Mark a version as deprecated."""
167 semver = SemVer.parse(version)
168 if semver is None:
169 raise ValueError(f"Invalid version: {version}")
170 sunset = datetime.now(timezone.utc) + timedelta(days=sunset_days)
171 self._deprecated[semver] = DeprecationInfo(
172 version=semver,
173 sunset_date=sunset,
174 **kwargs,
175 )
177 def is_deprecated(self, version: SemVer) -> bool:
178 return version in self._deprecated
180 def get_info(self, version: SemVer) -> Optional[DeprecationInfo]:
181 return self._deprecated.get(version)
183 def should_block(self, version: SemVer) -> bool:
184 info = self._deprecated.get(version)
185 return info is not None and info.is_sunset
187 def list_deprecated(self) -> Dict[str, Dict[str, Any]]:
188 return {
189 str(v): {
190 "sunset_date": d.sunset_date.isoformat() if d.sunset_date else None,
191 "is_sunset": d.is_sunset,
192 "migration_guide": d.migration_guide_url,
193 }
194 for v, d in self._deprecated.items()
195 }
198# ---------------------------------------------------------------------------
199# Versioned Router
200# ---------------------------------------------------------------------------
203class VersionedRouter:
204 """
205 Route requests to version-specific handlers.
207 Supports semantic version negotiation:
208 - Exact match: /v1.0.0 → v1.0.0 handler
209 - Minor fallback: /v1.2.x → v1.2.0 handler
210 - Major fallback: /v1.x.x → latest v1 handler
211 """
213 def __init__(self):
214 self._handlers: Dict[SemVer, Callable] = {}
215 self._default_version: Optional[SemVer] = None
217 def register(self, version: str, handler: Callable) -> None:
218 """Register a handler for a specific version."""
219 semver = SemVer.parse(version)
220 if semver is None:
221 raise ValueError(f"Invalid version: {version}")
222 self._handlers[semver] = handler
224 def set_default(self, version: str) -> None:
225 """Set the default version when no version is specified."""
226 self._default_version = SemVer.parse(version)
228 def resolve(self, requested: SemVer) -> Tuple[Optional[Callable], Optional[SemVer]]:
229 """
230 Resolve a version to its handler.
231 Returns (handler, actual_version) or (None, None).
232 """
233 # Exact match
234 if requested in self._handlers:
235 return self._handlers[requested], requested
237 # Minor fallback: within same major, find closest <= requested
238 candidates = [
239 v for v in self._handlers
240 if v.major == requested.major and v <= requested
241 ]
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: Optional[List[str]] = None,
279 default_version: Optional[str] = None,
280 deprecation_policy: Optional[DeprecationPolicy] = None,
281 strategies: Optional[List[VersionStrategy]] = 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(
328 self, scope, receive, send, info: DeprecationInfo
329 ) -> None:
330 """Wrap response with deprecation headers."""
332 async def send_wrapper(message):
333 if message["type"] == "http.response.start":
334 headers = dict(message.get("headers", []))
335 headers[b"deprecation"] = b"true"
336 headers[b"sunset"] = info.sunset_date.isoformat().encode() if info.sunset_date else b"unknown"
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