jmm/kiwi-scp
2
0
Fork 0
Code Projects Wiki Activity

KiwiCommand API

Browse source
This commit is contained in:
Jörn-Michael Miehe 2021-11-06 03:45:27 +01:00
parent 52f0e9b8a3
commit 2e37291a68
4 changed files with 71 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
kiwi_scp/commands/cli.py
View file

@ -5,7 +5,7 @@ from typing import List, Tuple, Iterable, Any, Type
import click
from ..instance import Instance
from ..instance import Instance, Project, Services
class KiwiCLI(click.MultiCommand):
@ -42,8 +42,8 @@ class KiwiCommand:
click.secho(header, fg="green", bold=True)
@staticmethod
def print_error(header: str):
click.secho(header, file=sys.stderr, fg="red", bold=True)
def print_error(error: str):
click.secho(error, file=sys.stderr, fg="red", bold=True)
@staticmethod
def print_list(content: Iterable[str]):
@ -77,24 +77,22 @@ class KiwiCommand:
@classmethod
def run_for_instance(cls, instance: Instance, **kwargs) -> None:
for project in instance.config.projects:
cls.run_for_project(instance, project.name, **kwargs)
for project_config in instance.config.projects:
project = instance.get_project(project_config.name)
cls.run_for_existing_project(instance, project, **kwargs)
@classmethod
def run_for_project(cls, instance: Instance, project_name: str, **kwargs) -> None:
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)
def run_for_new_project(cls, instance: Instance, project_name: str, **kwargs) -> None:
raise Exception
@classmethod
def run_for_services(cls, instance: Instance, project_name: str, service_names: List[str], **kwargs) -> None:
pass
def run_for_existing_project(cls, instance: Instance, project: Project, **kwargs) -> None:
service_names = [service.name for service in project.services.content]
cls.run_for_services(instance, project, service_names, **kwargs)
@classmethod
def run_for_services(cls, instance: Instance, project: Project, service_names: List[str], **kwargs) -> None:
raise Exception
class KiwiCommandType(Enum):

38
kiwi_scp/commands/cmd_list.py
View file

@ -4,7 +4,7 @@ import click
from .cli import KiwiCommandType, KiwiCommand
from .decorators import kiwi_command
from ..instance import Instance
from ..instance import Instance, Project
@click.option(
@ -34,37 +34,29 @@ class CMD(KiwiCommand):
)
@classmethod
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()
def run_for_existing_project(cls, instance: Instance, project: Project, show: bool = None, **kwargs) -> None:
if show:
KiwiCommand.print_header(f"Showing config for all services in project '{project_name}'.")
click.echo_via_pager(str(services))
KiwiCommand.print_header(f"Showing config for all services in project '{project.name}'.")
click.echo_via_pager(str(project.services))
else:
KiwiCommand.print_header(f"Services in project '{project_name}':")
KiwiCommand.print_list(service.name for service in services.content)
KiwiCommand.print_header(f"Services in project '{project.name}':")
KiwiCommand.print_list(service.name for service in project.services.content)
@classmethod
def run_for_services(cls, instance: Instance, project_name: str, service_names: List[str], show: bool = None,
def run_for_new_project(cls, instance: Instance, project_name: str, **kwargs) -> None:
KiwiCommand.print_error(f"Project '{project_name}' not in kiwi-scp instance at '{instance.directory}'!")
@classmethod
def run_for_services(cls, instance: Instance, project: Project, 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)
services = project.services.filter_existing(service_names)
if show:
service_names = [service.name for service in services.content]
KiwiCommand.print_header(
f"Showing config for services '{', '.join(service_names)}' in project '{project_name}'.")
f"Showing config for matching 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_header(f"Matching services in project '{project.name}':")
KiwiCommand.print_list(service.name for service in services.content)

32
kiwi_scp/commands/decorators.py
View file

@ -12,13 +12,14 @@ _pass_instance = click.make_pass_decorator(
)
_project_arg = click.argument(
"project",
"project_name",
metavar="[PROJECT]",
required=False,
type=str,
)
_services_arg = click.argument(
"services",
"service_names",
metavar="[SERVICE]...",
nargs=-1,
type=str,
@ -40,24 +41,35 @@ def kiwi_command(
**decorator_kwargs,
)
@_pass_instance
def cmd(ctx: Instance, project: Optional[str] = None, services: Optional[Tuple[str]] = None,
def cmd(ctx: Instance, project_name: Optional[str] = None, service_names: Optional[Tuple[str]] = None,
**kwargs) -> None:
_logger.debug(f"{ctx.directory!r}: {project!r}, {services!r}")
if project is None:
_logger.debug(f"{ctx.directory!r}: {project_name!r}, {service_names!r}")
if project_name is None:
# run for whole instance
_logger.debug(f"running for instance, kwargs={kwargs}")
command_cls.run_for_instance(ctx, **kwargs)
elif not services:
elif not service_names:
# run for one entire project
_logger.debug(f"running for project {project}, kwargs={kwargs}")
command_cls.run_for_project(ctx, project, **kwargs)
project = ctx.get_project(project_name)
if project is not None:
_logger.debug(f"running for existing project {project}, kwargs={kwargs}")
command_cls.run_for_existing_project(ctx, project, **kwargs)
else:
_logger.debug(f"running for new project {project_name}, kwargs={kwargs}")
command_cls.run_for_new_project(ctx, project_name, **kwargs)
else:
# run for some services
_logger.debug(f"running for services {services} in project {project}, kwargs={kwargs}")
command_cls.run_for_services(ctx, project, list(services), **kwargs)
project = ctx.get_project(project_name)
if project is not None:
_logger.debug(f"running for services {service_names} in project {project}, kwargs={kwargs}")
command_cls.run_for_services(ctx, project, list(service_names), **kwargs)
else:
KiwiCommand.print_error(f"Project '{project_name}' not in kiwi-scp instance at '{ctx.directory}'!")
if command_type is KiwiCommandType.PROJECT:
cmd = _project_arg(cmd)

29
kiwi_scp/instance.py
View file

@ -43,6 +43,16 @@ class Services:
}
}).strip()
def __bool__(self) -> bool:
return bool(self.content)
def filter_existing(self, service_names: List[str]):
return Services([
service
for service in self.content
if service.name in service_names
])
@attr.s
class Project:
@ -54,20 +64,17 @@ class Project:
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:
@property
def name(self) -> str:
return self.directory.name
@property
def services(self) -> Services:
yml = Project._parse_compose_file(self.directory)
services = [
return 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
])