Coverage for frappe_manager / site_manager / modules / bench_app.py: 11%

1""" 

2BenchAppManager - Frappe App Management Module 

3 

4This module handles all Frappe app-related operations within a bench including 

5app installation, removal, building, and branch management. 

6 

7Extracted from the monolithic Bench class and BenchOperations for better 

8separation of concerns. 

9""" 

10 

11import shlex 

12from collections.abc import Iterator 

13from pathlib import Path 

14from typing import cast 

15 

16from frappe_manager.docker import DockerClient, DockerException 

17from frappe_manager.docker.subprocess_output import SubprocessOutput 

18from frappe_manager.logger.contextual import ContextualLogger 

19from frappe_manager.output_manager import OutputHandler 

20from frappe_manager.output_manager.rich_output import RichOutputHandler 

21from frappe_manager.site_manager.bench_config import AppConfig, BenchConfig 

22from frappe_manager.site_manager.exceptions import ( 

23 BenchOperationBenchAppInSiteFailed, 

24 BenchOperationBenchBuildFailed, 

25 BenchOperationBenchInstallAppInPythonEnvFailed, 

26 BenchOperationBenchRemoveAppFromPythonEnvFailed, 

27 BenchOperationException, 

28) 

29from frappe_manager.site_manager.modules.app_cloner import AppCloner, AppClonerError 

30from frappe_manager.utils.docker import parameters_to_options 

31 

32 

33class BenchAppManager: 

34 """ 

35 Manages Frappe app operations within a bench. 

36 

37 This module is responsible for all app-related operations including: 

38 - App installation to Python environment 

39 - App installation to site 

40 - App removal from Python environment 

41 - App building (bench build) 

42 - App branch management 

43 - App listing 

44 

45 The module encapsulates bench command execution and provides a clean 

46 interface for app management operations. 

47 

48 Attributes: 

49 bench_name: Name of the bench 

50 bench_path: Path to the bench directory 

51 docker_client: Docker client for container operations 

52 bench_config: Bench configuration object 

53 logger: Logger instance 

54 frappe_bench_dir: Path to frappe-bench directory inside container 

55 bench_cli_cmd: Base bench command prefix 

56 

57 Example: 

58 >>> app_manager = BenchAppManager( 

59 ... bench_name="example.localhost", 

60 ... bench_path=Path("/home/user/frappe/example.localhost"), 

61 ... docker_client=docker_client, 

62 ... bench_config=bench_config, 

63 ... ) 

64 >>> app_manager.install_app_to_env("erpnext", branch="version-15") 

65 >>> app_manager.install_app_to_site("erpnext", "example.localhost") 

66 """ 

67 

68 def __init__( 

69 self, 

70 logger: ContextualLogger, 

71 bench_name: str, 

72 bench_path: Path, 

73 docker_client: DockerClient, 

74 bench_config: BenchConfig, 

75 output_handler: OutputHandler | None = None, 

76 ): 

77 """ 

78 Initialize BenchAppManager. 

79 

80 Args: 

81 logger: Contextual logger for audit/debug logging 

82 bench_name: Name of the bench 

83 bench_path: Path to the bench directory on host 

84 docker_client: Docker client for container operations 

85 bench_config: Bench configuration object 

86 output_handler: Handler for output operations 

87 """ 

88 self.logger = logger.child(component="app_manager") 

89 self.bench_name = bench_name 

90 self.bench_path = bench_path 

91 self.docker_client = docker_client 

92 self.bench_config = bench_config 

93 self.output = output_handler or RichOutputHandler() 

94 

95 self.frappe_bench_dir: Path = bench_path / "workspace" / "frappe-bench" 

96 self.bench_cli_cmd = ["/opt/user/.bin/bench"] 

97 

98 def get_current_runtime_versions(self, use_run: bool = False) -> dict[str, str | None]: 

99 """Get currently installed Python and Node versions from the container.""" 

100 import re 

101 

102 versions: dict[str, str | None] = {} 

103 versions["python"] = None 

104 versions["node"] = None 

105 

106 try: 

107 result = self._container_run( 

108 "/workspace/frappe-bench/env/bin/python --version", 

109 capture_output=True, 

110 raise_exception_obj=None, 

111 use_run=use_run, 

112 ) 

113 if result and result.exit_code == 0: 

114 output = " ".join(result.combined) 

115 match = re.search(r"Python (\d+\.\d+\.\d+)", output) 

116 if match: 

117 versions["python"] = match.group(1) 

118 except Exception: 

119 pass 

120 

121 try: 

122 result = self._container_run( 

123 "node --version", 

124 capture_output=True, 

125 raise_exception_obj=None, 

126 use_run=use_run, 

127 ) 

128 if result and result.exit_code == 0: 

129 output = " ".join(result.combined) 

130 match = re.search(r"v(\d+\.\d+\.\d+)", output) 

131 if match: 

132 versions["node"] = match.group(1) 

133 except Exception: 

134 pass 

135 

136 return versions 

137 

138 def setup_python_and_node_environments(self, use_run: bool = False, recreate_python_env: bool = False) -> bool: 

139 """ 

140 Setup Python and Node.js environments for the bench. 

141 

142 This method: 

143 1. Installs the required Python version via UV if not already present 

144 2. Creates venv with UV using that Python version 

145 3. Installs Node version via fnm if not present 

146 4. Sets fnm default to use that Node version 

147 

148 Returns: 

149 dict: { 

150 'venv_recreated': bool, 

151 'old_python_version': str or None, 

152 'old_node_version': str or None 

153 } 

154 """ 

155 from frappe_manager.site_manager.bench_config import ( 

156 parse_node_version_for_runtime, 

157 parse_python_version_for_runtime, 

158 ) 

159 

160 venv_recreated = False 

161 python_version_requirement = self.bench_config.python_version 

162 node_version_requirement = self.bench_config.node_version 

163 

164 if python_version_requirement: 

165 self.output.change_head("Checking Python version compatibility") 

166 

167 from frappe_manager.site_manager.bench_config import extract_python_version_requirement 

168 

169 frappe_app_path = self.bench_config.root_path / "workspace" / "frappe-bench" / "apps" / "frappe" 

170 frappe_python_req = None 

171 if frappe_app_path.exists(): 

172 frappe_python_req = extract_python_version_requirement(frappe_app_path) 

173 

174 try: 

175 check_current_version_cmd = "/workspace/frappe-bench/env/bin/python --version" 

176 result = self._container_run( 

177 check_current_version_cmd, 

178 capture_output=True, 

179 raise_exception_obj=None, 

180 use_run=use_run, 

181 ) 

182 

183 if result and result.exit_code == 0: 

184 import re 

185 

186 current_version_output = " ".join(result.combined) 

187 version_match = re.search(r"Python (\d+)\.(\d+)\.(\d+)", current_version_output) 

188 

189 if version_match: 

190 current_major = int(version_match.group(1)) 

191 current_minor = int(version_match.group(2)) 

192 

193 if self._python_version_satisfies_requirement( 

194 current_major, 

195 current_minor, 

196 python_version_requirement, 

197 ): 

198 if frappe_python_req: 

199 self.output.print( 

200 f"✓ Python {current_major}.{current_minor} already satisfies {python_version_requirement} " 

201 f"(Frappe requires: {frappe_python_req}) - skipping installation", 

202 ) 

203 else: 

204 self.output.print( 

205 f"✓ Python {current_major}.{current_minor} already satisfies {python_version_requirement} - skipping installation", 

206 ) 

207 python_version_requirement = None 

208 else: 

209 self.output.print( 

210 f"Python {current_major}.{current_minor} does not satisfy {python_version_requirement} - will recreate venv", 

211 ) 

212 

213 except Exception as e: 

214 self.logger.debug(f"Could not check current Python version: {e}") 

215 

216 if python_version_requirement: 

217 python_version = parse_python_version_for_runtime(python_version_requirement) 

218 if python_version: 

219 self.output.change_head(f"Setting up Python environment for requirement: {python_version_requirement}") 

220 try: 

221 scan_pythons_cmd = """ 

222if [ -d /workspace/frappe-bench/.uv/python ]; then 

223 for dir in /workspace/frappe-bench/.uv/python/cpython-*; do 

224 if [ -d "$dir" ]; then 

225 basename "$dir" 

226 fi 

227 done 2>/dev/null | sort -u 

228fi 

229""" 

230 result = self._container_run( 

231 scan_pythons_cmd, 

232 capture_output=True, 

233 raise_exception_obj=None, 

234 use_run=use_run, 

235 ) 

236 

237 selected_python_full = None 

238 selected_version = None 

239 if result and result.exit_code == 0: 

240 import re 

241 

242 candidates = [] 

243 

244 for line in result.combined: 

245 if "/usr/bin" in line or "download available" in line: 

246 continue 

247 

248 match = re.search(r"cpython-(\d+)\.(\d+)\.(\d+)", line) 

249 if match: 

250 major = int(match.group(1)) 

251 minor = int(match.group(2)) 

252 patch = int(match.group(3)) 

253 if self._python_version_satisfies_requirement(major, minor, python_version_requirement): 

254 # Store the FULL directory name with platform suffix, not just cpython-X.Y.Z 

255 candidates.append((major, minor, patch, line.strip())) 

256 

257 if candidates: 

258 candidates.sort(reverse=True) 

259 selected_version = candidates[0] 

260 selected_python_full = selected_version[3] 

261 self.output.print( 

262 f"Found Python {selected_version[0]}.{selected_version[1]}.{selected_version[2]} satisfying {python_version_requirement}", 

263 ) 

264 

265 if not selected_python_full: 

266 self.output.print(f"Installing Python {python_version} via uv..") 

267 quoted_pkg = shlex.quote(f"cpython-{python_version}") 

268 install_cmd = f"uv python install {quoted_pkg}" 

269 self._container_run(install_cmd, raise_exception_obj=None, use_run=use_run) 

270 

271 detect_installed_cmd = ( 

272 f"ls -1 /workspace/frappe-bench/.uv/python/ | grep '^{quoted_pkg}' | sort -V | tail -1" 

273 ) 

274 result = self._container_run( 

275 detect_installed_cmd, 

276 capture_output=True, 

277 raise_exception_obj=None, 

278 use_run=use_run, 

279 ) 

280 if result and result.exit_code == 0 and result.stdout: 

281 selected_python_full = result.stdout[0].strip() 

282 else: 

283 selected_python_full = f"cpython-{python_version}" 

284 

285 self.output.print(f"Installed Python {python_version} via uv") 

286 

287 if selected_python_full: 

288 update_symlink_cmd = f""" 

289 cd /workspace/frappe-bench/.uv 

290 rm -f python-default 

291 ln -sf python/{selected_python_full} python-default 

292 """ 

293 self._container_run(update_symlink_cmd, raise_exception_obj=None, use_run=use_run) 

294 

295 if recreate_python_env: 

296 self.output.change_head(f"Creating virtual environment with {selected_python_full}") 

297 quoted_python = shlex.quote(selected_python_full) 

298 recreate_venv_cmd = f""" 

299 cd /workspace/frappe-bench 

300 if [ -d env ]; then 

301 timestamp=$(date +%Y%m%d_%H%M%S) 

302 mv env env.bak-$timestamp 

303 fi 

304 uv venv env --python {quoted_python} --seed --link-mode=copy 

305 """ 

306 self._container_run(recreate_venv_cmd, raise_exception_obj=None, use_run=use_run) 

307 selected_version_str = ( 

308 f"{selected_version[0]}.{selected_version[1]}.{selected_version[2]}" 

309 if selected_version 

310 else selected_python_full 

311 ) 

312 self.output.print(f"Created virtual environment with Python {selected_version_str}") 

313 venv_recreated = True 

314 else: 

315 self.output.print(f"Skipping venv recreation - Python {selected_python_full} set as default") 

316 

317 except Exception as e: 

318 self.output.warning(f"Failed to setup Python {python_version}: {e}") 

319 self.output.warning("Continuing with default Python version") 

320 

321 node_version_requirement = self.bench_config.node_version 

322 if node_version_requirement: 

323 self.output.change_head("Checking Node version compatibility") 

324 

325 from frappe_manager.site_manager.bench_config import extract_node_version_requirement 

326 

327 frappe_app_path = self.bench_config.root_path / "workspace" / "frappe-bench" / "apps" / "frappe" 

328 frappe_node_req = None 

329 if frappe_app_path.exists(): 

330 frappe_node_req = extract_node_version_requirement(frappe_app_path) 

331 

332 try: 

333 check_current_node_cmd = "node --version" 

334 result = self._container_run( 

335 check_current_node_cmd, 

336 capture_output=True, 

337 raise_exception_obj=None, 

338 use_run=use_run, 

339 ) 

340 

341 if result and result.exit_code == 0: 

342 import re 

343 

344 current_node_output = " ".join(result.combined) 

345 version_match = re.search(r"v(\d+\.\d+\.\d+)", current_node_output) 

346 

347 if version_match: 

348 full_version = version_match.group(1) 

349 current_major = int(full_version.split(".")[0]) 

350 

351 if self._node_version_satisfies_requirement(current_major, node_version_requirement): 

352 if frappe_node_req: 

353 self.output.print( 

354 f"Node {current_major} already satisfies {node_version_requirement} " 

355 f"(Frappe requires: {frappe_node_req}) - skipping installation", 

356 ) 

357 else: 

358 self.output.print( 

359 f"Node {current_major} already satisfies {node_version_requirement} - skipping installation", 

360 ) 

361 node_version_requirement = None 

362 else: 

363 self.output.print( 

364 f"Node {current_major} does not satisfy {node_version_requirement} - will install", 

365 ) 

366 

367 except Exception as e: 

368 self.logger.debug(f"Could not check current Node version: {e}") 

369 

370 if node_version_requirement: 

371 node_version = parse_node_version_for_runtime(node_version_requirement) 

372 if node_version: 

373 self.output.change_head(f"Setting up Node {node_version} environment") 

374 try: 

375 check_cmd = f"fnm list | grep 'v{node_version}' || true" 

376 result = self._container_run( 

377 check_cmd, 

378 capture_output=True, 

379 raise_exception_obj=None, 

380 use_run=use_run, 

381 ) 

382 

383 needs_install = True 

384 if result and result.combined: 

385 output_text = " ".join(result.combined) 

386 if f"v{node_version}" in output_text: 

387 needs_install = False 

388 self.output.print(f"Node {node_version} already installed") 

389 

390 if needs_install: 

391 self.output.print(f"Installing Node {node_version} via fnm..") 

392 install_cmd = f"fnm install {node_version}" 

393 install_result = self._container_run( 

394 install_cmd, 

395 capture_output=True, 

396 raise_exception_obj=None, 

397 use_run=use_run, 

398 ) 

399 if install_result and install_result.exit_code == 0: 

400 self.output.print(f"Installed Node {node_version} via fnm") 

401 else: 

402 raise Exception( 

403 f"fnm install {node_version} failed with exit code {install_result.exit_code if install_result else 'unknown'}", 

404 ) 

405 

406 set_default_cmd = f"fnm default {node_version}" 

407 default_result = self._container_run( 

408 set_default_cmd, 

409 capture_output=True, 

410 raise_exception_obj=None, 

411 use_run=use_run, 

412 ) 

413 if default_result and default_result.exit_code == 0: 

414 self.output.print(f"Set Node {node_version} as default") 

415 else: 

416 self.output.warning(f"Could not set Node {node_version} as default, but continuing") 

417 

418 # Verify yarn is available (should be auto-enabled by FNM_COREPACK_ENABLED) 

419 verify_yarn = f"yarn --version" 

420 yarn_result = self._container_run( 

421 verify_yarn, capture_output=True, raise_exception_obj=None, use_run=use_run 

422 ) 

423 if yarn_result and yarn_result.exit_code == 0: 

424 self.output.print(f"Yarn is available for Node {node_version}") 

425 else: 

426 self.output.warning( 

427 f"Yarn not available after Node {node_version} installation - corepack may have failed" 

428 ) 

429 

430 except Exception as e: 

431 self.output.warning(f"Failed to setup Node {node_version}: {e}") 

432 self.output.warning("Continuing with default Node version") 

433 

434 return venv_recreated 

435 

436 def _python_version_satisfies_requirement(self, current_major: int, current_minor: int, requirement: str) -> bool: 

437 import re 

438 

439 requirement = requirement.strip() 

440 current_version = f"{current_major}.{current_minor}" 

441 

442 if ">=" in requirement and "<" in requirement: 

443 match_min = re.search(r">=(\d+)\.(\d+)", requirement) 

444 match_max = re.search(r"<(\d+)\.(\d+)", requirement) 

445 

446 if match_min and match_max: 

447 min_major, min_minor = int(match_min.group(1)), int(match_min.group(2)) 

448 max_major, max_minor = int(match_max.group(1)), int(match_max.group(2)) 

449 

450 if current_major > min_major or (current_major == min_major and current_minor >= min_minor): 

451 if current_major < max_major or (current_major == max_major and current_minor < max_minor): 

452 return True 

453 

454 elif requirement.startswith("^"): 

455 match = re.search(r"\^(\d+)\.(\d+)", requirement) 

456 if match: 

457 req_major, req_minor = int(match.group(1)), int(match.group(2)) 

458 return current_major == req_major and current_minor >= req_minor 

459 

460 elif re.match(r"^\d+\.\d+$", requirement): 

461 return current_version == requirement 

462 

463 return False 

464 

465 def _node_version_satisfies_requirement(self, current_major: int, requirement: str) -> bool: 

466 import re 

467 

468 requirement = requirement.strip() 

469 

470 if ">=" in requirement: 

471 match = re.search(r">=(\d+)", requirement) 

472 if match: 

473 min_major = int(match.group(1)) 

474 return current_major >= min_major 

475 

476 elif requirement.startswith("^"): 

477 match = re.search(r"\^(\d+)", requirement) 

478 if match: 

479 req_major = int(match.group(1)) 

480 return current_major == req_major 

481 

482 elif re.match(r"^\d+$", requirement): 

483 return current_major == int(requirement) 

484 

485 return False 

486 

487 def install_apps( 

488 self, 

489 apps_list: list[AppConfig], 

490 github_token: str | None = None, 

491 use_uv: bool = True, 

492 force_reinstall: bool = False, 

493 clone_only: bool = False, 

494 skip_clone: bool = False, 

495 use_run: bool = False, 

496 ) -> list[AppConfig]: 

497 """ 

498 Install apps to the bench (clone, install Python/Node deps, build). 

499 

500 Returns: 

501 List of AppConfig objects with corrected module names after cloning 

502 """ 

503 if not apps_list: 

504 self.output.print("No apps to install") 

505 return [] 

506 

507 apps_config = apps_list 

508 

509 if not skip_clone: 

510 self.output.change_head(f"Cloning {len(apps_config)} apps in parallel") 

511 try: 

512 cloner = AppCloner( 

513 logger=self.logger, 

514 apps_dir=self.frappe_bench_dir / "apps", 

515 github_token=github_token, 

516 output_handler=self.output, 

517 ) 

518 cloned_apps = cloner.clone_apps_parallel(apps_config) 

519 self.output.print(f"Cloned {len(cloned_apps)} apps successfully") 

520 

521 self._update_apps_txt(apps_config) 

522 self._update_apps_list_with_corrected_names(apps_config) 

523 except AppClonerError as e: 

524 raise BenchOperationBenchInstallAppInPythonEnvFailed( 

525 bench_name=self.bench_name, 

526 app_name="multiple apps", 

527 ) from e 

528 

529 if clone_only: 

530 return apps_config 

531 

532 self.output.change_head("Installing Python dependencies") 

533 self._install_python_deps_with_uv(apps_config, use_uv=use_uv, use_run=use_run) 

534 self.output.print("Installed Python dependencies") 

535 

536 self.output.change_head("Installing Node dependencies") 

537 self._install_node_deps(use_run=use_run) 

538 self.output.print("Installed Node dependencies") 

539 

540 self.output.change_head("Building frontend assets") 

541 self.build(use_run=use_run) 

542 self.output.print("Built frontend assets") 

543 

544 return apps_config 

545 

546 def _install_python_deps_with_uv(self, apps: list[AppConfig], use_uv: bool = True, use_run: bool = False) -> None: 

547 """ 

548 Install Python dependencies using UV (much faster than pip). 

549 

550 Strategy: 

551 1. Try UV first (10-100x faster) 

552 2. Fall back to pip if UV fails 

553 

554 UV targets the bench virtual environment at /workspace/frappe-bench/env 

555 """ 

556 if use_uv: 

557 try: 

558 check_cmd = "which uv" 

559 self._container_run(check_cmd, capture_output=True, raise_exception_obj=None, use_run=use_run) 

560 

561 for app in apps: 

562 install_cmd = [ 

563 "uv", 

564 "pip", 

565 "install", 

566 "--python", 

567 "/workspace/frappe-bench/env/bin/python", 

568 "--no-cache-dir", 

569 "-e", 

570 f"apps/{app.name}", 

571 ] 

572 install_cmd_str = " ".join(install_cmd) 

573 self._container_run(install_cmd_str, raise_exception_obj=None, use_run=use_run) 

574 

575 return 

576 except Exception as uv_error: 

577 self.logger.debug(f"UV installation failed, attempting uv pip fallback: {uv_error}") 

578 

579 try: 

580 for app in apps: 

581 fallback_cmd = " ".join( 

582 [ 

583 "uv", 

584 "pip", 

585 "install", 

586 "--python", 

587 "/workspace/frappe-bench/env/bin/python", 

588 "--no-cache-dir", 

589 "-e", 

590 f"apps/{app.name}", 

591 ] 

592 ) 

593 self._container_run(fallback_cmd, use_run=use_run) 

594 self.output.warning("UV installation failed on first attempt, succeeded on retry") 

595 except Exception: 

596 raise 

597 

598 return 

599 

600 for app in apps: 

601 install_cmd = " ".join( 

602 [ 

603 "uv", 

604 "pip", 

605 "install", 

606 "--python", 

607 "/workspace/frappe-bench/env/bin/python", 

608 "--no-cache-dir", 

609 "-e", 

610 f"apps/{app.name}", 

611 ] 

612 ) 

613 self._container_run(install_cmd, use_run=use_run) 

614 

615 def _install_node_deps(self, use_run: bool = False) -> None: 

616 """Install Node.js dependencies for all apps.""" 

617 node_cmd = " ".join(self.bench_cli_cmd + ["setup", "requirements", "--node"]) 

618 self._container_run(node_cmd, use_run=use_run) 

619 

620 def _update_apps_txt(self, apps: list[AppConfig]) -> None: 

621 """ 

622 Update apps.txt to register newly cloned apps. 

623 

624 The apps.txt file lists all apps in the bench and is required 

625 for Frappe to recognize and install apps to sites. 

626 

627 Args: 

628 apps: List of AppConfig objects for apps to add 

629 """ 

630 apps_txt_path = self.frappe_bench_dir / "sites" / "apps.txt" 

631 

632 # Read existing apps 

633 existing_apps = [] 

634 if apps_txt_path.exists(): 

635 existing_apps = apps_txt_path.read_text().strip().split("\n") 

636 existing_apps = [app.strip() for app in existing_apps if app.strip()] 

637 

638 # Add new apps (avoid duplicates) 

639 for app_config in apps: 

640 if app_config.name not in existing_apps: 

641 existing_apps.append(app_config.name) 

642 

643 # Write back to apps.txt 

644 apps_txt_path.write_text("\n".join(existing_apps) + "\n") 

645 self.logger.debug(f"Updated apps.txt with {len(apps)} new apps") 

646 

647 def _update_apps_list_with_corrected_names(self, apps_config: list[AppConfig]) -> None: 

648 for i, app_config in enumerate(apps_config): 

649 if i < len(self.bench_config.apps_list): 

650 self.bench_config.apps_list[i] = app_config 

651 

652 def install_app_to_env( 

653 self, 

654 app: str, 

655 branch: str | None = None, 

656 overwrite: bool = True, 

657 skip_assets: bool = False, 

658 ) -> None: 

659 """ 

660 Install an app to the bench Python environment. 

661 

662 This runs 'bench get-app' to clone and install the app in the 

663 bench's Python environment. 

664 

665 Args: 

666 app: App name or URL 

667 branch: Git branch to install (optional) 

668 overwrite: Whether to overwrite if app exists 

669 skip_assets: Whether to skip building assets 

670 

671 Raises: 

672 BenchOperationBenchInstallAppInPythonEnvFailed: If installation fails 

673 

674 Example: 

675 >>> app_manager.install_app_to_env("erpnext", branch="version-15") 

676 """ 

677 parameters: dict = { 

678 "branch": branch, 

679 "overwrite": overwrite, 

680 "skip_assets": skip_assets, 

681 } 

682 

683 app_install_env_command = self.bench_cli_cmd + ["get-app"] 

684 app_install_env_command += parameters_to_options(parameters, exclude=["app"]) 

685 app_install_env_command += [app] 

686 

687 app_install_env_command = " ".join(app_install_env_command) 

688 app_install_exception = BenchOperationBenchInstallAppInPythonEnvFailed(bench_name=self.bench_name, app_name=app) 

689 

690 self._container_run( 

691 app_install_env_command, 

692 raise_exception_obj=app_install_exception, 

693 ) 

694 

695 def remove_app_from_env( 

696 self, 

697 app: str, 

698 no_backup: bool = True, 

699 force: bool = True, 

700 ) -> None: 

701 """ 

702 Remove an app from the bench Python environment. 

703 

704 This runs 'bench remove-app' to remove the app from the environment. 

705 

706 Args: 

707 app: App name to remove 

708 no_backup: Skip backup before removal 

709 force: Force removal without confirmation 

710 

711 Raises: 

712 BenchOperationBenchRemoveAppFromPythonEnvFailed: If removal fails 

713 

714 Example: 

715 >>> app_manager.remove_app_from_env("erpnext") 

716 """ 

717 parameters: dict = { 

718 "no_backup": no_backup, 

719 "force": force, 

720 } 

721 

722 app_rm_env_command = self.bench_cli_cmd + ["remove-app"] 

723 app_rm_env_command += parameters_to_options(parameters, exclude=["app"]) 

724 app_rm_env_command += [app] 

725 

726 app_rm_env_command = " ".join(app_rm_env_command) 

727 

728 self._container_run( 

729 app_rm_env_command, 

730 raise_exception_obj=BenchOperationBenchRemoveAppFromPythonEnvFailed( 

731 bench_name=self.bench_name, 

732 app_name=app, 

733 ), 

734 ) 

735 

736 def install_app_to_site( 

737 self, 

738 app: str, 

739 site_name: str | None = None, 

740 ) -> None: 

741 """ 

742 Install an app to a Frappe site. 

743 

744 This runs 'bench --site <site> install-app' to install the app 

745 to the specified site. 

746 

747 Args: 

748 app: App name to install 

749 site_name: Site name. Defaults to bench_name. 

750 

751 Raises: 

752 BenchOperationBenchAppInSiteFailed: If installation fails 

753 

754 Example: 

755 >>> app_manager.install_app_to_site("erpnext", "example.localhost") 

756 """ 

757 if site_name is None: 

758 site_name = self.bench_name 

759 

760 app_install_site_command = self.bench_cli_cmd + ["--site", site_name] 

761 app_install_site_command += ["install-app", app] 

762 app_install_site_command = " ".join(app_install_site_command) 

763 

764 self._container_run( 

765 app_install_site_command, 

766 raise_exception_obj=BenchOperationBenchAppInSiteFailed(bench_name=self.bench_name, app_name=app), 

767 ) 

768 

769 def install_apps_to_site( 

770 self, 

771 site_name: str | None = None, 

772 ) -> None: 

773 """ 

774 Install all available apps to a site in the order specified in bench_config. 

775 

776 Installs apps in the same order as provided by the user to respect dependencies. 

777 

778 Args: 

779 site_name: Site name. Defaults to bench_name. 

780 

781 Example: 

782 >>> app_manager.install_apps_to_site("example.localhost") 

783 """ 

784 if site_name is None: 

785 site_name = self.bench_name 

786 

787 for app_config in self.bench_config.apps_list: 

788 app_name = app_config.name 

789 self.output.change_head(f"Installing app {app_name} in site") 

790 self.install_app_to_site(app_name, site_name) 

791 self.output.print(f"Installed app {app_name} in site") 

792 

793 def build(self, app_list: list[str] | None = None, use_run: bool = False) -> None: 

794 """ 

795 Build bench assets. 

796 

797 This runs 'bench build' to compile and bundle frontend assets. 

798 

799 Args: 

800 app_list: List of specific apps to build. If None, builds all apps. 

801 use_run: If True, use 'docker compose run --rm' instead of 'exec' 

802 

803 Raises: 

804 BenchOperationBenchBuildFailed: If build fails 

805 

806 Example: 

807 >>> app_manager.build() # Build all apps 

808 >>> app_manager.build(["frappe", "erpnext"]) # Build specific apps 

809 """ 

810 build_cmd = self.bench_cli_cmd + ["build"] 

811 

812 if app_list is not None: 

813 for app in app_list: 

814 build_cmd += ["--app"] + [app] 

815 

816 build_exception = BenchOperationBenchBuildFailed(bench_name=self.bench_name, apps=app_list) 

817 

818 build_cmd = " ".join(build_cmd) 

819 self._container_run(build_cmd, build_exception, use_run=use_run) 

820 

821 def get_installed_apps_list(self) -> list[Path]: 

822 """ 

823 Get list of installed apps in the bench. 

824 

825 Returns list of app directories found in the apps folder. 

826 

827 Returns: 

828 List of Path objects for each app directory 

829 

830 Example: 

831 >>> apps = app_manager.get_installed_apps_list() 

832 >>> print([app.name for app in apps]) 

833 ['frappe', 'erpnext', 'hrms'] 

834 """ 

835 apps_dir = self.frappe_bench_dir / "apps" 

836 apps_dirs: list[Path] = [item for item in apps_dir.iterdir() if item.is_dir()] 

837 return apps_dirs 

838 

839 def _filter_docker_warnings(self, output: SubprocessOutput) -> SubprocessOutput: 

840 """Filter out Docker Compose warning messages from captured output.""" 

841 import re 

842 

843 if not output.combined: 

844 return output 

845 

846 filtered_lines = [] 

847 docker_warning_pattern = re.compile(r'^time=".*?" level=(warning|info|error) msg=') 

848 

849 for line in output.combined: 

850 if not docker_warning_pattern.match(line.strip()): 

851 filtered_lines.append(line) 

852 

853 output.combined = filtered_lines 

854 return output 

855 

856 def _container_run( 

857 self, 

858 command: str, 

859 raise_exception_obj: BenchOperationException | None = None, 

860 capture_output: bool = False, 

861 user: str = "frappe", 

862 workdir: str = "/workspace/frappe-bench", 

863 service: str = "frappe", 

864 use_run: bool = False, 

865 ) -> SubprocessOutput | None: 

866 """ 

867 Execute a command inside the bench container. 

868 

869 This is an internal helper method that wraps docker_client.compose.exec 

870 or docker_client.compose.run depending on use_run parameter.