Coverage for agentos/api/versioning.py: 36%
179 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 08:01 +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
22import time
23from dataclasses import dataclass, field
24from datetime import datetime, timedelta, timezone
25from enum import Enum
26from typing import Any, Callable, Dict, List, Optional, Set, Tuple
28from starlette.requests import Request
29from starlette.responses import JSONResponse, Response
32# ---------------------------------------------------------------------------
33# Semantic Version
34# ---------------------------------------------------------------------------
37@dataclass(order=True, frozen=True)
38class SemVer:
39 """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) -> Optional["SemVer"]:
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(str, Enum):
72 """How to extract API version from requests."""
73 URL_PATH = "url_path" # /v1/resource
74 HEADER = "header" # Accept: application/json; version=1
75 QUERY_PARAM = "query_param" # /resource?api_version=1
78class VersionExtractor:
79 """Extract API version from incoming requests."""
81 STRATEGIES: List[VersionStrategy] = [
82 VersionStrategy.URL_PATH,
83 VersionStrategy.HEADER,
84 VersionStrategy.QUERY_PARAM,
85 ]
87 def __init__(self, strategies: Optional[List[VersionStrategy]] = None):
88 self._strategies = strategies or self.STRATEGIES
90 def extract(self, request: Request) -> Optional[SemVer]:
91 """Try each strategy in order; return first match."""
92 for strategy in self._strategies:
93 version = self._try_strategy(request, strategy)
94 if version is not None:
95 return version
96 return None
98 def _try_strategy(self, request: Request, strategy: VersionStrategy) -> Optional[SemVer]:
99 if strategy == VersionStrategy.URL_PATH:
100 return self._from_url(request)
101 elif strategy == VersionStrategy.HEADER:
102 return self._from_header(request)
103 elif strategy == VersionStrategy.QUERY_PARAM:
104 return self._from_query(request)
105 return None
107 @staticmethod
108 def _from_url(request: Request) -> Optional[SemVer]:
109 """Extract from /v{major}/... or /v{major}.{minor}/..."""
110 path = request.url.path.lstrip("/")
111 parts = path.split("/")
112 for i, part in enumerate(parts):
113 if part.lower().startswith("v") and part[1:].replace(".", "").isdigit():
114 return SemVer.parse(part)
115 return None
117 @staticmethod
118 def _from_header(request: Request) -> Optional[SemVer]:
119 """Extract from Accept header or custom X-API-Version."""
120 custom = request.headers.get("X-API-Version")
121 if custom:
122 return SemVer.parse(custom)
124 accept = request.headers.get("Accept", "")
125 version_match = re.search(r"version=(\d+(?:\.\d+)*)", accept)
126 if version_match:
127 return SemVer.parse(version_match.group(1))
128 return None
130 @staticmethod
131 def _from_query(request: Request) -> Optional[SemVer]:
132 """Extract from ?api_version=1 or ?v=2.0."""
133 for param in ("api_version", "v", "version"):
134 value = request.query_params.get(param)
135 if value:
136 return SemVer.parse(value)
137 return None
140# ---------------------------------------------------------------------------
141# Deprecation Policy
142# ---------------------------------------------------------------------------
145@dataclass
146class DeprecationInfo:
147 """Information about a deprecated API version."""
148 version: SemVer
149 sunset_date: Optional[datetime] = None
150 migration_guide_url: Optional[str] = None
151 message: str = "This API version is deprecated."
153 @property
154 def is_sunset(self) -> bool:
155 if self.sunset_date is None:
156 return False
157 return datetime.now(timezone.utc) > self.sunset_date
160class DeprecationPolicy:
161 """Manage API version deprecation."""
163 def __init__(self):
164 self._deprecated: Dict[SemVer, DeprecationInfo] = {}
166 def deprecate(self, version: str, sunset_days: int = 90, **kwargs) -> None:
167 """Mark a version as deprecated."""
168 semver = SemVer.parse(version)
169 if semver is None:
170 raise ValueError(f"Invalid version: {version}")
171 sunset = datetime.now(timezone.utc) + timedelta(days=sunset_days)
172 self._deprecated[semver] = DeprecationInfo(
173 version=semver,
174 sunset_date=sunset,
175 **kwargs,
176 )
178 def is_deprecated(self, version: SemVer) -> bool:
179 return version in self._deprecated
181 def get_info(self, version: SemVer) -> Optional[DeprecationInfo]:
182 return self._deprecated.get(version)
184 def should_block(self, version: SemVer) -> bool:
185 info = self._deprecated.get(version)
186 return info is not None and info.is_sunset
188 def list_deprecated(self) -> Dict[str, Dict[str, Any]]:
189 return {
190 str(v): {
191 "sunset_date": d.sunset_date.isoformat() if d.sunset_date else None,
192 "is_sunset": d.is_sunset,
193 "migration_guide": d.migration_guide_url,
194 }
195 for v, d in self._deprecated.items()
196 }
199# ---------------------------------------------------------------------------
200# Versioned Router
201# ---------------------------------------------------------------------------
204class VersionedRouter:
205 """
206 Route requests to version-specific handlers.
208 Supports semantic version negotiation:
209 - Exact match: /v1.0.0 → v1.0.0 handler
210 - Minor fallback: /v1.2.x → v1.2.0 handler
211 - Major fallback: /v1.x.x → latest v1 handler
212 """
214 def __init__(self):
215 self._handlers: Dict[SemVer, Callable] = {}
216 self._default_version: Optional[SemVer] = None
218 def register(self, version: str, handler: Callable) -> None:
219 """Register a handler for a specific version."""
220 semver = SemVer.parse(version)
221 if semver is None:
222 raise ValueError(f"Invalid version: {version}")
223 self._handlers[semver] = handler
225 def set_default(self, version: str) -> None:
226 """Set the default version when no version is specified."""
227 self._default_version = SemVer.parse(version)
229 def resolve(self, requested: SemVer) -> Tuple[Optional[Callable], Optional[SemVer]]:
230 """
231 Resolve a version to its handler.
232 Returns (handler, actual_version) or (None, None).
233 """
234 # Exact match
235 if requested in self._handlers:
236 return self._handlers[requested], requested
238 # Minor fallback: within same major, find closest <= requested
239 candidates = [
240 v for v in self._handlers
241 if v.major == requested.major and v <= requested
242 ]
243 if candidates:
244 best = max(candidates) # highest compatible version
245 return self._handlers[best], best
247 # No match → default or None
248 if self._default_version:
249 default = self._default_version
250 return self._handlers.get(default), default
252 return None, None
254 def list_versions(self) -> List[str]:
255 return sorted(str(v) for v in self._handlers.keys())
258# ---------------------------------------------------------------------------
259# Starlette Middleware
260# ---------------------------------------------------------------------------
263class APIVersioningMiddleware:
264 """
265 Starlette-compatible API versioning middleware.
267 Usage:
268 app = Starlette()
269 app.add_middleware(
270 APIVersioningMiddleware,
271 supported_versions=["v1", "v2"],
272 default_version="v1",
273 )
274 """
276 def __init__(
277 self,
278 app,
279 supported_versions: Optional[List[str]] = None,
280 default_version: Optional[str] = None,
281 deprecation_policy: Optional[DeprecationPolicy] = None,
282 strategies: Optional[List[VersionStrategy]] = None,
283 ):
284 self.app = app
285 self._supported = {SemVer.parse(v) for v in (supported_versions or []) if SemVer.parse(v)}
286 self._default = SemVer.parse(default_version) if default_version else None
287 self._deprecation = deprecation_policy or DeprecationPolicy()
288 self._extractor = VersionExtractor(strategies)
290 async def __call__(self, scope, receive, send) -> None:
291 if scope["type"] != "http":
292 await self.app(scope, receive, send)
293 return
295 request = Request(scope, receive)
296 version = self._extractor.extract(request) or self._default
298 # No version → pass through
299 if version is None:
300 await self.app(scope, receive, send)
301 return
303 # Check if version is supported
304 if self._supported and version not in self._supported:
305 response = self._version_error_response(version, "unsupported")
306 await response(scope, receive, send)
307 return
309 # Check deprecation → add warning header
310 if self._deprecation.should_block(version):
311 response = self._version_error_response(version, "sunset")
312 await response(scope, receive, send)
313 return
315 # Add version to request state
316 request.state.api_version = version
317 request.state.api_version_str = str(version)
319 # Add deprecation warning header
320 if self._deprecation.is_deprecated(version):
321 info = self._deprecation.get_info(version)
322 if info:
323 response = await self._with_deprecation_warning(scope, receive, send, info)
324 return
326 await self.app(scope, receive, send)
328 async def _with_deprecation_warning(
329 self, scope, receive, send, info: DeprecationInfo
330 ) -> None:
331 """Wrap response with deprecation headers."""
333 async def send_wrapper(message):
334 if message["type"] == "http.response.start":
335 headers = dict(message.get("headers", []))
336 headers[b"deprecation"] = b"true"
337 headers[b"sunset"] = info.sunset_date.isoformat().encode() if info.sunset_date else b"unknown"
338 if info.migration_guide_url:
339 headers[b"link"] = f'<{info.migration_guide_url}>; rel="deprecation"'.encode()
340 message["headers"] = list(headers.items())
341 await send(message)
343 await self.app(scope, receive, send_wrapper)
345 @staticmethod
346 def _version_error_response(version: SemVer, reason: str) -> Response:
347 if reason == "unsupported":
348 detail = f"API version {version} is not supported."
349 status_code = 400
350 else:
351 detail = f"API version {version} has been sunset."
352 status_code = 410
354 return JSONResponse(
355 {"error": reason, "detail": detail, "version": str(version)},
356 status_code=status_code,
357 )
360# ---------------------------------------------------------------------------
361# Backward Compatibility Aliases
362# ---------------------------------------------------------------------------
364# Old API → New API mapping
365APIVersion = SemVer # SemVer replaces APIVersion
366VersionConfig = DeprecationPolicy # DeprecationPolicy replaces VersionConfig
367VersionNegotiator = VersionedRouter # VersionedRouter replaces VersionNegotiator