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

Logic for PROJECTS command type; kiwi_command args -> KiwiCommand vars

Browse source
This commit is contained in:
Jörn-Michael Miehe 2021-12-01 17:47:03 +01:00
parent ed659ba50f
commit 3cdc68885c
2 changed files with 68 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
kiwi_scp/commands/cli.py
View file

@ -3,7 +3,7 @@ import logging
import os
import sys
from enum import Enum, auto
from typing import List, Tuple, Iterable, Type, Optional, TypeVar
from typing import List, Iterable, Type, Optional, TypeVar
import click
@ -40,15 +40,19 @@ class KiwiCLI(click.MultiCommand):
return member
class KiwiCommandType(Enum):
INSTANCE = auto()
PROJECT = auto()
PROJECTS = auto()
SERVICES = auto()
T = TypeVar("T")
class KiwiCommand:
@staticmethod
def print_multi_color(*content: Tuple[str, str]) -> None:
for message, color in content:
click.secho(message, fg=color, nl=False)
click.echo()
type: KiwiCommandType = KiwiCommandType.SERVICES
enabled_only: bool = False
@staticmethod
def print_header(header: str) -> None:
@ -61,10 +65,7 @@ class KiwiCommand:
@staticmethod
def print_list(content: Iterable[str]) -> None:
for item in content:
KiwiCommand.print_multi_color(
(" - ", "green"),
(item, "blue"),
)
click.echo(click.style(" - ", fg="green") + click.style(item, fg="blue"))
@staticmethod
def user_query(description: str, default: T, cast_to: Type[T] = str) -> T:
@ -127,39 +128,58 @@ class KiwiCommand:
return answer == "yes"
@classmethod
def run(cls, instance: Instance, project_name: Optional[str] = None, service_names: Optional[List[str]] = None,
**kwargs):
def run(cls, instance: Instance, project_names: List[str], service_names: List[str], **kwargs):
_logger.debug(f"{instance.directory!r}: {project_name!r}, {service_names!r}")
if project_name is None:
_logger.debug(f"{instance.directory!r}: {project_names!r}, {service_names!r}")
projects = [
instance.get_project(project_name)
for project_name in project_names
]
if not projects:
# run for whole instance
_logger.debug(f"running for instance, kwargs={kwargs}")
cls.run_for_instance(instance, **kwargs)
elif not service_names:
# run for one entire project
project = instance.get_project(project_name)
# run for entire project(s)
for project_name, project in zip(project_names, projects):
if project is None:
_logger.debug(f"running for new project {project_name}, kwargs={kwargs}")
cls.run_for_new_project(instance, project_name, **kwargs)
else:
if cls.enabled_only and not project.project_config.enabled:
cls.print_error(f"Can't interact with disabled project {project_name}!")
return
_logger.debug(f"running for project {project.name}, kwargs={kwargs}")
cls.run_for_project(instance, project, **kwargs)
else:
# run for some services
project = instance.get_project(project_name)
if project is not None:
_logger.debug(f"running for services {service_names} in project {project}, kwargs={kwargs}")
cls.run_for_services(instance, project, service_names, **kwargs)
project_name = project_names[0]
project = projects[0]
if project is None:
cls.print_error(f"Project '{project_name}' not in kiwi-scp instance at '{instance.directory}'!")
else:
cls.print_error(f"Project '{project_name}' not in kiwi-scp instance at '{instance.directory}'!")
if cls.enabled_only and not project.project_config.enabled:
cls.print_error(f"Can't interact with disabled project {project_name}!")
return
_logger.debug(f"running for services {service_names} in project {project_name}, kwargs={kwargs}")
cls.run_for_services(instance, project, service_names, **kwargs)
@classmethod
def run_for_instance(cls, instance: Instance, **kwargs) -> None:
for project_config in instance.config.projects:
if cls.enabled_only and not project_config.enabled:
cls.print_header(f"Skipping disabled project {project_config.name}")
continue
project = instance.get_project(project_config.name)
cls.run_for_project(instance, project, **kwargs)
@ -189,10 +209,3 @@ class KiwiCommand:
def run_for_filtered_services(cls, instance: Instance, project: Project, services: Services,
new_service_names: List[str], **kwargs) -> None:
raise Exception
class KiwiCommandType(Enum):
INSTANCE = auto()
PROJECT = auto()
PROJECTS = auto()
SERVICE = auto()

29
kiwi_scp/commands/decorators.py
View file

@ -13,7 +13,6 @@ _pass_instance = click.make_pass_decorator(
_project_arg = click.argument(
"project_name",
metavar="PROJECT",
required=False, # TODO remove this line when PROJECTS logic is implemented
type=str,
)
@ -40,7 +39,6 @@ _services_arg_s = click.argument(
def kiwi_command(
cmd_type: KiwiCommandType = KiwiCommandType.SERVICE,
**decorator_kwargs,
) -> Callable:
def decorator(command_cls: Type[KiwiCommand]) -> Callable:
@ -50,20 +48,33 @@ def kiwi_command(
**decorator_kwargs,
)
@_pass_instance
def cmd(ctx: Instance, project_name: Optional[str] = None, service_names: Optional[Tuple[str]] = None,
**kwargs) -> None:
if service_names is not None:
def cmd(ctx: Instance, project_name: Optional[str] = None, project_names: Optional[Tuple[str]] = None,
service_names: Optional[Tuple[str]] = None, **kwargs) -> None:
if command_cls.type is KiwiCommandType.INSTANCE:
project_names = []
elif command_cls.type is KiwiCommandType.PROJECTS:
project_names = list(project_names)
else:
if project_name is None:
project_names = []
else:
project_names = [project_name]
if command_cls.type is KiwiCommandType.SERVICES:
service_names = list(service_names)
command_cls.run(ctx, project_name, service_names, **kwargs)
command_cls.run(ctx, project_names, service_names, **kwargs)
if cmd_type is KiwiCommandType.PROJECT:
if command_cls.type is KiwiCommandType.PROJECT:
cmd = _project_arg(cmd)
elif cmd_type is KiwiCommandType.PROJECTS:
elif command_cls.type is KiwiCommandType.PROJECTS:
cmd = _projects_arg(cmd)
elif cmd_type is KiwiCommandType.SERVICE:
elif command_cls.type is KiwiCommandType.SERVICES:
cmd = _services_arg_p(cmd)
cmd = _services_arg_s(cmd)