Coverage for frappe_manager / site_manager / bench_service.py: 44%
126 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-07-02 18:13 +0530
1"""
2BenchService - Service layer for bench operations
4This module provides a clean service layer between the CLI and domain models.
5It encapsulates bench creation, retrieval, and listing logic to reduce coupling
6between the CLI commands and internal implementation details.
8Benefits:
9- Single responsibility: manages bench lifecycle
10- Dependency injection: receives services, not globals
11- Testability: easy to mock in tests
12- Reusability: can be used by CLI, API, or other interfaces
13"""
15from pathlib import Path
17from rich.table import Table
19from frappe_manager.docker import ComposeFile, DockerClient
20from frappe_manager.logger import log
21from frappe_manager.output_manager import OutputHandler
22from frappe_manager.output_manager.rich_output import RichOutputHandler
23from frappe_manager.services_manager.services import ServicesManager
24from frappe_manager.site_manager.bench_config import BenchConfig, FMBenchEnvType
25from frappe_manager.site_manager.site import Bench
28class BenchService:
29 """
30 Service layer for bench operations.
32 Provides high-level operations for managing benches without exposing
33 internal implementation details to the CLI layer.
35 Attributes:
36 benches_directory: Root directory containing all benches
37 services: Global services manager instance
38 verbose: Whether to enable verbose output
40 Example:
41 >>> service = BenchService(CLI_BENCHES_DIRECTORY, services_manager)
42 >>> bench = service.get_bench("mysite.localhost")
43 >>> benches = service.list_benches()
44 """
46 def __init__(
47 self,
48 benches_directory: Path,
49 services: ServicesManager,
50 verbose: bool = False,
51 output_handler: OutputHandler | None = None,
52 ):
53 """
54 Initialize bench service.
56 Args:
57 benches_directory: Path to directory containing benches
58 services: Global services manager
59 verbose: Enable verbose output
60 output_handler: Handler for output operations
61 """
62 self.benches_directory = benches_directory
63 self.services = services
64 self.verbose = verbose
65 self.output = output_handler or RichOutputHandler()
66 self.logger = log.get_logger()
68 def get_bench(
69 self,
70 bench_name: str,
71 workers_check: bool = True,
72 admin_tools_check: bool = True,
73 ) -> Bench:
74 """
75 Get a bench instance by name.
77 This is a convenience method that wraps Bench.get_object() with
78 the service's configuration.
80 Args:
81 bench_name: Name of the bench to retrieve
82 workers_check: Whether to check worker status
83 admin_tools_check: Whether to check admin tools status
85 Returns:
86 Bench instance
88 Raises:
89 FileNotFoundError: If bench config not found
91 Example:
92 >>> bench = service.get_bench("mysite.localhost")
93 >>> bench.start()
94 """
95 return Bench.get_object(
96 bench_name=bench_name,
97 services=self.services,
98 workers_check=workers_check,
99 admin_tools_check=admin_tools_check,
100 verbose=self.verbose,
101 output_handler=self.output,
102 )
104 def create_bench(
105 self,
106 bench_name: str,
107 bench_config: BenchConfig,
108 is_template: bool = False,
109 ) -> Bench:
110 """
111 Create a new bench.
113 Handles all the setup for creating a new bench including:
114 - Creating directory structure
115 - Initializing docker compose files
116 - Creating bench instance
117 - Running bench creation process
119 Args:
120 bench_name: Name for the new bench
121 bench_config: Configuration for the bench
122 is_template: Whether to create a template bench
124 Returns:
125 Created Bench instance
127 Example:
128 >>> config = BenchConfig(name="site.localhost", ...)
129 >>> bench = service.create_bench("site.localhost", config)
130 """
131 bench_path = self.benches_directory / bench_name
132 compose_path = bench_path / "docker-compose.yml"
134 compose_file_manager = ComposeFile(compose_path)
135 docker_client = DockerClient(compose_file_path=compose_path, output=self.output)
137 from frappe_manager.logger.context import LoggerContext
138 from frappe_manager.logger.contextual import ContextualLogger
140 bench_logger = ContextualLogger(self.logger, context=LoggerContext(bench=bench_name, operation="create"))
142 bench = Bench(
143 logger=bench_logger,
144 path=bench_path,
145 name=bench_name,
146 bench_config=bench_config,
147 compose_file_manager=compose_file_manager,
148 docker_client=docker_client,
149 services=self.services,
150 verbose=self.verbose,
151 output_handler=self.output,
152 )
154 bench.create(is_template_bench=is_template)
155 return bench
157 def delete_bench(
158 self,
159 bench_name: str,
160 yes: bool = False,
161 delete_db_from_global_db: bool | None = None,
162 ) -> bool:
163 try:
164 bench = self.get_bench(bench_name, workers_check=False, admin_tools_check=False)
165 except FileNotFoundError:
166 bench = self._create_cleanup_bench(bench_name)
168 if yes:
169 from frappe_manager.output_manager import spinner
171 with spinner(self.output, "Removing bench"):
172 try:
173 bench.remove_certificate()
174 except Exception as e:
175 self.output.warning(str(e))
177 try:
178 self._handle_database_deletion(bench, delete_db_from_global_db)
179 except Exception as e:
180 self.output.warning(f"Database deletion failed: {e!s}")
181 self.output.warning("Continuing with bench removal...")
183 bench.remove_containers_and_dirs()
184 return True
185 return bench.remove_bench(delete_db_from_global_db=delete_db_from_global_db)
187 def discover_benches(self) -> dict[str, Path]:
188 """
189 Discover all benches in the benches directory.
191 Returns:
192 Dictionary mapping bench names to their docker-compose.yml paths
194 Example:
195 >>> benches = service.discover_benches()
196 >>> print(f"Found {len(benches)} benches")
197 """
198 benches = {}
200 if not self.benches_directory.exists():
201 return benches
203 for bench_dir in self.benches_directory.iterdir():
204 if not bench_dir.is_dir():
205 continue
207 bench_name = bench_dir.name
208 compose_file = bench_dir / "docker-compose.yml"
210 if compose_file.exists():
211 benches[bench_name] = compose_file
213 return benches
215 def get_bench_names(self) -> list[str]:
216 """
217 Get list of all bench names.
219 Returns:
220 List of bench names
222 Example:
223 >>> names = service.get_bench_names()
224 >>> for name in names:
225 ... bench = service.get_bench(name)
226 """
227 return list(self.discover_benches().keys())
229 def list_benches_table(self) -> Table:
230 """
231 Generate a formatted table of all benches.
233 Returns:
234 Rich Table object with bench information
236 Example:
237 >>> table = service.list_benches_table()
238 >>> self.output.print(table)
239 """
240 self.output.change_head("Generating bench list")
242 bench_dict = self.discover_benches()
244 if not bench_dict:
245 self.output.stop()
246 self.output.print(
247 "Seems like you haven't created any sites yet. "
248 "To create a bench, use the command: 'fm create <benchname>'.",
249 emoji_code=":white_check_mark:",
250 )
251 table = Table(show_lines=True, show_header=True, highlight=True)
252 table.add_column("Site")
253 table.add_column("Status", vertical="middle")
254 table.add_column("Path")
255 return table
257 table = Table(show_lines=True, show_header=True, highlight=True)
258 table.add_column("Site")
259 table.add_column("Status", vertical="middle")
260 table.add_column("Path")
262 for bench_name in bench_dict.keys():
263 try:
264 bench = self.get_bench(bench_name, workers_check=False, admin_tools_check=False)
266 row_data = f"[link=http://{bench.name}]{bench.name}[/link]"
267 path_data = f"[link=file://{bench.path}]{bench.path}[/link]"
269 status_color = "white"
270 status_msg = "Inactive"
272 if bench.running:
273 status_color = "green"
274 status_msg = "Active"
276 status_data = f"[{status_color}]{status_msg}[/{status_color}]"
278 table.add_row(row_data, status_data, path_data, style=f"{status_color}")
279 self.output.update_live(table, padding=(0, 0, 0, 0))
281 except FileNotFoundError as e:
282 self.output.warning(f"[red][bold]{bench_name}[/bold][/red] : Bench config not found at {e.filename}")
284 self.output.stop()
285 return table
287 def _create_cleanup_bench(self, bench_name: str) -> Bench:
288 """
289 Create a minimal bench instance for cleanup purposes.
291 Used when bench config is missing but we need to clean up containers/files.
293 Args:
294 bench_name: Name of bench to clean up
296 Returns:
297 Minimal Bench instance
298 """
299 import os
301 bench_path = self.benches_directory / bench_name
302 compose_path = bench_path / "docker-compose.yml"
304 compose_file_manager = ComposeFile(compose_path)
305 docker_client = DockerClient(compose_file_path=compose_path, output=self.output)
307 fake_config = BenchConfig(
308 name=bench_name,
309 userid=os.getuid(),
310 usergroup=os.getgid(),
311 apps_list=[],
312 developer_mode=False,
313 admin_tools=False,
314 admin_pass="pass",
315 environment_type=FMBenchEnvType.dev,
316 root_path=bench_path / "bench_config.toml",
317 admin_tools_username=None,
318 admin_tools_password=None,
319 github_token=None,
320 use_uv=True,
321 python_version=None,
322 node_version=None,
323 db_name=None,
324 migration_state=None,
325 )
327 from frappe_manager.logger.context import LoggerContext
328 from frappe_manager.logger.contextual import ContextualLogger
330 cleanup_logger = ContextualLogger(self.logger, context=LoggerContext(bench=bench_name, operation="cleanup"))
332 return Bench(
333 logger=cleanup_logger,
334 path=bench_path,
335 name=bench_name,
336 bench_config=fake_config,
337 compose_file_manager=compose_file_manager,
338 docker_client=docker_client,
339 services=self.services,
340 workers_check=False,
341 admin_tools_check=False,
342 verbose=self.verbose,
343 output_handler=self.output,
344 )
346 def _is_using_global_db(self, bench: Bench) -> bool:
347 """
348 Check if bench is using FM's managed global-db service.
350 Args:
351 bench: Bench instance to check
353 Returns:
354 True if bench uses global-db, False otherwise
355 """
356 try:
357 db_info = bench.database.get_connection_info()
358 db_host = db_info.get("host", "")
360 return db_host == "global-db"
361 except Exception:
362 return False
364 def _handle_database_deletion(self, bench: Bench, delete_db_from_global_db: bool | None):
365 """
366 Handle database deletion based on user preference and database location.
368 Args:
369 bench: Bench instance
370 delete_db_from_global_db: User preference for database deletion.
371 None = prompt if using global-db
372 True = delete from global-db
373 False = don't delete from global-db
374 """
375 is_global_db = self._is_using_global_db(bench)
377 if not is_global_db:
378 self.output.print("Bench is not using FM's managed global-db. Skipping database deletion")
379 return
381 should_delete = delete_db_from_global_db
383 if should_delete is None:
384 params = {
385 "prompt": f"🗄️ Do you want to remove the database '[bold]{bench.name}[/bold]' from global-db?",
386 "choices": ["yes", "no"],
387 "default": "yes",
388 "required_flag": "--delete-db-from-global-db or --no-delete-db-from-global-db",
389 }
390 choice = self.output.prompt_ask(**params)
391 should_delete = choice == "yes"
393 if should_delete:
394 bench.remove_database_and_user()
395 else:
396 self.output.print("Skipping database deletion from global-db")