"kiwi list" done

kiwi_scp/commands/cli.py

@@ -1,6 +1,7 @@
 import os
+import sys
 from enum import Enum, auto
-from typing import List
+from typing import List, Tuple, Iterable
 
 import click
 
@@ -30,6 +31,28 @@ class KiwiCLI(click.MultiCommand):
 
 
 class KiwiCommand:
+    @staticmethod
+    def print_multi_color(*content: Tuple[str, str]):
+        for message, color in content:
+            click.secho(message, fg=color, nl=False)
+        click.echo()
+
+    @staticmethod
+    def print_header(header: str):
+        click.secho(header, fg="green", bold=True)
+
+    @staticmethod
+    def print_error(header: str):
+        click.secho(header, file=sys.stderr, fg="red", bold=True)
+
+    @staticmethod
+    def print_list(content: Iterable[str]):
+        for item in content:
+            KiwiCommand.print_multi_color(
+                ("  - ", "green"),
+                (item, "blue"),
+            )
+
     @classmethod
     def run_for_instance(cls, instance: Instance, **kwargs) -> None:
         for project in instance.config.projects:
@@ -37,11 +60,18 @@ class KiwiCommand:
 
     @classmethod
     def run_for_project(cls, instance: Instance, project_name: str, **kwargs) -> None:
-        service_names = [service.name for service in instance.get_services(project_name, None).content]
+        project = instance.get_project(project_name)
+
+        if project is None:
+            click.secho(f"No project '{project_name}' in kiwi-scp instance at '{instance.directory}'.", fg="red", bold=True)
+            return
+
+        service_names = [service.name for service in project.get_services().content]
+
         cls.run_for_services(instance, project_name, service_names, **kwargs)
 
     @classmethod
-    def run_for_services(cls, instance: Instance, project_name: str, services: List[str], **kwargs) -> None:
+    def run_for_services(cls, instance: Instance, project_name: str, service_names: List[str], **kwargs) -> None:
         pass
 
 

kiwi_scp/commands/cmd_list.py

@@ -21,23 +21,50 @@ class CMD(KiwiCommand):
     """List projects in this instance, services inside a project or service(s) inside a project"""
 
     @classmethod
-    def run_for_instance(cls, instance: Instance, show: bool = None, **kwargs):
+    def run_for_instance(cls, instance: Instance, show: bool = None, **kwargs) -> None:
         if show:
-            click.secho(f"Showing config for kiwi-scp instance at '{instance.directory}'.", fg="green", bold=True)
+            KiwiCommand.print_header(f"Showing config for kiwi-scp instance at '{instance.directory}'.")
             click.echo_via_pager(instance.config.kiwi_yml)
 
         else:
-            click.secho(f"Projects in kiwi-scp instance at '{instance.directory}':", fg="green", bold=True)
-
-            for project in instance.config.projects:
-                click.echo(
-                    click.style("  - ", fg="green") +
-                    click.style(project.name, fg="blue") +
-                    click.style(' (disabled)' if not project.enabled else '', fg="red")
-                )
+            KiwiCommand.print_header(f"Projects in kiwi-scp instance at '{instance.directory}':")
+            KiwiCommand.print_list(
+                project.name + click.style(" (disabled)" if not project.enabled else "", fg="red")
+                for project in instance.config.projects
+            )
 
     @classmethod
-    def run_for_services(cls, instance: Instance, project_name: str, services: List[str], show: bool = None,
-                         **kwargs):
-        print(show)
-        print(services)
+    def run_for_project(cls, instance: Instance, project_name: str, show: bool = None, **kwargs) -> None:
+        project = instance.get_project(project_name)
+
+        if project is None:
+            KiwiCommand.print_error(f"No project '{project_name}' in kiwi-scp instance at '{instance.directory}'.")
+            return
+
+        services = project.get_services()
+        if show:
+            KiwiCommand.print_header(f"Showing config for all services in project '{project_name}'.")
+            click.echo_via_pager(str(services))
+
+        else:
+            KiwiCommand.print_header(f"Services in project '{project_name}':")
+            KiwiCommand.print_list(service.name for service in services.content)
+
+    @classmethod
+    def run_for_services(cls, instance: Instance, project_name: str, service_names: List[str], show: bool = None,
+                         **kwargs) -> None:
+        project = instance.get_project(project_name)
+
+        if project is None:
+            KiwiCommand.print_error(f"No project '{project_name}' in kiwi-scp instance at '{instance.directory}'.")
+            return
+
+        services = project.get_services(service_names)
+        if show:
+            KiwiCommand.print_header(
+                f"Showing config for services '{', '.join(service_names)}' in project '{project_name}'.")
+            click.echo_via_pager(str(services))
+
+        else:
+            KiwiCommand.print_header(f"Matching services in project '{project_name}':")
+            KiwiCommand.print_list(service.name for service in services.content)

kiwi_scp/instance.py

@@ -16,21 +16,14 @@ _RE_CONFDIR = re.compile(r"^\s*\$(?:CONFDIR|{CONFDIR})/+(.*)$", flags=re.UNICODE
 @attr.s
 class Service:
     name: str = attr.ib()
-    description: CommentedMap = attr.ib()
-
-    def __str__(self) -> str:
-        return YAML().dump({
-            "service": {
-                self.name: self.description
-            }
-        }).strip()
+    content: CommentedMap = attr.ib()
 
     @property
     def configs(self) -> Generator[Path, None, None]:
-        if "volumes" not in self.description:
+        if "volumes" not in self.content:
             return
 
-        for volume in self.description["volumes"]:
+        for volume in self.content["volumes"]:
             host_part = volume.split(":")[0]
             cd_match = _RE_CONFDIR.match(host_part)
 
@@ -40,18 +33,44 @@ class Service:
 
 @attr.s
 class Services:
-    project_name: str = attr.ib()
     content: List[Service] = attr.ib()
 
     def __str__(self) -> str:
         return YAML().dump({
             "services": {
-                service.name: service.description
+                service.name: service.content
                 for service in self.content
             }
         }).strip()
 
 
+@attr.s
+class Project:
+    directory: Path = attr.ib()
+
+    @staticmethod
+    @functools.lru_cache(maxsize=10)
+    def _parse_compose_file(directory: Path) -> CommentedMap:
+        with open(directory.joinpath(COMPOSE_FILE_NAME), "r") as cf:
+            return YAML().load(cf)
+
+    def get_services(self, service_names: Optional[List[str]] = None) -> Services:
+        yml = Project._parse_compose_file(self.directory)
+        services = [
+            Service(name, description)
+            for name, description in yml["services"].items()
+        ]
+
+        if not service_names:
+            return Services(services)
+        else:
+            return Services([
+                service
+                for service in services
+                if service.name in service_names
+            ])
+
+
 @attr.s
 class Instance:
     directory: Path = attr.ib(default=Path('.'))
@@ -62,24 +81,7 @@ class Instance:
 
         return KiwiConfig.from_directory(self.directory)
 
-    @staticmethod
-    @functools.lru_cache(maxsize=10)
-    def _parse_compose_file(directory: Path):
-        with open(directory.joinpath(COMPOSE_FILE_NAME), "r") as cf:
-            return YAML().load(cf)
-
-    def get_services(self, project_name: str, service_names: Optional[Tuple[str]] = None) -> Services:
-        yml = Instance._parse_compose_file(self.directory.joinpath(project_name))
-        services = [
-            Service(name, description)
-            for name, description in yml["services"].items()
-        ]
-
-        if not service_names:
-            return Services(project_name, services)
-        else:
-            return Services(project_name, [
-                service
-                for service in services
-                if service.name in service_names
-            ])
+    def get_project(self, project_name: str) -> Optional[Project]:
+        for project in self.config.projects:
+            if project.name == project_name:
+                return Project(self.directory.joinpath(project.name))