Coverage for frappe_manager / migration_manager / migration_discovery.py: 88%

1""" 

2Migration discovery - dynamically loads migration classes from migrations/ directory. 

3""" 

4 

5import importlib 

6import pkgutil 

7from pathlib import Path 

8from typing import TYPE_CHECKING 

9 

10from frappe_manager.logger import log 

11from frappe_manager.migration_manager.version import Version 

12from frappe_manager.utils.helpers import capture_and_format_exception 

13 

14if TYPE_CHECKING: 

15 from frappe_manager.migration_manager.migration_base import MigrationBase 

16 from frappe_manager.output_manager import OutputHandler 

17 

18 

19class MigrationDiscovery: 

20 """ 

21 Discovers and loads migration classes from the migrations/ directory. 

22 

23 Handles dynamic import and validation of migration modules. 

24 """ 

25 

26 def __init__(self, migrations_path: Path, output_handler: "OutputHandler"): 

27 self.migrations_path = migrations_path 

28 self.output = output_handler 

29 self.logger = log.get_logger() 

30 

31 def discover_migrations( 

32 self, 

33 from_version: Version, 

34 to_version: Version, 

35 migration_executor: "object", 

36 ) -> list["MigrationBase"]: 

37 """ 

38 Discover and load migrations needed to go from from_version to to_version. 

39 

40 Args: 

41 from_version: Starting version 

42 to_version: Target version 

43 migration_executor: MigrationExecutor instance to inject into migrations 

44 

45 Returns: 

46 Sorted list of migration instances to execute 

47 """ 

48 migrations = [] 

49 

50 for _, module_name, _ in pkgutil.iter_modules([str(self.migrations_path)]): 

51 try: 

52 migration_class = self._load_migration_class(module_name) 

53 if migration_class: 

54 migration_instance = self._instantiate_migration(migration_class, migration_executor) 

55 

56 if self._should_include_migration(migration_instance, from_version, to_version): 

57 migrations.append(migration_instance) 

58 

59 except Exception: 

60 exception_str = capture_and_format_exception() 

61 self.logger.error(f"Failed to register migration {module_name}: {exception_str}") 

62 self.output.warning( 

63 f"Skipping migration module '{module_name}' due to load failure. Check logs for details.", 

64 ) 

65 

66 return sorted(migrations, key=lambda m: m.version) 

67 

68 def _load_migration_class(self, module_name: str) -> type | None: 

69 """ 

70 Load a migration class from a module. 

71 

72 Returns None if module doesn't contain a valid migration class. 

73 """ 

74 module = importlib.import_module(f".migrations.{module_name}", "frappe_manager.migration_manager") 

75 

76 for attr_name in dir(module): 

77 attr = getattr(module, attr_name) 

78 

79 if self._is_valid_migration_class(attr): 

80 if attr.version != Version("0.0.0"): 

81 return attr 

82 

83 return None 

84 

85 def _is_valid_migration_class(self, obj: object) -> bool: 

86 """Check if object is a valid migration class.""" 

87 return ( 

88 isinstance(obj, type) 

89 and hasattr(obj, "up") 

90 and hasattr(obj, "down") 

91 and hasattr(obj, "set_migration_executor") 

92 and hasattr(obj, "version") 

93 ) 

94 

95 def _instantiate_migration( 

96 self, 

97 migration_class: type, 

98 migration_executor: "object", 

99 ) -> "MigrationBase": 

100 """Create migration instance and inject executor.""" 

101 migration = migration_class(output_handler=self.output) 

102 migration.set_migration_executor(migration_executor=migration_executor) 

103 return migration 

104 

105 def _should_include_migration( 

106 self, 

107 migration: "MigrationBase", 

108 from_version: Version, 

109 to_version: Version, 

110 ) -> bool: 

111 """ 

112 Check if migration version is in the range we need to execute. 

113 

114 Special handling for dev versions: If to_version is a dev/pre-release version 

115 (e.g., 0.19.0.dev0), we normalize it to the base release version (0.19.0) for 

116 comparison. This ensures migrations for the target release are included during 

117 development cycles. 

118 

119 In PEP 440, dev versions come BEFORE their release: 0.19.0.dev0 < 0.19.0 

120 Without normalization, migration 0.19.0 would be excluded when running 0.19.0.dev0. 

121 """ 

122 # Normalize to_version to base version (strips .devN, .a, .b, .rc suffixes) 

123 # This allows migrations for release X.Y.Z to run during X.Y.Z.devN development 

124 normalized_to = Version(to_version.base_version) 

125 return from_version < migration.version <= normalized_to