Coverage for src / dataknobs_llm / prompts / versioning / version_manager.py: 18%
123 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-15 10:28 -0700
1"""Version management for prompts.
3This module provides version control capabilities including:
4- Creating and retrieving prompt versions
5- Semantic version management
6- Version history tracking
7- Tagging and status management
8"""
10import re
11import uuid
12from typing import Any, Dict, List
13from datetime import datetime
15from dataknobs_llm.exceptions import VersioningError
17from .types import (
18 PromptVersion,
19 VersionStatus,
20)
23class VersionManager:
24 """Manages prompt versions with semantic versioning.
26 Handles version creation, retrieval, and lifecycle management.
27 Supports semantic versioning (major.minor.patch) and version tagging.
29 Example:
30 ```python
31 manager = VersionManager(storage_backend)
33 # Create a version
34 v1 = await manager.create_version(
35 name="greeting",
36 prompt_type="system",
37 template="Hello {{name}}!",
38 version="1.0.0"
39 )
41 # Get latest version
42 latest = await manager.get_version(
43 name="greeting",
44 prompt_type="system"
45 )
47 # Tag a version
48 await manager.tag_version(v1.version_id, "production")
49 ```
50 """
52 # Semantic version pattern: major.minor.patch
53 VERSION_PATTERN = re.compile(r"^(\d+)\.(\d+)\.(\d+)$")
55 def __init__(self, storage: Any | None = None):
56 """Initialize version manager.
58 Args:
59 storage: Backend storage (dict for in-memory, database for persistence)
60 If None, uses in-memory dictionary
61 """
62 self.storage = storage if storage is not None else {}
63 self._versions: Dict[str, PromptVersion] = {} # version_id -> PromptVersion
64 self._version_index: Dict[str, List[str]] = {} # "{name}:{type}" -> [version_ids]
66 async def create_version(
67 self,
68 name: str,
69 prompt_type: str,
70 template: str,
71 version: str | None = None,
72 defaults: Dict[str, Any] | None = None,
73 validation: Dict[str, Any] | None = None,
74 metadata: Dict[str, Any] | None = None,
75 created_by: str | None = None,
76 parent_version: str | None = None,
77 tags: List[str] | None = None,
78 status: VersionStatus = VersionStatus.ACTIVE,
79 ) -> PromptVersion:
80 """Create a new prompt version.
82 Args:
83 name: Prompt name
84 prompt_type: Prompt type ("system", "user", "message")
85 template: Template content
86 version: Semantic version (e.g., "1.2.3"). If None, auto-increments from latest
87 defaults: Default parameter values
88 validation: Validation configuration
89 metadata: Additional metadata
90 created_by: Creator username/ID
91 parent_version: Previous version ID for history tracking
92 tags: List of tags
93 status: Initial version status
95 Returns:
96 Created PromptVersion
98 Raises:
99 VersioningError: If version format is invalid or version already exists
100 """
101 # Auto-increment version if not provided
102 if version is None:
103 version = await self._auto_increment_version(name, prompt_type)
104 # If no parent_version specified, use the latest version
105 if parent_version is None:
106 latest = await self.get_version(name, prompt_type)
107 if latest:
108 parent_version = latest.version_id
109 else:
110 # Validate version format
111 if not self.VERSION_PATTERN.match(version):
112 raise VersioningError(
113 f"Invalid version format: {version}. "
114 f"Expected semantic version (e.g., '1.0.0')"
115 )
117 # Check if version already exists
118 key = self._make_key(name, prompt_type)
119 existing_versions = await self.list_versions(name, prompt_type)
120 if any(v.version == version for v in existing_versions):
121 raise VersioningError(
122 f"Version {version} already exists for {name} ({prompt_type})"
123 )
125 # Generate unique version ID
126 version_id = str(uuid.uuid4())
128 # Create version object
129 prompt_version = PromptVersion(
130 version_id=version_id,
131 name=name,
132 prompt_type=prompt_type,
133 version=version,
134 template=template,
135 defaults=defaults or {},
136 validation=validation,
137 metadata=metadata or {},
138 created_at=datetime.utcnow(),
139 created_by=created_by,
140 parent_version=parent_version,
141 tags=tags or [],
142 status=status,
143 )
145 # Store version
146 self._versions[version_id] = prompt_version
148 # Update index
149 if key not in self._version_index:
150 self._version_index[key] = []
151 self._version_index[key].append(version_id)
153 # Persist to backend if available
154 if hasattr(self.storage, "set"):
155 await self._persist_version(prompt_version)
157 return prompt_version
159 async def get_version(
160 self,
161 name: str,
162 prompt_type: str,
163 version: str = "latest",
164 version_id: str | None = None,
165 ) -> PromptVersion | None:
166 """Retrieve a prompt version.
168 Args:
169 name: Prompt name
170 prompt_type: Prompt type
171 version: Version string or "latest" for most recent
172 version_id: Specific version ID (takes precedence over version)
174 Returns:
175 PromptVersion if found, None otherwise
176 """
177 # Direct lookup by version_id
178 if version_id:
179 return self._versions.get(version_id)
181 # Get all versions for this prompt
182 versions = await self.list_versions(name, prompt_type)
183 if not versions:
184 return None
186 # Return latest version
187 if version == "latest":
188 return self._get_latest_version(versions)
190 # Find specific version
191 for v in versions:
192 if v.version == version:
193 return v
195 return None
197 async def list_versions(
198 self,
199 name: str,
200 prompt_type: str,
201 tags: List[str] | None = None,
202 status: VersionStatus | None = None,
203 ) -> List[PromptVersion]:
204 """List all versions of a prompt.
206 Args:
207 name: Prompt name
208 prompt_type: Prompt type
209 tags: Filter by tags (returns versions with ANY of these tags)
210 status: Filter by status
212 Returns:
213 List of PromptVersion objects, sorted by version (newest first)
214 """
215 key = self._make_key(name, prompt_type)
216 version_ids = self._version_index.get(key, [])
218 versions = [self._versions[vid] for vid in version_ids]
220 # Apply filters
221 if tags:
222 versions = [v for v in versions if any(t in v.tags for t in tags)]
224 if status:
225 versions = [v for v in versions if v.status == status]
227 # Sort by version (newest first)
228 return sorted(versions, key=lambda v: self._parse_version(v.version), reverse=True)
230 async def tag_version(
231 self,
232 version_id: str,
233 tag: str,
234 ) -> PromptVersion:
235 """Add a tag to a version.
237 Args:
238 version_id: Version ID to tag
239 tag: Tag to add (e.g., "production", "deprecated")
241 Returns:
242 Updated PromptVersion
244 Raises:
245 VersioningError: If version not found
246 """
247 version = self._versions.get(version_id)
248 if not version:
249 raise VersioningError(f"Version not found: {version_id}")
251 if tag not in version.tags:
252 version.tags.append(tag)
254 # Persist if backend available
255 if hasattr(self.storage, "set"):
256 await self._persist_version(version)
258 return version
260 async def untag_version(
261 self,
262 version_id: str,
263 tag: str,
264 ) -> PromptVersion:
265 """Remove a tag from a version.
267 Args:
268 version_id: Version ID
269 tag: Tag to remove
271 Returns:
272 Updated PromptVersion
274 Raises:
275 VersioningError: If version not found
276 """
277 version = self._versions.get(version_id)
278 if not version:
279 raise VersioningError(f"Version not found: {version_id}")
281 if tag in version.tags:
282 version.tags.remove(tag)
284 # Persist if backend available
285 if hasattr(self.storage, "set"):
286 await self._persist_version(version)
288 return version
290 async def update_status(
291 self,
292 version_id: str,
293 status: VersionStatus,
294 ) -> PromptVersion:
295 """Update version status.
297 Args:
298 version_id: Version ID
299 status: New status
301 Returns:
302 Updated PromptVersion
304 Raises:
305 VersioningError: If version not found
306 """
307 version = self._versions.get(version_id)
308 if not version:
309 raise VersioningError(f"Version not found: {version_id}")
311 version.status = status
313 # Persist if backend available
314 if hasattr(self.storage, "set"):
315 await self._persist_version(version)
317 return version
319 async def delete_version(
320 self,
321 version_id: str,
322 ) -> bool:
323 """Delete a version.
325 Note: This permanently removes the version. Consider using
326 update_status() with ARCHIVED instead.
328 Args:
329 version_id: Version ID to delete
331 Returns:
332 True if deleted, False if not found
333 """
334 version = self._versions.get(version_id)
335 if not version:
336 return False
338 # Remove from index
339 key = self._make_key(version.name, version.prompt_type)
340 if key in self._version_index:
341 self._version_index[key] = [
342 vid for vid in self._version_index[key]
343 if vid != version_id
344 ]
346 # Remove from storage
347 del self._versions[version_id]
349 # Persist deletion if backend available
350 if hasattr(self.storage, "delete"):
351 await self.storage.delete(f"version:{version_id}")
353 return True
355 # ===== Helper Methods =====
357 def _make_key(self, name: str, prompt_type: str) -> str:
358 """Create index key for prompt name and type."""
359 return f"{name}:{prompt_type}"
361 def _parse_version(self, version: str) -> tuple:
362 """Parse semantic version into (major, minor, patch) tuple."""
363 match = self.VERSION_PATTERN.match(version)
364 if not match:
365 return (0, 0, 0)
366 return tuple(int(x) for x in match.groups())
368 def _get_latest_version(self, versions: List[PromptVersion]) -> PromptVersion | None:
369 """Get the latest version from a list."""
370 if not versions:
371 return None
373 # Filter out archived/deprecated if active versions exist
374 active = [v for v in versions if v.status in (VersionStatus.ACTIVE, VersionStatus.PRODUCTION)]
375 if active:
376 versions = active
378 # Sort by version number
379 sorted_versions = sorted(
380 versions,
381 key=lambda v: self._parse_version(v.version),
382 reverse=True
383 )
384 return sorted_versions[0]
386 async def _auto_increment_version(
387 self,
388 name: str,
389 prompt_type: str,
390 ) -> str:
391 """Auto-increment version number from latest version.
393 Increments patch version by default (1.0.0 -> 1.0.1).
394 If no versions exist, returns "1.0.0".
395 """
396 versions = await self.list_versions(name, prompt_type)
397 if not versions:
398 return "1.0.0"
400 latest = self._get_latest_version(versions)
401 if not latest:
402 return "1.0.0"
404 major, minor, patch = self._parse_version(latest.version)
405 return f"{major}.{minor}.{patch + 1}"
407 async def _persist_version(self, version: PromptVersion):
408 """Persist version to backend storage."""
409 if hasattr(self.storage, "set"):
410 key = f"version:{version.version_id}"
411 await self.storage.set(key, version.to_dict())
413 async def get_version_history(
414 self,
415 name: str,
416 prompt_type: str,
417 ) -> List[PromptVersion]:
418 """Get version history with parent relationships.
420 Returns versions in chronological order (oldest first).
422 Args:
423 name: Prompt name
424 prompt_type: Prompt type
426 Returns:
427 List of versions in chronological order
428 """
429 versions = await self.list_versions(name, prompt_type)
430 return sorted(versions, key=lambda v: v.created_at)