Coverage for src/dataknobs_llm/prompts/versioning/version_manager.py: 88%
122 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-08 13:51 -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 .types import (
16 PromptVersion,
17 VersionStatus,
18 VersioningError,
19)
22class VersionManager:
23 """Manages prompt versions with semantic versioning.
25 Handles version creation, retrieval, and lifecycle management.
26 Supports semantic versioning (major.minor.patch) and version tagging.
28 Example:
29 ```python
30 manager = VersionManager(storage_backend)
32 # Create a version
33 v1 = await manager.create_version(
34 name="greeting",
35 prompt_type="system",
36 template="Hello {{name}}!",
37 version="1.0.0"
38 )
40 # Get latest version
41 latest = await manager.get_version(
42 name="greeting",
43 prompt_type="system"
44 )
46 # Tag a version
47 await manager.tag_version(v1.version_id, "production")
48 ```
49 """
51 # Semantic version pattern: major.minor.patch
52 VERSION_PATTERN = re.compile(r"^(\d+)\.(\d+)\.(\d+)$")
54 def __init__(self, storage: Any | None = None):
55 """Initialize version manager.
57 Args:
58 storage: Backend storage (dict for in-memory, database for persistence)
59 If None, uses in-memory dictionary
60 """
61 self.storage = storage if storage is not None else {}
62 self._versions: Dict[str, PromptVersion] = {} # version_id -> PromptVersion
63 self._version_index: Dict[str, List[str]] = {} # "{name}:{type}" -> [version_ids]
65 async def create_version(
66 self,
67 name: str,
68 prompt_type: str,
69 template: str,
70 version: str | None = None,
71 defaults: Dict[str, Any] | None = None,
72 validation: Dict[str, Any] | None = None,
73 metadata: Dict[str, Any] | None = None,
74 created_by: str | None = None,
75 parent_version: str | None = None,
76 tags: List[str] | None = None,
77 status: VersionStatus = VersionStatus.ACTIVE,
78 ) -> PromptVersion:
79 """Create a new prompt version.
81 Args:
82 name: Prompt name
83 prompt_type: Prompt type ("system", "user", "message")
84 template: Template content
85 version: Semantic version (e.g., "1.2.3"). If None, auto-increments from latest
86 defaults: Default parameter values
87 validation: Validation configuration
88 metadata: Additional metadata
89 created_by: Creator username/ID
90 parent_version: Previous version ID for history tracking
91 tags: List of tags
92 status: Initial version status
94 Returns:
95 Created PromptVersion
97 Raises:
98 VersioningError: If version format is invalid or version already exists
99 """
100 # Auto-increment version if not provided
101 if version is None:
102 version = await self._auto_increment_version(name, prompt_type)
103 # If no parent_version specified, use the latest version
104 if parent_version is None:
105 latest = await self.get_version(name, prompt_type)
106 if latest:
107 parent_version = latest.version_id
108 else:
109 # Validate version format
110 if not self.VERSION_PATTERN.match(version):
111 raise VersioningError(
112 f"Invalid version format: {version}. "
113 f"Expected semantic version (e.g., '1.0.0')"
114 )
116 # Check if version already exists
117 key = self._make_key(name, prompt_type)
118 existing_versions = await self.list_versions(name, prompt_type)
119 if any(v.version == version for v in existing_versions):
120 raise VersioningError(
121 f"Version {version} already exists for {name} ({prompt_type})"
122 )
124 # Generate unique version ID
125 version_id = str(uuid.uuid4())
127 # Create version object
128 prompt_version = PromptVersion(
129 version_id=version_id,
130 name=name,
131 prompt_type=prompt_type,
132 version=version,
133 template=template,
134 defaults=defaults or {},
135 validation=validation,
136 metadata=metadata or {},
137 created_at=datetime.utcnow(),
138 created_by=created_by,
139 parent_version=parent_version,
140 tags=tags or [],
141 status=status,
142 )
144 # Store version
145 self._versions[version_id] = prompt_version
147 # Update index
148 if key not in self._version_index:
149 self._version_index[key] = []
150 self._version_index[key].append(version_id)
152 # Persist to backend if available
153 if hasattr(self.storage, "set"):
154 await self._persist_version(prompt_version)
156 return prompt_version
158 async def get_version(
159 self,
160 name: str,
161 prompt_type: str,
162 version: str = "latest",
163 version_id: str | None = None,
164 ) -> PromptVersion | None:
165 """Retrieve a prompt version.
167 Args:
168 name: Prompt name
169 prompt_type: Prompt type
170 version: Version string or "latest" for most recent
171 version_id: Specific version ID (takes precedence over version)
173 Returns:
174 PromptVersion if found, None otherwise
175 """
176 # Direct lookup by version_id
177 if version_id:
178 return self._versions.get(version_id)
180 # Get all versions for this prompt
181 versions = await self.list_versions(name, prompt_type)
182 if not versions:
183 return None
185 # Return latest version
186 if version == "latest":
187 return self._get_latest_version(versions)
189 # Find specific version
190 for v in versions:
191 if v.version == version:
192 return v
194 return None
196 async def list_versions(
197 self,
198 name: str,
199 prompt_type: str,
200 tags: List[str] | None = None,
201 status: VersionStatus | None = None,
202 ) -> List[PromptVersion]:
203 """List all versions of a prompt.
205 Args:
206 name: Prompt name
207 prompt_type: Prompt type
208 tags: Filter by tags (returns versions with ANY of these tags)
209 status: Filter by status
211 Returns:
212 List of PromptVersion objects, sorted by version (newest first)
213 """
214 key = self._make_key(name, prompt_type)
215 version_ids = self._version_index.get(key, [])
217 versions = [self._versions[vid] for vid in version_ids]
219 # Apply filters
220 if tags:
221 versions = [v for v in versions if any(t in v.tags for t in tags)]
223 if status:
224 versions = [v for v in versions if v.status == status]
226 # Sort by version (newest first)
227 return sorted(versions, key=lambda v: self._parse_version(v.version), reverse=True)
229 async def tag_version(
230 self,
231 version_id: str,
232 tag: str,
233 ) -> PromptVersion:
234 """Add a tag to a version.
236 Args:
237 version_id: Version ID to tag
238 tag: Tag to add (e.g., "production", "deprecated")
240 Returns:
241 Updated PromptVersion
243 Raises:
244 VersioningError: If version not found
245 """
246 version = self._versions.get(version_id)
247 if not version:
248 raise VersioningError(f"Version not found: {version_id}")
250 if tag not in version.tags:
251 version.tags.append(tag)
253 # Persist if backend available
254 if hasattr(self.storage, "set"):
255 await self._persist_version(version)
257 return version
259 async def untag_version(
260 self,
261 version_id: str,
262 tag: str,
263 ) -> PromptVersion:
264 """Remove a tag from a version.
266 Args:
267 version_id: Version ID
268 tag: Tag to remove
270 Returns:
271 Updated PromptVersion
273 Raises:
274 VersioningError: If version not found
275 """
276 version = self._versions.get(version_id)
277 if not version:
278 raise VersioningError(f"Version not found: {version_id}")
280 if tag in version.tags:
281 version.tags.remove(tag)
283 # Persist if backend available
284 if hasattr(self.storage, "set"):
285 await self._persist_version(version)
287 return version
289 async def update_status(
290 self,
291 version_id: str,
292 status: VersionStatus,
293 ) -> PromptVersion:
294 """Update version status.
296 Args:
297 version_id: Version ID
298 status: New status
300 Returns:
301 Updated PromptVersion
303 Raises:
304 VersioningError: If version not found
305 """
306 version = self._versions.get(version_id)
307 if not version:
308 raise VersioningError(f"Version not found: {version_id}")
310 version.status = status
312 # Persist if backend available
313 if hasattr(self.storage, "set"):
314 await self._persist_version(version)
316 return version
318 async def delete_version(
319 self,
320 version_id: str,
321 ) -> bool:
322 """Delete a version.
324 Note: This permanently removes the version. Consider using
325 update_status() with ARCHIVED instead.
327 Args:
328 version_id: Version ID to delete
330 Returns:
331 True if deleted, False if not found
332 """
333 version = self._versions.get(version_id)
334 if not version:
335 return False
337 # Remove from index
338 key = self._make_key(version.name, version.prompt_type)
339 if key in self._version_index:
340 self._version_index[key] = [
341 vid for vid in self._version_index[key]
342 if vid != version_id
343 ]
345 # Remove from storage
346 del self._versions[version_id]
348 # Persist deletion if backend available
349 if hasattr(self.storage, "delete"):
350 await self.storage.delete(f"version:{version_id}")
352 return True
354 # ===== Helper Methods =====
356 def _make_key(self, name: str, prompt_type: str) -> str:
357 """Create index key for prompt name and type."""
358 return f"{name}:{prompt_type}"
360 def _parse_version(self, version: str) -> tuple:
361 """Parse semantic version into (major, minor, patch) tuple."""
362 match = self.VERSION_PATTERN.match(version)
363 if not match:
364 return (0, 0, 0)
365 return tuple(int(x) for x in match.groups())
367 def _get_latest_version(self, versions: List[PromptVersion]) -> PromptVersion | None:
368 """Get the latest version from a list."""
369 if not versions:
370 return None
372 # Filter out archived/deprecated if active versions exist
373 active = [v for v in versions if v.status in (VersionStatus.ACTIVE, VersionStatus.PRODUCTION)]
374 if active:
375 versions = active
377 # Sort by version number
378 sorted_versions = sorted(
379 versions,
380 key=lambda v: self._parse_version(v.version),
381 reverse=True
382 )
383 return sorted_versions[0]
385 async def _auto_increment_version(
386 self,
387 name: str,
388 prompt_type: str,
389 ) -> str:
390 """Auto-increment version number from latest version.
392 Increments patch version by default (1.0.0 -> 1.0.1).
393 If no versions exist, returns "1.0.0".
394 """
395 versions = await self.list_versions(name, prompt_type)
396 if not versions:
397 return "1.0.0"
399 latest = self._get_latest_version(versions)
400 if not latest:
401 return "1.0.0"
403 major, minor, patch = self._parse_version(latest.version)
404 return f"{major}.{minor}.{patch + 1}"
406 async def _persist_version(self, version: PromptVersion):
407 """Persist version to backend storage."""
408 if hasattr(self.storage, "set"):
409 key = f"version:{version.version_id}"
410 await self.storage.set(key, version.to_dict())
412 async def get_version_history(
413 self,
414 name: str,
415 prompt_type: str,
416 ) -> List[PromptVersion]:
417 """Get version history with parent relationships.
419 Returns versions in chronological order (oldest first).
421 Args:
422 name: Prompt name
423 prompt_type: Prompt type
425 Returns:
426 List of versions in chronological order
427 """
428 versions = await self.list_versions(name, prompt_type)
429 return sorted(versions, key=lambda v: v.created_at)