Coverage for agentos/api/versioning.py: 38%

1"""AgentOS API version negotiation. 

2 

3Supports header-based and URL-path-based versioning for the AgentOS REST API. 

4""" 

5 

6from __future__ import annotations 

7 

8from dataclasses import dataclass, field 

9from enum import Enum 

10from typing import Optional 

11 

12 

13class VersionStrategy(Enum): 

14 

15 """版本策略枚举。""" 

16 

17 HEADER = "header" # Accept: application/json; version=1 

18 PATH = "path" # /api/v1/... 

19 QUERY = "query" # /api/endpoint?version=1 

20 

21 

22@dataclass 

23class APIVersion: 

24 """API 版本记录。""" 

25 major: int 

26 minor: int = 0 

27 

28 def __str__(self) -> str: 

29 return f"v{self.major}.{self.minor}" 

30 

31 @classmethod 

32 def parse(cls, raw: str) -> APIVersion: 

33 raw = raw.strip().lstrip("vV") 

34 parts = [int(p) for p in raw.split(".")] 

35 return cls(major=parts[0], minor=parts[1] if len(parts) > 1 else 0) 

36 

37 def __lt__(self, other: APIVersion) -> bool: 

38 return (self.major, self.minor) < (other.major, other.minor) 

39 

40 def __le__(self, other: APIVersion) -> bool: 

41 return (self.major, self.minor) <= (other.major, other.minor) 

42 

43 

44@dataclass 

45class VersionConfig: 

46 """版本管理配置。""" 

47 current: APIVersion = field(default_factory=lambda: APIVersion(1, 0)) 

48 min_supported: APIVersion = field(default_factory=lambda: APIVersion(1, 0)) 

49 deprecated: list[APIVersion] = field(default_factory=list) # versions that emit deprecation warnings 

50 strategy: VersionStrategy = VersionStrategy.HEADER 

51 header_name: str = "Accept" 

52 

53 

54class VersionNegotiator: 

55 """Negotiate and validate API version from incoming requests.""" 

56 

57 def __init__(self, config: Optional[VersionConfig] = None): 

58 self.config = config or VersionConfig() 

59 

60 def extract_from_headers(self, headers: dict) -> Optional[APIVersion]: 

61 """Extract version from Accept header (e.g. 'application/json; version=1').""" 

62 accept = headers.get(self.config.header_name.lower(), headers.get(self.config.header_name, "")) 

63 for part in accept.split(";"): 

64 part = part.strip() 

65 if part.startswith("version="): 

66 try: 

67 return APIVersion.parse(part.split("=", 1)[1]) 

68 except (ValueError, IndexError): 

69 return None 

70 return None 

71 

72 def extract_from_path(self, path: str) -> Optional[APIVersion]: 

73 """Extract version from URL path (e.g. '/api/v2/endpoint').""" 

74 import re 

75 m = re.match(r"^/api/v(\d+(?:\.\d+)?)/", path) 

76 if m: 

77 try: 

78 return APIVersion.parse(m.group(1)) 

79 except ValueError: 

80 return None 

81 return None 

82 

83 def extract_from_query(self, query: str) -> Optional[APIVersion]: 

84 """Extract version from query string (e.g. 'version=2').""" 

85 import urllib.parse 

86 params = urllib.parse.parse_qs(query) 

87 versions = params.get("version", []) 

88 if versions: 

89 try: 

90 return APIVersion.parse(versions[0]) 

91 except ValueError: 

92 return None 

93 return None 

94 

95 def negotiate(self, headers: dict, path: str = "", query: str = "") -> tuple[APIVersion, list[str]]: 

96 """Return (resolved_version, warnings).""" 

97 version: Optional[APIVersion] = None 

98 

99 if self.config.strategy == VersionStrategy.HEADER: 

100 version = self.extract_from_headers(headers) 

101 elif self.config.strategy == VersionStrategy.PATH: 

102 version = self.extract_from_path(path) 

103 elif self.config.strategy == VersionStrategy.QUERY: 

104 version = self.extract_from_query(query) 

105 

106 if version is None: 

107 version = self.config.current 

108 

109 warnings: list[str] = [] 

110 

111 if version < self.config.min_supported: 

112 warnings.append( 

113 f"API {version} is no longer supported (min: {self.config.min_supported})" 

114 ) 

115 

116 if version in self.config.deprecated: 

117 warnings.append(f"API {version} is deprecated — upgrade to {self.config.current}") 

118 

119 return version, warnings