Coverage for src / tracekit / plugins / versioning.py: 100%
118 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Plugin versioning and migration support.
3This module provides version compatibility checking, migration support
4between plugin versions, and multi-version compatibility layers.
5"""
7from __future__ import annotations
9import logging
10from dataclasses import dataclass
11from typing import TYPE_CHECKING, Any
13if TYPE_CHECKING:
14 from collections.abc import Callable
16 from tracekit.plugins.base import PluginBase
18logger = logging.getLogger(__name__)
21@dataclass
22class VersionRange:
23 """Version range specification.
25 Supports version range syntax:
26 - "1.0.0" - exact version
27 - ">=1.0.0" - greater than or equal
28 - "<2.0.0" - less than
29 - "^1.5.0" - compatible with (same major)
30 - "~1.5.0" - approximately (same major.minor)
31 - "*" - any version
33 References:
34 PLUG-005: Plugin Dependencies - version range support
35 """
37 spec: str
39 def matches(self, version: str) -> bool:
40 """Check if version matches this range.
42 Args:
43 version: Version string to check (semver format)
45 Returns:
46 True if version matches range
48 References:
49 PLUG-005: Plugin Dependencies - version range support
50 """
51 if self.spec == "*":
52 return True
54 # Parse version
55 try:
56 v_major, v_minor, v_patch = self._parse_version(version)
57 except ValueError:
58 return False
60 # Handle different operators
61 if self.spec.startswith(">="):
62 target = self.spec[2:].strip()
63 t_major, t_minor, t_patch = self._parse_version(target)
64 return (v_major, v_minor, v_patch) >= (t_major, t_minor, t_patch)
66 elif self.spec.startswith("<="):
67 target = self.spec[2:].strip()
68 t_major, t_minor, t_patch = self._parse_version(target)
69 return (v_major, v_minor, v_patch) <= (t_major, t_minor, t_patch)
71 elif self.spec.startswith(">"):
72 target = self.spec[1:].strip()
73 t_major, t_minor, t_patch = self._parse_version(target)
74 return (v_major, v_minor, v_patch) > (t_major, t_minor, t_patch)
76 elif self.spec.startswith("<"):
77 target = self.spec[1:].strip()
78 t_major, t_minor, t_patch = self._parse_version(target)
79 return (v_major, v_minor, v_patch) < (t_major, t_minor, t_patch)
81 elif self.spec.startswith("^"):
82 # Compatible: same major version
83 target = self.spec[1:].strip()
84 t_major, t_minor, t_patch = self._parse_version(target)
85 return v_major == t_major and (v_minor, v_patch) >= (t_minor, t_patch)
87 elif self.spec.startswith("~"):
88 # Approximately: same major.minor version
89 target = self.spec[1:].strip()
90 t_major, t_minor, t_patch = self._parse_version(target)
91 return v_major == t_major and v_minor == t_minor and v_patch >= t_patch
93 else:
94 # Exact match
95 return version == self.spec
97 def _parse_version(self, version: str) -> tuple[int, int, int]:
98 """Parse semver version string.
100 Args:
101 version: Version string (e.g., "1.2.3")
103 Returns:
104 Tuple of (major, minor, patch)
106 Raises:
107 ValueError: If version format is invalid
108 """
109 # Handle version with metadata (e.g., "1.2.3-beta+build")
110 version = version.split("-")[0].split("+")[0]
112 parts = version.split(".")
113 if len(parts) != 3:
114 raise ValueError(f"Invalid version format: {version}")
116 try:
117 major = int(parts[0])
118 minor = int(parts[1])
119 patch = int(parts[2])
120 return (major, minor, patch)
121 except ValueError as e:
122 raise ValueError(f"Invalid version format: {version}") from e
125@dataclass
126class Migration:
127 """Plugin migration definition.
129 Defines migration path from one version to another.
131 Attributes:
132 from_version: Source version
133 to_version: Target version
134 migrate_func: Migration function
135 description: Migration description
137 References:
138 PLUG-003: Plugin Versioning - migration support
139 """
141 from_version: str
142 to_version: str
143 migrate_func: Callable[[dict[str, Any]], dict[str, Any]]
144 description: str = ""
146 def apply(self, config: dict[str, Any]) -> dict[str, Any]:
147 """Apply migration to configuration.
149 Args:
150 config: Plugin configuration
152 Returns:
153 Migrated configuration
155 References:
156 PLUG-003: Plugin Versioning - migration support
157 """
158 logger.info(f"Migrating plugin config from v{self.from_version} to v{self.to_version}")
159 return self.migrate_func(config)
162class VersionCompatibilityLayer:
163 """Multi-version compatibility layer for plugins.
165 Allows plugins to support multiple API versions by adapting
166 the interface based on the current TraceKit API version.
168 References:
169 PLUG-003: Plugin Versioning - multi-version compatibility layer
170 """
172 def __init__(self, plugin: PluginBase) -> None:
173 """Initialize compatibility layer.
175 Args:
176 plugin: Plugin instance to wrap
177 """
178 self._plugin = plugin
179 self._api_version = "1.0.0" # Default
180 self._adapters: dict[str, Callable] = {} # type: ignore[type-arg]
182 def set_api_version(self, api_version: str) -> None:
183 """Set target API version.
185 Args:
186 api_version: TraceKit API version
188 References:
189 PLUG-003: Plugin Versioning - multi-version compatibility
190 """
191 self._api_version = api_version
192 logger.debug(f"Set API version to {api_version} for plugin {self._plugin.name}")
194 def register_adapter(
195 self,
196 api_version: str,
197 method_name: str,
198 adapter: Callable, # type: ignore[type-arg]
199 ) -> None:
200 """Register method adapter for specific API version.
202 Args:
203 api_version: API version this adapter is for
204 method_name: Method name to adapt
205 adapter: Adapter function
207 References:
208 PLUG-003: Plugin Versioning - multi-version compatibility
209 """
210 key = f"{api_version}:{method_name}"
211 self._adapters[key] = adapter
213 def call_adapted(self, method_name: str, *args: Any, **kwargs: Any) -> Any:
214 """Call plugin method with version adaptation.
216 Args:
217 method_name: Method to call
218 *args: Positional arguments
219 **kwargs: Keyword arguments
221 Returns:
222 Method result
224 References:
225 PLUG-003: Plugin Versioning - multi-version compatibility
226 """
227 key = f"{self._api_version}:{method_name}"
229 if key in self._adapters:
230 # Use adapter
231 adapter = self._adapters[key]
232 return adapter(self._plugin, *args, **kwargs)
233 else:
234 # Call directly
235 method = getattr(self._plugin, method_name)
236 return method(*args, **kwargs)
239class MigrationManager:
240 """Manager for plugin configuration migrations.
242 Tracks and applies migrations between plugin versions.
244 References:
245 PLUG-003: Plugin Versioning - migration support
246 """
248 def __init__(self) -> None:
249 """Initialize migration manager."""
250 self._migrations: dict[str, list[Migration]] = {}
252 def register_migration(self, plugin_name: str, migration: Migration) -> None:
253 """Register a migration for a plugin.
255 Args:
256 plugin_name: Plugin name
257 migration: Migration definition
259 References:
260 PLUG-003: Plugin Versioning - migration support
261 """
262 if plugin_name not in self._migrations:
263 self._migrations[plugin_name] = []
265 self._migrations[plugin_name].append(migration)
266 logger.debug(
267 f"Registered migration for {plugin_name}: "
268 f"v{migration.from_version} -> v{migration.to_version}"
269 )
271 def get_migration_path(
272 self,
273 plugin_name: str,
274 from_version: str,
275 to_version: str,
276 ) -> list[Migration]:
277 """Get migration path between versions.
279 Args:
280 plugin_name: Plugin name
281 from_version: Source version
282 to_version: Target version
284 Returns:
285 List of migrations in order
287 Raises:
288 ValueError: If no migration path exists
290 References:
291 PLUG-003: Plugin Versioning - migration support
292 """
293 if plugin_name not in self._migrations:
294 return []
296 # Simple linear path (could be enhanced with graph search)
297 migrations = self._migrations[plugin_name]
298 path: list[Migration] = []
300 current = from_version
301 while current != to_version:
302 # Find next migration
303 next_migration = None
304 for migration in migrations:
305 if migration.from_version == current:
306 next_migration = migration
307 break
309 if next_migration is None:
310 raise ValueError(f"No migration path from v{from_version} to v{to_version}")
312 path.append(next_migration)
313 current = next_migration.to_version
315 return path
317 def migrate(
318 self,
319 plugin_name: str,
320 config: dict[str, Any],
321 from_version: str,
322 to_version: str,
323 ) -> dict[str, Any]:
324 """Migrate configuration between versions.
326 Args:
327 plugin_name: Plugin name
328 config: Current configuration
329 from_version: Source version
330 to_version: Target version
332 Returns:
333 Migrated configuration
335 References:
336 PLUG-003: Plugin Versioning - migration support
337 """
338 if from_version == to_version:
339 return config
341 path = self.get_migration_path(plugin_name, from_version, to_version)
343 result = config
344 for migration in path:
345 result = migration.apply(result)
347 return result
350# Global migration manager
351_migration_manager: MigrationManager | None = None
354def get_migration_manager() -> MigrationManager:
355 """Get global migration manager.
357 Returns:
358 Global MigrationManager instance
359 """
360 global _migration_manager
361 if _migration_manager is None:
362 _migration_manager = MigrationManager()
363 return _migration_manager
366__all__ = [
367 "Migration",
368 "MigrationManager",
369 "VersionCompatibilityLayer",
370 "VersionRange",
371 "get_migration_manager",
372]