Coverage for frappe_manager / commands / migrate.py: 21%
117 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-07-02 18:13 +0530
1from enum import Enum
2from typing import Annotated
4import typer
5from typer_examples import example
6from rich.table import Table
8from frappe_manager import CLI_BENCH_CONFIG_FILE_NAME, CLI_BENCHES_DIRECTORY
9from frappe_manager.metadata_manager import FMConfigManager
10from frappe_manager.migration_manager.bench_migration_state import (
11 get_bench_migration_version,
12 set_bench_migration_version,
13)
14from frappe_manager.migration_manager.migration_executor import MigrationExecutor
15from frappe_manager.migration_manager.version import Version
16from frappe_manager.output_manager import get_global_output_handler, spinner
17from frappe_manager.utils.helpers import get_current_fm_version
20class MigrationFailureAction(str, Enum):
21 """Actions to take when migration fails."""
23 prompt = "prompt"
24 archive = "archive"
25 rollback = "rollback"
28@example(
29 "Migrate FM infrastructure only (safe)",
30 "",
31 detail="Applies only FM infrastructure migrations without touching benches. Safe to run for CLI updates.",
32)
33@example(
34 "Migrate specific bench",
35 "{benchname}",
36 detail="Runs migration steps for a specific bench to bring it in line with the FM version.",
37 benchname="mybench",
38)
39@example(
40 "Migrate all benches",
41 "--all-benches",
42 detail="Applies migrations to all benches managed by FM. Use with caution and consider backups.",
43)
44@example(
45 "Skip confirmation prompt",
46 "--all-benches --auto-proceed",
47 detail="Runs migrations across all benches without interactive prompts. Useful for automation.",
48)
49@example(
50 "Auto-proceed with auto-rollback on failure",
51 "--all-benches --auto-proceed --on-failure=rollback",
52 detail="Automatically proceeds with migrations and rolls back if a failure occurs.",
53)
54@example(
55 "Auto-proceed, archive failed benches (partial success OK)",
56 "--all-benches --auto-proceed --on-failure=archive",
57 detail="Archives benches that fail migration while continuing others; useful for large fleets.",
58)
59@example(
60 "Skip all backups (dangerous)",
61 "--all-benches --skip-all-backup",
62 detail="Disables taking backups before migration. This is risky and should only be used in controlled scenarios.",
63)
64@example(
65 "Exclude specific benches",
66 "--all-benches --exclude-bench mybench1,mybench2",
67 detail="Excludes specific benches from a full migration run.",
68)
69def migrate(
70 ctx: typer.Context,
71 benchname: Annotated[
72 str | None,
73 typer.Argument(help="Bench name to migrate"),
74 ] = None,
75 all_benches: Annotated[
76 bool,
77 typer.Option("--all-benches", help="Migrate all benches"),
78 ] = False,
79 skip_backup: Annotated[
80 bool,
81 typer.Option("--skip-all-backup", help="Skip all backups (DANGEROUS - use only if backups fail)"),
82 ] = False,
83 skip_backup_for: Annotated[
84 str | None,
85 typer.Option("--skip-backup-for", help="Skip backup for specific benches (comma-separated)"),
86 ] = None,
87 exclude_bench: Annotated[
88 str | None,
89 typer.Option("--exclude-bench", help="Exclude specific benches from migration (only with --all-benches)"),
90 ] = None,
91 auto_proceed: Annotated[
92 bool,
93 typer.Option("--auto-proceed", help="Skip migration confirmation prompt (proceed automatically)"),
94 ] = False,
95 rerun: Annotated[
96 bool,
97 typer.Option(
98 "--rerun",
99 help="Re-run migration even if already migrated (for testing idempotency). "
100 "Config transforms and supervisor regeneration are re-applied, but the "
101 "runtime environment is only rebuilt when Python/Node versions change.",
102 ),
103 ] = False,
104 on_failure: Annotated[
105 MigrationFailureAction | None,
106 typer.Option(
107 "--on-failure",
108 help="What to do if migration fails: prompt (ask user), archive (save failed benches), rollback (revert all)",
109 ),
110 ] = None,
111):
112 """
113 Migrate Frappe Manager to current version.
115 Migration operates at two levels:
116 - FM Infrastructure: CLI config + global database services (always checked and migrated if needed)
117 - Benches: Individual bench environments (you choose which ones to migrate)
119 Without arguments, migrates only FM infrastructure. Specify a benchname to migrate that bench,
120 or use --all-benches to migrate all benches. Use --auto-proceed to skip confirmation prompts.
121 Control failure handling with --on-failure: prompt (ask), archive (save failed), or rollback (revert all).
122 Use --rerun to test idempotency — re-applies all migration steps even when already up to date.
123 """
124 fm_config_manager: FMConfigManager = ctx.obj["fm_config_manager"]
125 output = get_global_output_handler()
127 failure_action = on_failure.value if on_failure else "prompt"
129 if benchname and all_benches:
130 output.display_error("Cannot specify both <benchname> and --all-benches")
131 output.stop()
132 typer.echo(ctx.get_help())
133 raise typer.Exit(1)
135 if exclude_bench and not all_benches:
136 output.display_error("--exclude-bench can only be used with --all-benches")
137 output.stop()
138 typer.echo(ctx.get_help())
139 raise typer.Exit(1)
141 current_version = Version(get_current_fm_version())
143 skip_backup_list = []
144 if skip_backup_for:
145 skip_backup_list = [b.strip() for b in skip_backup_for.split(",")]
147 exclude_bench_list = []
148 if exclude_bench:
149 exclude_bench_list = [b.strip() for b in exclude_bench.split(",")]
151 target_benches = None
152 if benchname:
153 bench_path = CLI_BENCHES_DIRECTORY / benchname
154 if not bench_path.exists():
155 output.display_error(f"Bench '{benchname}' does not exist")
156 raise typer.Exit(1)
157 target_benches = [benchname]
158 elif all_benches:
159 target_benches = []
160 if CLI_BENCHES_DIRECTORY.exists():
161 for bench_path in CLI_BENCHES_DIRECTORY.iterdir():
162 if bench_path.is_dir() and (bench_path / CLI_BENCH_CONFIG_FILE_NAME).exists():
163 if bench_path.name not in exclude_bench_list:
164 target_benches.append(bench_path.name)
166 fm_infrastructure_version = fm_config_manager.get_system_migration_version()
167 fm_infrastructure_needs_migration = rerun or (fm_infrastructure_version < current_version)
169 benches_checked = []
170 benches_migrated = []
171 benches_skipped = []
172 benches_failed = []
174 if not fm_infrastructure_needs_migration and not target_benches:
175 output.print("✓ FM infrastructure already up to date (no benches specified)")
176 raise typer.Exit(0)
178 if target_benches:
179 for bench_name in target_benches:
180 bench_path = CLI_BENCHES_DIRECTORY / bench_name
181 if bench_path.exists():
182 bench_version = get_bench_migration_version(bench_path)
183 benches_checked.append((bench_name, bench_version))
185 output_handler = get_global_output_handler()
187 migrations = MigrationExecutor(
188 fm_config_manager,
189 skip_backup=skip_backup,
190 skip_backup_for=skip_backup_list,
191 exclude_benches=exclude_bench_list,
192 auto_proceed=auto_proceed,
193 rerun=rerun,
194 on_failure=failure_action,
195 target_benches=target_benches,
196 migrate_fm_infrastructure=fm_infrastructure_needs_migration,
197 output_handler=output_handler,
198 )
200 with spinner(output_handler, "Starting migration..."):
201 migration_status = migrations.execute()
203 if not migration_status:
204 raise typer.Exit(1)
206 if target_benches:
207 for bench_name in target_benches:
208 if bench_name in migrations.migrate_benches:
209 bench_data = migrations.migrate_benches[bench_name]
210 last_migrated = bench_data["last_migration_version"]
212 # Compare base versions to handle dev releases (0.19.0.dev0 matches 0.19.0)
213 versions_match = (
214 last_migrated is not None and last_migrated.base_version == current_version.base_version
215 )
217 if versions_match and not bench_data["exception"]:
218 bench_path = CLI_BENCHES_DIRECTORY / bench_name
219 if bench_path.exists() and (bench_path / CLI_BENCH_CONFIG_FILE_NAME).exists():
220 set_bench_migration_version(bench_path, current_version)
221 benches_migrated.append(bench_name)
222 elif bench_data["exception"]:
223 benches_failed.append(bench_name)
224 else:
225 benches_skipped.append(bench_name)
227 if fm_infrastructure_needs_migration:
228 fm_config_manager.set_system_migration_version(current_version)
229 fm_config_manager.export_to_toml()
231 table = Table(show_header=False, box=None, padding=(0, 2, 0, 0))
233 show_infrastructure_status = fm_infrastructure_needs_migration or target_benches is None
235 if show_infrastructure_status:
236 if fm_infrastructure_needs_migration:
237 table.add_row(
238 "✅",
239 "[cyan]FM Infrastructure[/cyan]",
240 f"[yellow]v{fm_infrastructure_version}[/yellow] → [green]v{current_version}[/green]",
241 )
242 else:
243 table.add_row(
244 "⏭️ ",
245 "[cyan]FM Infrastructure[/cyan]",
246 f"[yellow]v{fm_infrastructure_version}[/yellow] (already up to date)",
247 )
249 if benches_migrated:
250 for bench_name in benches_migrated:
251 orig_version = next((v for n, v in benches_checked if n == bench_name), None)
252 table.add_row(
253 "✅",
254 f"[cyan]{bench_name}[/cyan]",
255 f"[yellow]v{orig_version}[/yellow] → [green]v{current_version}[/green]",
256 )
258 if benches_skipped:
259 for bench_name in benches_skipped:
260 orig_version = next((v for n, v in benches_checked if n == bench_name), None)
261 table.add_row("⏭️ ", f"[cyan]{bench_name}[/cyan]", f"[yellow]v{orig_version}[/yellow] (already up to date)")
263 if benches_failed:
264 for bench_name in benches_failed:
265 table.add_row("❌", f"[cyan]{bench_name}[/cyan]", "[red]Migration failed[/red]")
267 output.print_data(table)
269 if benches_failed:
270 output.display_error("Check logs for details", emoji_code=":warning:")