Coverage for frappe_manager / docker / compose_file.py: 61%
531 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-07-02 18:13 +0530
1import copy
2from pathlib import Path
3from typing import Any
5from jinja2 import Template
6from ruamel.yaml import YAML
7from ruamel.yaml.comments import CommentedMap as OrderedDict
8from ruamel.yaml.comments import CommentedSeq as OrderedList
10from frappe_manager import CLI_DEFAULT_DELIMETER
11from frappe_manager.docker import DockerVolumeMount
12from frappe_manager.docker.compose_exceptions import ComposeSecretNotFoundError, ComposeServiceNotFound
13from frappe_manager.migration_manager.version import Version
14from frappe_manager.output_manager import get_global_output_handler
15from frappe_manager.utils.helpers import get_template_path, represent_null_empty, get_docker_image_tag
16from frappe_manager.utils.site import parse_docker_volume
18yaml = YAML(typ="rt")
19yaml.representer.ignore_aliases = lambda *args: True
21# Set the default flow style to None to preserve the null representation
22yaml.default_flow_style = False
23yaml.default_style = None
26class ComposeFile:
27 yml: dict[Any, Any]
29 def __init__(
30 self,
31 loadfile: Path,
32 template_name: str = "docker-compose.tmpl",
33 template_dir: str | None = None,
34 auto_save: bool = True,
35 ):
36 self.compose_path: Path = loadfile
37 self.template_name = template_name
38 self.is_template_loaded = False
40 self.template_dir = "templates"
42 if template_dir:
43 self.template_dir = template_dir
45 # New: Transaction support
46 self._auto_save = auto_save
47 self._pending_changes: list[tuple[str, Any]] = []
48 self._snapshot: dict | None = None
50 # check for if the docker-compose.yml file is present if not then use template provided
51 if self.exists():
52 with open(self.compose_path) as f:
53 self.yml = yaml.load(f)
54 else:
55 self.yml = self.load_template()
56 self.is_template_loaded = True
58 def exists(self):
59 """
60 Check if the compose file exists.
62 Returns:
63 bool: True if the compose file exists, False otherwise.
64 """
65 return self.compose_path.exists()
67 def get_compose_path(self):
68 """
69 Returns the path of the compose file.
70 """
71 return self.compose_path
73 def load_template(self):
74 """
75 Load the template file and return its contents as a YAML object.
76 Renders Jinja2 template variables for dynamic Docker image tags.
78 Returns:
79 dict: The contents of the template file as a YAML object.
80 """
81 template_path: Path = get_template_path(self.template_name, self.template_dir)
83 # Render Jinja2 template with dynamic image tags
84 template = Template(template_path.read_text())
85 image_tag = get_docker_image_tag()
86 rendered_template = template.render(frappe_image_tag=image_tag, nginx_image_tag=image_tag)
88 # Parse rendered YAML
89 yml = yaml.load(rendered_template)
90 return yml
92 def set_container_names(self, prefix):
93 """
94 Sets the container names for each service in the Compose file.
96 Args:
97 prefix (str): The prefix to be added to the container names.
98 """
99 for service in self.yml["services"].keys():
100 self.yml["services"][service]["container_name"] = prefix + CLI_DEFAULT_DELIMETER + service
102 def get_container_names(self) -> dict:
103 """
104 Returns a dictionary of container names for each service defined in the Compose file.
106 Returns:
107 dict: A dictionary where the keys are service names and the values are container names.
108 """
109 container_names: dict = {}
110 if self.exists():
111 services = self.get_services_list()
112 for service in services:
113 container_names[service] = self.yml["services"][service]["container_name"]
114 return container_names
116 def get_services_list(self, exclude_disabled: bool = False) -> list:
117 """
118 Returns a list of services defined in the Compose file.
120 Args:
121 exclude_disabled: When True, services marked as disabled via the
122 ``disabled`` profile are excluded from the returned list.
124 Returns:
125 list: A list of service names. May be filtered if ``exclude_disabled``
126 is True.
127 """
128 services = list(self.yml["services"].keys())
129 if exclude_disabled:
130 services = [s for s in services if not self.is_service_profile_disabled(s)]
131 return services
133 def is_services_name_same_as_template(self):
134 """
135 Checks if the service names in the current Compose file are the same as the template file.
137 Returns:
138 bool: True if the service names are the same, False otherwise.
139 """
140 template_yml = self.load_template()
141 template_service_name_list = list(template_yml["services"].keys())
142 template_service_name_list.sort()
143 current_service_name_list = list(self.yml["services"].keys())
144 current_service_name_list.sort()
145 return current_service_name_list == template_service_name_list
147 def set_user(self, service, uid, gid):
148 """
149 Set the user for a specific service in the Compose file.
151 Args:
152 service (str): The name of the service.
153 uid (str): The user ID.
154 gid (str): The group ID.
155 """
156 try:
157 self.yml["services"][service]["user"] = f"{uid}:{gid}"
158 except KeyError as e:
159 output = get_global_output_handler()
160 output.error("Issue in docker template. Not able to set user.", e)
162 def get_user(self, service):
163 """
164 Get the user associated with the specified service.
166 Args:
167 service (str): The name of the service.
169 Returns:
170 str or None: The user associated with the service, or None if not found.
171 """
172 try:
173 user = self.yml[service]["user"]
174 uid = user.split(":")[0]
175 uid = user.split(":")[1]
177 except KeyError:
178 return None
179 return user
181 def set_root_networks_name(self, networks_name, prefix, external: bool = False):
182 """
183 Sets the name of the top-level network in the Compose file.
185 Args:
186 networks_name (str): The name of the network.
187 prefix (str): The prefix to be added to the network name.
188 """
189 if not self.yml["networks"][networks_name]:
190 self.yml["networks"][networks_name] = {"name": prefix + f"{CLI_DEFAULT_DELIMETER}network"}
191 else:
192 self.yml["networks"][networks_name]["name"] = prefix + f"{CLI_DEFAULT_DELIMETER}network"
193 self.yml["networks"][networks_name]["external"] = external
195 def set_network_alias(self, service_name, network_name, alias: list = []):
196 """
197 Sets the network alias for a given service in the Compose file.
199 Args:
200 service_name (str): The name of the service.
201 network_name (str): The name of the network.
202 alias (list, optional): List of network aliases to be set. Defaults to [].
204 Returns:
205 bool: True if the network alias is set successfully, False otherwise.
206 """
207 if alias:
208 try:
209 all_networks = self.yml["services"][service_name]["networks"]
210 if network_name in all_networks:
211 self.yml["services"][service_name]["networks"][network_name] = {"aliases": alias}
212 return True
213 except KeyError as e:
214 return False
215 else:
216 return False
218 def get_network_alias(self, service_name, network_name) -> list | None:
219 """
220 Retrieves the network aliases for a given service and network name.
222 Args:
223 service_name (str): The name of the service.
224 network_name (str): The name of the network.
226 Returns:
227 list | None: A list of network aliases if found, otherwise None.
228 """
229 try:
230 all_networks = self.yml["services"][service_name]["networks"]
231 if network_name not in all_networks:
232 return None
234 aliases = self.yml["services"][service_name]["networks"][network_name]["aliases"]
235 return aliases
236 except KeyError as e:
237 return None
239 def get_version(self):
240 """
241 Get the version of the compose file.
243 Returns:
244 int: The version of the compose file, or 0 if the version is not specified.
245 """
246 try:
247 compose_version = self.yml["x-version"]
248 return Version(compose_version)
249 except KeyError:
250 return Version("0.0.0")
252 def set_version(self, version):
253 """
254 Sets the version of the Compose file.
256 Args:
257 version (str): The version to set.
259 Returns:
260 None
261 """
262 self.yml["x-version"] = version
264 def get_all_users(self):
265 """
266 Retrieves a dictionary of all users defined in the Compose file.
268 Returns:
269 dict: A dictionary where the keys are service names and the values are dictionaries
270 containing the user's UID and GID.
271 """
272 users: dict = {}
274 if self.exists():
275 services = self.get_services_list()
276 for service in services:
277 if "user" in self.yml["services"][service]:
278 user_data = self.yml["services"][service]["user"]
279 uid = user_data.split(":")[0]
280 gid = user_data.split(":")[1]
281 users[service] = {"uid": uid, "gid": gid}
282 return users
284 def set_all_users(self, users: dict):
285 for service in users:
286 user_data = users[service]
287 if isinstance(user_data, tuple):
288 uid, gid = user_data
289 else:
290 uid = user_data["uid"]
291 gid = user_data["gid"]
292 self.set_user(service, uid, gid)
294 def get_all_envs(self) -> dict[Any, Any]:
295 """
296 Retrieves all the environment variables for each service in the Compose file.
298 Returns:
299 dict: A dictionary containing the service names as keys and their respective environment variables as values.
300 """
301 envs = {}
303 for service in self.yml["services"].keys():
304 try:
305 env = self.yml["services"][service]["environment"]
306 envs[service] = env
307 except KeyError:
308 pass
310 return envs
312 def set_all_envs(self, environments: dict, append: bool = True):
313 """
314 Sets environment variables for all containers in the Compose file.
316 Args:
317 environments (dict): A dictionary containing container names as keys and environment variables as values.
319 """
320 for container_name in environments:
321 self.set_envs(container_name, environments[container_name], append=append)
323 def get_all_labels(self):
324 """
325 Retrieves all the labels for each service in the Compose file.
327 Returns:
328 dict: A dictionary containing the service names as keys and their respective labels as values.
329 """
330 labels = {}
331 for service in self.yml["services"].keys():
332 try:
333 label = self.yml["services"][service]["labels"]
334 labels[service] = label
335 except KeyError:
336 pass
337 return labels
339 def set_all_labels(self, labels: dict):
340 """
341 Sets labels for all containers in the ComposeFile.
343 Args:
344 labels (dict): A dictionary containing container names as keys and labels as values.
345 """
346 for container_name in labels:
347 self.set_labels(container_name, labels[container_name])
349 def get_all_extrahosts(self):
350 """
351 Returns a dictionary of all the extra hosts for each service in the Compose file.
353 Returns:
354 dict: A dictionary where the keys are the service names and the values are the extra hosts.
355 """
356 extrahosts = {}
357 for service in self.yml["services"].keys():
358 try:
359 extrahost = self.yml["services"][service]["extra_hosts"]
360 extrahosts[service] = extrahost
361 except KeyError:
362 pass
363 return extrahosts
365 def set_all_extrahosts(self, extrahosts: dict, skip_not_found: bool = False):
366 """
367 Sets the extrahosts for all containers in the ComposeFile.
369 Args:
370 extrahosts (dict): A dictionary containing container names as keys and their corresponding extrahosts as values.
371 skip_not_found (bool, optional): If True, skips setting extrahosts for containers that are not found. Defaults to False.
372 """
373 for container_name in extrahosts:
374 self.set_extrahosts(container_name, extrahosts[container_name])
376 def set_envs(self, container: str, env: dict, append=False):
377 """
378 Sets the environment variables for a specific container in the Compose file.
380 Args:
381 container (str): The name of the container.
382 env (dict): A dictionary containing the environment variables to be set.
383 append (bool, optional): If True, appends the new environment variables to the existing ones.
384 """
385 new_env = OrderedDict(env)
387 if append and type(env) == dict:
388 prev_env = self.get_envs(container)
389 if prev_env:
390 if not type(prev_env) == OrderedList:
391 env = OrderedDict(env)
392 new_env = prev_env | env
394 try:
395 self.yml["services"][container]["environment"] = new_env
396 except KeyError as e:
397 pass
399 def get_envs(self, container: str) -> dict:
400 """
401 Get the environment variables for a specific container.
403 Args:
404 container (str): The name of the container.
406 Returns:
407 dict: A dictionary containing the environment variables for the container.
408 Returns None if the container or environment variables are not found.
409 """
410 try:
411 env = self.yml["services"][container]["environment"]
412 return env
413 except KeyError:
414 return None
416 def set_labels(self, container: str, labels: dict):
417 """
418 Sets the labels for a specific container in the Compose file.
420 Args:
421 container (str): The name of the container.
422 labels (dict): A dictionary containing the labels to be set.
424 """
425 try:
426 self.yml["services"][container]["labels"] = labels
427 except KeyError as e:
428 pass
430 def get_labels(self, container: str) -> dict:
431 """
432 Get the labels of a specific container.
434 Args:
435 container (str): The name of the container.
437 Returns:
438 dict: The labels of the container, or None if the container or labels are not found.
439 """
440 try:
441 labels = self.yml["services"][container]["labels"]
442 return labels
443 except KeyError:
444 return None
446 def set_extrahosts(self, container: str, extrahosts: list):
447 """
448 Set the extra hosts for a specific container in the Compose file.
450 Args:
451 container (str): The name of the container.
452 extrahosts (list): A list of extra hosts to be added.
454 """
455 try:
456 self.yml["services"][container]["extra_hosts"] = extrahosts
457 except KeyError as e:
458 pass
460 def get_extrahosts(self, container: str) -> list:
461 """
462 Get the extra hosts for a specific container.
464 Args:
465 container (str): The name of the container.
467 Returns:
468 list: A list of extra hosts for the container, or None if not found.
469 """
470 try:
471 extra_hosts = self.yml["services"][container]["extra_hosts"]
472 return extra_hosts
473 except KeyError:
474 return None
476 def write_to_file(self):
477 """
478 Writes the Docker Compose file to the specified path.
479 """
480 try:
481 # saving the docker compose to the directory
482 with open(self.compose_path, "w") as f:
483 yaml.dump(self.yml, f, transform=represent_null_empty)
484 except Exception as e:
485 output = get_global_output_handler()
486 output.error("Error in writing compose file.", e)
488 def get_all_volumes(self):
489 """
490 Get all the root volumes.
491 """
493 try:
494 volumes = self.yml["volumes"]
495 except KeyError as e:
496 return {}
498 return volumes
500 def get_all_services_volumes(self) -> dict[str, list[DockerVolumeMount]]:
501 """
502 Get all the volume mounts mapped by service name.
504 Returns:
505 dict[str, List[DockerVolumeMount]]: Dictionary mapping service names to their volume mounts
506 """
507 volumes_map: dict[str, list[DockerVolumeMount]] = {}
509 services = self.get_services_list()
510 for service in services:
511 volumes = self.get_service_volumes(service)
512 volumes_map[service] = volumes
514 return volumes_map
516 def set_all_services_volumes(self, volumes_map: dict[str, list[DockerVolumeMount]]) -> None:
517 """
518 Set volume mounts for all services.
520 Args:
521 volumes_map (dict[str, List[DockerVolumeMount]]): Dictionary mapping service names to their volume mounts
522 """
523 services = self.get_services_list()
524 for service in services:
525 if service in volumes_map:
526 self.set_service_volumes(service, volumes_map[service])
528 def get_service_volumes(self, service: str) -> list[DockerVolumeMount]:
529 """
530 Get specific service volume mounts.
531 """
532 volumes_set = set()
534 try:
535 volumes_list = self.yml["services"][service]["volumes"]
536 for volume in volumes_list:
537 volumes_set.add(volume)
538 except KeyError as e:
539 raise ComposeServiceNotFound(service_name=service)
541 volumes_list = []
543 for volume in volumes_set:
544 volumes_list.append(parse_docker_volume(volume, self.get_all_volumes(), self.compose_path))
546 return volumes_list
548 def set_service_volumes(self, service: str, volumes: list[DockerVolumeMount]) -> None:
549 """
550 Set specific service volume mounts.
551 """
552 try:
553 # Convert DockerVolumeMount objects to strings
554 volumes_list = [str(volume) for volume in volumes]
556 self.yml["services"][service]["volumes"] = volumes_list
557 except KeyError as e:
558 raise ComposeServiceNotFound(service_name=service)
560 def set_root_volumes_names(self, volume_prefix: str) -> None:
561 """
562 Set names for root level volumes in the compose file with the given prefix.
564 Args:
565 volume_prefix (str): Prefix to add to volume names
566 """
567 try:
568 volumes = self.yml.get("volumes", {})
569 if volumes:
570 for volume_name in volumes:
571 if volumes[volume_name] is None:
572 volumes[volume_name] = {}
573 volumes[volume_name]["name"] = volume_prefix + CLI_DEFAULT_DELIMETER + volume_name
575 except KeyError as e:
576 output = get_global_output_handler()
577 output.warning(f"Error setting volume names: {e!s}")
579 def set_secret_file_path(self, secret_name, file_path):
580 try:
581 self.yml["secrets"][secret_name]["file"] = file_path
582 except KeyError:
583 output = get_global_output_handler()
584 output.warning("Not able to set secrets in compose")
586 def get_secret_file_path(self, secret_name) -> Path:
587 try:
588 file_path = self.yml["secrets"][secret_name]["file"]
589 return Path(file_path)
590 except KeyError:
591 raise ComposeSecretNotFoundError(secret_name, str(self.compose_path.absolute()))
593 def remove_secrets_from_container(self, container):
594 try:
595 del self.yml["services"][container]["secrets"]
596 except KeyError:
597 output = get_global_output_handler()
598 output.warning(f"Not able to remove secrets from {container}")
600 def remove_root_secrets_compose(self):
601 try:
602 del self.yml["secrets"]
603 except KeyError:
604 output = get_global_output_handler()
605 output.warning("root level secrets not present")
607 def remove_container_user(self, container):
608 try:
609 del self.yml["services"][container]["user"]
610 except KeyError:
611 output = get_global_output_handler()
612 output.warning("user not present")
614 def get_all_images(self):
615 """
616 Retrieves all the images for each service in the Compose file.
618 Returns:
619 dict: A dictionary containing the service names as keys and their respective image names and tags as values.
620 """
621 images = {}
622 for service in self.yml["services"].keys():
623 try:
624 image = self.yml["services"][service]["image"]
625 name, tag = image.split(":") if ":" in image else (image, "latest")
626 images[service] = {"name": name, "tag": tag, "image": image}
627 except KeyError:
628 pass
629 return images
631 def set_all_images(self, images: dict):
632 """
633 Sets the image for all services in the ComposeFile.
635 Args:
636 images (dict): A dictionary containing the service names as keys and the image names and tags as values.
637 """
638 for service, image_info in images.items():
639 image = f'{image_info["name"]}:{image_info["tag"]}'
640 if service in self.yml["services"]:
641 self.yml["services"][service]["image"] = image
643 def set_service_command(self, service: str, command: str) -> None:
644 """
645 Set the command for a specific service in the compose file.
647 Args:
648 service (str): The name of the service
649 command (str): The command to set for the service
650 """
651 if service not in self.yml["services"]:
652 raise KeyError(f"Service {service} not found in compose file")
654 self.yml["services"][service]["command"] = command
656 def get_service_command(self, service: str) -> str:
657 """
658 Get the command for a specific service from the compose file.
660 Args:
661 service (str): The name of the service
663 Returns:
664 str: The command configured for the service, or empty string if not set
666 Raises:
667 KeyError: If the service doesn't exist in the compose file
668 """
669 if service not in self.yml["services"]:
670 raise KeyError(f"Service {service} not found in compose file")
672 return self.yml["services"][service].get("command", "")
674 def set_service_restart(self, service: str, restart_policy: str) -> None:
675 if service not in self.yml["services"]:
676 raise KeyError(f"Service {service} not found in compose file")
678 self.yml["services"][service]["restart"] = restart_policy
680 def set_all_services_restart(self, restart_policy: str) -> None:
681 services = self.get_services_list()
682 for service in services:
683 self.set_service_restart(service, restart_policy)
685 def get_service_restart(self, service: str) -> str | None:
686 if service not in self.yml["services"]:
687 raise KeyError(f"Service {service} not found in compose file")
689 return self.yml["services"][service].get("restart", None)
691 # ==================== NEW: Transaction Support ====================
693 def _apply_change(self, change: tuple[str, Any]):
694 """
695 Internal: Apply a single pending change.
697 Args:
698 change: Tuple containing (change_type, *args)
699 """
700 change_type = change[0]
702 if change_type == "envs":
703 _, envs, append = change
704 self.set_all_envs(envs, append=append)
705 elif change_type == "labels":
706 _, labels = change
707 self.set_all_labels(labels)
708 elif change_type == "prefix":
709 _, prefix, network_name = change
710 self.set_container_names(prefix)
711 self.set_root_volumes_names(prefix)
712 self.set_root_networks_name(network_name, prefix)
713 elif change_type == "version":
714 _, version = change
715 self.set_version(version)
716 elif change_type == "users":
717 _, users = change
718 self.set_all_users(users)
719 elif change_type == "images":
720 _, images = change
721 self.set_all_images(images)
722 elif change_type == "volumes":
723 _, volumes = change
724 self.set_all_services_volumes(volumes)
725 elif change_type == "extrahosts":
726 _, extrahosts = change
727 self.set_all_extrahosts(extrahosts)
728 elif change_type == "restart":
729 _, restart_policy = change
730 self.set_all_services_restart(restart_policy)
732 def commit(self) -> "ComposeFile":
733 """
734 Apply all pending changes and save to file.
736 Returns:
737 Self for chaining
738 """
739 for change in self._pending_changes:
740 self._apply_change(change)
741 self._pending_changes.clear()
742 self.write_to_file()
743 return self
745 def rollback(self) -> "ComposeFile":
746 """
747 Discard all pending changes without applying them.
749 Returns:
750 Self for chaining
751 """
752 self._pending_changes.clear()
753 return self
755 # ==================== NEW: Builder/Fluent Interface ====================
757 def with_envs(self, envs: dict, append: bool = True) -> "ComposeFile":
758 """
759 Fluent setter for environment variables.
761 Args:
762 envs: Dictionary of service names to environment variables
763 append: If True, append to existing envs; if False, replace
765 Returns:
766 Self for chaining
767 """
768 self._pending_changes.append(("envs", envs, append))
769 return self
771 def with_labels(self, labels: dict) -> "ComposeFile":
772 """
773 Fluent setter for labels.
775 Args:
776 labels: Dictionary of service names to labels
778 Returns:
779 Self for chaining
780 """
781 self._pending_changes.append(("labels", labels))
782 return self
784 def with_prefix(self, prefix: str, network_name: str = "site-network") -> "ComposeFile":
785 """
786 Set prefix for containers, volumes, and networks at once.
788 Args:
789 prefix: Prefix to apply
790 network_name: Network name to configure (default: "site-network")
792 Returns:
793 Self for chaining
794 """
795 self._pending_changes.append(("prefix", prefix, network_name))
796 return self
798 def with_version(self, version: str) -> "ComposeFile":
799 """
800 Fluent setter for compose file version.
802 Args:
803 version: Version string to set
805 Returns:
806 Self for chaining
807 """
808 self._pending_changes.append(("version", version))
809 return self
811 def with_users(self, users: dict) -> "ComposeFile":
812 converted_users = {}
813 for service, user_data in users.items():
814 if isinstance(user_data, tuple):
815 converted_users[service] = {"uid": str(user_data[0]), "gid": str(user_data[1])}
816 else:
817 converted_users[service] = user_data
818 self._pending_changes.append(("users", converted_users))
819 return self
821 def with_images(self, images: dict) -> "ComposeFile":
822 """
823 Fluent setter for service images.
825 Args:
826 images: Dictionary of service names to image info
828 Returns:
829 Self for chaining
830 """
831 self._pending_changes.append(("images", images))
832 return self
834 def with_volumes(self, volumes: dict[str, list[DockerVolumeMount]]) -> "ComposeFile":
835 """
836 Fluent setter for service volumes.
838 Args:
839 volumes: Dictionary mapping service names to volume mounts
841 Returns:
842 Self for chaining
843 """
844 self._pending_changes.append(("volumes", volumes))
845 return self
847 def with_extrahosts(self, extrahosts: dict) -> "ComposeFile":
848 """
849 Fluent setter for extra hosts.
851 Args:
852 extrahosts: Dictionary of service names to extra hosts
854 Returns:
855 Self for chaining
856 """
857 self._pending_changes.append(("extrahosts", extrahosts))
858 return self
860 def with_restart(self, restart_policy: str) -> "ComposeFile":
861 """
862 Fluent setter for restart policy (applies to all services).
864 Args:
865 restart_policy: Docker restart policy ("no", "always", "on-failure", "unless-stopped")
867 Returns:
868 Self for chaining
869 """
870 self._pending_changes.append(("restart", restart_policy))
871 return self
873 # ==================== NEW: Atomic Configuration Methods ====================
875 def configure_bench(
876 self,
877 prefix: str,
878 version: str,
879 envs: dict | None = None,
880 labels: dict | None = None,
881 users: dict | None = None,
882 network_name: str = "site-network",
883 auto_save: bool = True,
884 ) -> "ComposeFile":
885 """
886 Configure a complete bench in one call.
888 Args:
889 prefix: Container/volume/network prefix
890 version: Compose file version
891 envs: Environment variables by service
892 labels: Labels by service
893 users: User configurations by service (supports both dict and tuple formats)
894 network_name: Network name to configure
895 auto_save: Whether to save immediately
897 Returns:
898 Self for chaining
899 """
900 if envs:
901 self.set_all_envs(envs)
902 if labels:
903 self.set_all_labels(labels)
904 if users:
905 # Convert tuple format to dict format if needed
906 # Supports both {'service': (uid, gid)} and {'service': {'uid': uid, 'gid': gid}}
907 converted_users = {}
908 for service, user_data in users.items():
909 if isinstance(user_data, tuple):
910 converted_users[service] = {"uid": str(user_data[0]), "gid": str(user_data[1])}
911 else:
912 converted_users[service] = user_data
913 self.set_all_users(converted_users)
915 self.set_container_names(prefix)
916 self.set_root_volumes_names(prefix)
917 self.set_root_networks_name(network_name, prefix)
918 self.set_version(version)
920 if auto_save:
921 self.write_to_file()
923 return self
925 def configure_service(
926 self,
927 service: str,
928 env: dict | None = None,
929 labels: dict | None = None,
930 user: tuple[str, str] | None = None, # (uid, gid)
931 command: str | None = None,
932 volumes: list[DockerVolumeMount] | None = None,
933 auto_save: bool = True,
934 ) -> "ComposeFile":
935 """
936 Configure a single service in one call.
938 Args:
939 service: Service name
940 env: Environment variables
941 labels: Service labels
942 user: Tuple of (uid, gid)
943 command: Service command
944 volumes: Volume mounts
945 auto_save: Whether to save immediately
947 Returns:
948 Self for chaining
949 """
950 if env:
951 self.set_envs(service, env)
952 if labels:
953 self.set_labels(service, labels)
954 if user:
955 uid, gid = user
956 self.set_user(service, uid, gid)
957 if command:
958 self.set_service_command(service, command)
959 if volumes:
960 self.set_service_volumes(service, volumes)
962 if auto_save:
963 self.write_to_file()
965 return self
967 def migrate_images(
968 self,
969 tag_updates: dict[str, str],
970 new_version: str | None = None,
971 auto_save: bool = True,
972 ) -> "ComposeFile":
973 """
974 Update multiple image tags and optionally version (common in migrations).
976 Args:
977 tag_updates: Dict of {service: new_tag}
978 new_version: New compose file version
979 auto_save: Whether to save immediately
981 Returns:
982 Self for chaining
983 """
984 images = self.get_all_images()
986 for service, new_tag in tag_updates.items():
987 if service in images:
988 images[service]["tag"] = new_tag
990 self.set_all_images(images)
992 if new_version:
993 self.set_version(new_version)
995 if auto_save:
996 self.write_to_file()
998 return self
1000 # ==================== NEW: Context Manager Support ====================
1002 def __enter__(self) -> "ComposeFile":
1003 """Enter context: snapshot current state"""
1004 self._snapshot = copy.deepcopy(self.yml)
1005 return self
1007 def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
1008 """Exit context: save or rollback"""
1009 if exc_type is None:
1010 # Success: save changes
1011 self.write_to_file()
1012 else:
1013 # Error: rollback changes
1014 if self._snapshot:
1015 self.yml = self._snapshot
1016 self._snapshot = None
1017 output = get_global_output_handler()
1018 output.warning(f"ComposeFile changes rolled back due to error: {exc_val}")
1019 return False # Don't suppress exceptions
1021 # ==================== NEW: Update Helper Methods ====================
1023 def update_env(self, service: str, key: str, value: str, auto_save: bool | None = None) -> "ComposeFile":
1024 """
1025 Update a single environment variable without get-set dance.
1027 Args:
1028 service: Service name
1029 key: Environment variable key
1030 value: Environment variable value
1031 auto_save: Override instance auto_save setting
1033 Returns:
1034 Self for chaining
1035 """
1036 if service not in self.yml["services"]:
1037 raise ComposeServiceNotFound(service)
1039 if "environment" not in self.yml["services"][service]:
1040 self.yml["services"][service]["environment"] = OrderedDict()
1042 self.yml["services"][service]["environment"][key] = value
1044 should_save = auto_save if auto_save is not None else self._auto_save
1045 if should_save:
1046 self.write_to_file()
1048 return self
1050 def delete_env(self, service: str, key: str, auto_save: bool = None) -> "ComposeFile":
1051 """
1052 Delete a single environment variable.
1054 Args:
1055 service: Service name
1056 key: Environment variable key to delete
1057 auto_save: Override instance auto_save setting
1059 Returns:
1060 Self for chaining
1061 """
1062 if service not in self.yml["services"]:
1063 raise ComposeServiceNotFound(service)
1065 if "environment" in self.yml["services"][service]:
1066 self.yml["services"][service]["environment"].pop(key, None)
1068 should_save = auto_save if auto_save is not None else self._auto_save
1069 if should_save:
1070 self.write_to_file()
1072 return self
1074 def update_label(self, service: str, key: str, value: str, auto_save: bool = None) -> "ComposeFile":
1075 """
1076 Update a single label.
1078 Args:
1079 service: Service name
1080 key: Label key
1081 value: Label value
1082 auto_save: Override instance auto_save setting
1084 Returns:
1085 Self for chaining
1086 """
1087 if service not in self.yml["services"]:
1088 raise ComposeServiceNotFound(service)
1090 if "labels" not in self.yml["services"][service]:
1091 self.yml["services"][service]["labels"] = OrderedDict()
1093 self.yml["services"][service]["labels"][key] = value
1095 should_save = auto_save if auto_save is not None else self._auto_save
1096 if should_save:
1097 self.write_to_file()
1099 return self
1101 def update_image_tag(self, service: str, new_tag: str, auto_save: bool = None) -> "ComposeFile":
1102 """
1103 Update a single service image tag.
1105 Args:
1106 service: Service name
1107 new_tag: New tag to set
1108 auto_save: Override instance auto_save setting
1110 Returns:
1111 Self for chaining
1112 """
1113 if service not in self.yml["services"]:
1114 raise ComposeServiceNotFound(service)
1116 if "image" not in self.yml["services"][service]:
1117 raise KeyError(f"Service {service} has no image defined")
1119 current_image = self.yml["services"][service]["image"]
1120 image_name = current_image.split(":")[0] if ":" in current_image else current_image
1121 self.yml["services"][service]["image"] = f"{image_name}:{new_tag}"
1123 should_save = auto_save if auto_save is not None else self._auto_save
1124 if should_save:
1125 self.write_to_file()
1127 return self
1129 # ==================== NEW: Static Template Access ====================
1131 @classmethod
1132 def load_template_yml(cls, template_name: str = "docker-compose.tmpl", template_dir: str = "templates") -> dict:
1133 """
1134 Load template YAML without instantiating ComposeFile.
1135 Renders Jinja2 template variables for dynamic Docker image tags.
1137 Args:
1138 template_name: Template file name
1139 template_dir: Template directory
1141 Returns:
1142 Parsed YAML dictionary
1143 """
1144 template_path: Path = get_template_path(template_name, template_dir)
1146 # Render Jinja2 template with dynamic image tags
1147 template = Template(template_path.read_text())
1148 image_tag = get_docker_image_tag()
1149 rendered_template = template.render(frappe_image_tag=image_tag, nginx_image_tag=image_tag)
1151 # Parse rendered YAML
1152 yml = yaml.load(rendered_template)
1153 return yml
1155 @classmethod
1156 def get_template_images(
1157 cls,
1158 template_name: str = "docker-compose.tmpl",
1159 template_dir: str = "templates",
1160 ) -> dict[str, dict[str, str]]:
1161 """
1162 Get all images from a template.
1164 Args:
1165 template_name: Template file name
1166 template_dir: Template directory
1168 Returns:
1169 Dict of {service: {name, tag, image}}
1170 """
1171 yml = cls.load_template_yml(template_name, template_dir)
1173 images = {}
1174 for service in yml.get("services", {}).keys():
1175 if "image" in yml["services"][service]:
1176 image = yml["services"][service]["image"]
1177 name, tag = image.split(":") if ":" in image else (image, "latest")
1178 images[service] = {"name": name, "tag": tag, "image": image}
1180 return images
1182 def is_service_profile_disabled(self, service: str) -> bool:
1183 try:
1184 service_definition = self.yml["services"][service]
1185 if not hasattr(service_definition, "get"):
1186 return False
1187 profiles = service_definition.get("profiles", [])
1188 if isinstance(profiles, str):
1189 return profiles == "disabled"
1190 return "disabled" in profiles
1191 except (KeyError, TypeError, AttributeError):
1192 return False
1194 @classmethod
1195 def get_template_services(
1196 cls,
1197 template_name: str = "docker-compose.tmpl",
1198 template_dir: str = "templates",
1199 ) -> list[str]:
1200 """
1201 Get list of services from template.
1203 Args:
1204 template_name: Template file name
1205 template_dir: Template directory
1207 Returns:
1208 List of service names
1209 """
1210 yml = cls.load_template_yml(template_name, template_dir)
1211 return list(yml.get("services", {}).keys())