Coverage for pyPhotoAlbum/version_manager.py: 71%
133 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-20 12:55 +0000
1"""
2Version management and migration system for pyPhotoAlbum projects
3"""
5import os
6import uuid
7from datetime import datetime, timezone
8from typing import Dict, Any, Optional, Callable, List
10# Current data version - increment when making breaking changes to data format
11CURRENT_DATA_VERSION = "3.0"
13# Version history and compatibility information
14VERSION_HISTORY = {
15 "1.0": {
16 "description": "Initial format with basic serialization",
17 "released": "2024-01-01",
18 "breaking_changes": [],
19 "compatible_with": ["1.0"],
20 },
21 "2.0": {
22 "description": "Fixed asset path handling - paths now stored relative to project folder",
23 "released": "2025-01-11",
24 "breaking_changes": [
25 "Asset paths changed from absolute/full-project-relative to project-relative",
26 "Added automatic path normalization for legacy projects",
27 ],
28 "compatible_with": ["1.0", "2.0"], # 2.0 can read 1.0 with migration
29 },
30 "3.0": {
31 "description": "Added merge conflict resolution support with UUIDs, timestamps, and project IDs",
32 "released": "2025-01-22",
33 "breaking_changes": [
34 "Added required UUID fields to all pages and elements",
35 "Added created/last_modified timestamps to projects, pages, and elements",
36 "Added project_id for merge detection (same ID = merge, different ID = concatenate)",
37 "Added deletion tracking (deleted flag and deleted_at timestamp)",
38 ],
39 "compatible_with": ["1.0", "2.0", "3.0"], # 3.0 can read older versions with migration
40 },
41}
44class VersionCompatibility:
45 """Handles version compatibility checks and migrations"""
47 @staticmethod
48 def is_compatible(file_version: str) -> bool:
49 """
50 Check if a file version is compatible with the current version.
52 Args:
53 file_version: Version string from the file
55 Returns:
56 True if compatible, False otherwise
57 """
58 current_info = VERSION_HISTORY.get(CURRENT_DATA_VERSION, {})
59 compatible_versions = current_info.get("compatible_with", [])
60 return file_version in compatible_versions
62 @staticmethod
63 def needs_migration(file_version: str) -> bool:
64 """
65 Check if a file needs migration to work with current version.
67 Args:
68 file_version: Version string from the file
70 Returns:
71 True if migration is needed, False otherwise
72 """
73 # If versions don't match but are compatible, migration may be needed
74 return file_version != CURRENT_DATA_VERSION and VersionCompatibility.is_compatible(file_version)
76 @staticmethod
77 def get_version_info(version: str) -> Optional[Dict[str, Any]]:
78 """Get information about a specific version."""
79 return VERSION_HISTORY.get(version)
81 @staticmethod
82 def get_migration_path(from_version: str, to_version: str) -> Optional[List[str]]:
83 """
84 Get the migration path from one version to another.
86 Args:
87 from_version: Starting version
88 to_version: Target version
90 Returns:
91 List of version steps needed, or None if no path exists
92 """
93 # For now, we only support direct migration paths
94 # In the future, we could implement multi-step migrations
96 if from_version == to_version:
97 return []
99 from_info = VERSION_HISTORY.get(from_version)
100 to_info = VERSION_HISTORY.get(to_version)
102 if not from_info or not to_info:
103 return None
105 # Check if direct migration is possible
106 compatible_versions = to_info.get("compatible_with", [])
107 if from_version in compatible_versions:
108 return [from_version, to_version]
110 return None
113class DataMigration:
114 """Handles data migrations between versions"""
116 # Registry of migration functions
117 _migrations: Dict[tuple, Callable] = {}
119 @classmethod
120 def register_migration(cls, from_version: str, to_version: str):
121 """Decorator to register a migration function"""
123 def decorator(func):
124 cls._migrations[(from_version, to_version)] = func
125 return func
127 return decorator
129 @classmethod
130 def migrate(cls, data: Dict[str, Any], from_version: str, to_version: str) -> Dict[str, Any]:
131 """
132 Migrate data from one version to another.
134 Args:
135 data: Project data dictionary
136 from_version: Current version of the data
137 to_version: Target version
139 Returns:
140 Migrated data dictionary
141 """
142 if from_version == to_version:
143 return data
145 # Get migration path
146 migration_path = VersionCompatibility.get_migration_path(from_version, to_version)
147 if not migration_path:
148 raise ValueError(f"No migration path from {from_version} to {to_version}")
150 # Apply migrations in sequence
151 current_data = data
152 for i in range(len(migration_path) - 1):
153 step_from = migration_path[i]
154 step_to = migration_path[i + 1]
155 migration_key = (step_from, step_to)
157 if migration_key in cls._migrations:
158 print(f"Applying migration: {step_from} → {step_to}")
159 current_data = cls._migrations[migration_key](current_data)
160 else:
161 print(f"Warning: No explicit migration for {step_from} → {step_to}, using as-is")
163 return current_data
166# Register migrations
169@DataMigration.register_migration("1.0", "2.0")
170def migrate_1_0_to_2_0(data: Dict[str, Any]) -> Dict[str, Any]:
171 """
172 Migrate from version 1.0 to 2.0.
174 Main changes:
175 - Asset paths are normalized to be relative to project folder
176 - This is now handled automatically in load_from_zip via _normalize_asset_paths
177 """
178 print("Migration 1.0 → 2.0: Asset paths will be normalized during load")
180 # Update version in data
181 data["data_version"] = "2.0"
183 # Note: Actual path normalization is handled in load_from_zip
184 # This migration mainly updates the version number
186 return data
189@DataMigration.register_migration("2.0", "3.0")
190def migrate_2_0_to_3_0(data: Dict[str, Any]) -> Dict[str, Any]:
191 """
192 Migrate from version 2.0 to 3.0.
194 Main changes:
195 - Add UUIDs to all pages and elements
196 - Add timestamps (created, last_modified) to project, pages, and elements
197 - Add project_id to project
198 - Add deletion tracking (deleted, deleted_at) to pages and elements
199 """
200 print("Migration 2.0 → 3.0: Adding UUIDs, timestamps, and project_id")
202 # Get current timestamp for migration
203 now = datetime.now(timezone.utc).isoformat()
205 # Add project-level fields
206 if "project_id" not in data:
207 data["project_id"] = str(uuid.uuid4())
208 print(f" Generated project_id: {data['project_id']}")
210 if "created" not in data:
211 data["created"] = now
213 if "last_modified" not in data:
214 data["last_modified"] = now
216 # Migrate pages
217 for page_data in data.get("pages", []):
218 # Add UUID
219 if "uuid" not in page_data:
220 page_data["uuid"] = str(uuid.uuid4())
222 # Add timestamps
223 if "created" not in page_data:
224 page_data["created"] = now
225 if "last_modified" not in page_data:
226 page_data["last_modified"] = now
228 # Add deletion tracking
229 if "deleted" not in page_data:
230 page_data["deleted"] = False
231 if "deleted_at" not in page_data:
232 page_data["deleted_at"] = None
234 # Migrate elements in page layout
235 layout_data = page_data.get("layout", {})
236 for element_data in layout_data.get("elements", []):
237 # Add UUID
238 if "uuid" not in element_data:
239 element_data["uuid"] = str(uuid.uuid4())
241 # Add timestamps
242 if "created" not in element_data:
243 element_data["created"] = now
244 if "last_modified" not in element_data:
245 element_data["last_modified"] = now
247 # Add deletion tracking
248 if "deleted" not in element_data:
249 element_data["deleted"] = False
250 if "deleted_at" not in element_data:
251 element_data["deleted_at"] = None
253 # Update version
254 data["data_version"] = "3.0"
256 print(f" Migrated {len(data.get('pages', []))} pages to v3.0")
258 return data
261def check_version_compatibility(file_version: str, file_path: str = "") -> tuple[bool, Optional[str]]:
262 """
263 Check version compatibility and provide user-friendly messages.
265 Args:
266 file_version: Version from the file
267 file_path: Optional path to the file (for error messages)
269 Returns:
270 Tuple of (is_compatible, error_message)
271 """
272 if not file_version:
273 return True, None # No version specified, assume compatible
275 if VersionCompatibility.is_compatible(file_version):
276 if VersionCompatibility.needs_migration(file_version):
277 print(f"File version {file_version} is compatible but needs migration to {CURRENT_DATA_VERSION}")
278 return True, None
280 # Not compatible
281 file_info = VersionCompatibility.get_version_info(file_version)
282 current_info = VersionCompatibility.get_version_info(CURRENT_DATA_VERSION)
284 error_msg = f"Incompatible file version: {file_version}\n\n"
285 error_msg += f"This file was created with version {file_version}, "
286 error_msg += f"but this application uses version {CURRENT_DATA_VERSION}.\n\n"
288 if file_info:
289 error_msg += f"File version info:\n"
290 error_msg += f" Description: {file_info.get('description', 'Unknown')}\n"
291 error_msg += f" Released: {file_info.get('released', 'Unknown')}\n"
292 breaking_changes = file_info.get("breaking_changes", [])
293 if breaking_changes:
294 error_msg += f" Breaking changes:\n"
295 for change in breaking_changes:
296 error_msg += f" - {change}\n"
298 error_msg += f"\nPlease use a compatible version of pyPhotoAlbum to open this file."
300 return False, error_msg
303def format_version_info() -> str:
304 """Format version information for display"""
305 info = [
306 f"pyPhotoAlbum Data Format Version: {CURRENT_DATA_VERSION}",
307 "",
308 "Version History:",
309 ]
311 for version in sorted(VERSION_HISTORY.keys(), reverse=True):
312 version_info = VERSION_HISTORY[version]
313 info.append(f"\n Version {version}")
314 info.append(f" Description: {version_info.get('description', 'Unknown')}")
315 info.append(f" Released: {version_info.get('released', 'Unknown')}")
317 breaking_changes = version_info.get("breaking_changes", [])
318 if breaking_changes:
319 info.append(f" Breaking changes:")
320 for change in breaking_changes:
321 info.append(f" - {change}")
323 return "\n".join(info)