mirror of
https://github.com/yavook/kiwi-scp.git
synced 2024-11-21 20:33:00 +00:00
moved decorators to decorators.py
This commit is contained in:
parent
57a9a50c90
commit
5d69f34de1
5 changed files with 93 additions and 107 deletions
|
@ -1,4 +1,6 @@
|
||||||
import os
|
import os
|
||||||
|
from enum import Enum, auto
|
||||||
|
from typing import List
|
||||||
|
|
||||||
import click
|
import click
|
||||||
|
|
||||||
|
@ -24,7 +26,28 @@ class KiwiCLI(click.MultiCommand):
|
||||||
mod = __import__(f"kiwi_scp.commands.cmd_{name}", None, None, ["cmd"])
|
mod = __import__(f"kiwi_scp.commands.cmd_{name}", None, None, ["cmd"])
|
||||||
except ImportError:
|
except ImportError:
|
||||||
return
|
return
|
||||||
return mod.cmd
|
return mod.CMD
|
||||||
|
|
||||||
|
|
||||||
|
class KiwiCommand:
|
||||||
|
@classmethod
|
||||||
|
def run_for_instance(cls, instance: Instance, **kwargs):
|
||||||
|
for project in instance.config.projects:
|
||||||
|
cls.run_for_project(instance, project.name, **kwargs)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def run_for_project(cls, instance: Instance, project_name: str, **kwargs):
|
||||||
|
service_names = [service.name for service in instance.get_services(project_name, None).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):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class KiwiCommandType(Enum):
|
||||||
|
INSTANCE = auto()
|
||||||
|
PROJECT = auto()
|
||||||
|
SERVICE = auto()
|
||||||
|
|
||||||
|
|
||||||
pass_instance = click.make_pass_decorator(Instance, ensure=True)
|
|
||||||
|
|
|
@ -5,7 +5,7 @@ from pathlib import Path
|
||||||
|
|
||||||
import click
|
import click
|
||||||
|
|
||||||
from .cli import pass_instance
|
from .decorators import _pass_instance as pass_instance
|
||||||
from .._constants import KIWI_CONF_NAME
|
from .._constants import KIWI_CONF_NAME
|
||||||
from ..config import KiwiConfig
|
from ..config import KiwiConfig
|
||||||
from ..instance import Instance
|
from ..instance import Instance
|
||||||
|
@ -37,7 +37,7 @@ from ..misc import user_query
|
||||||
help=f"show effective {KIWI_CONF_NAME} contents instead",
|
help=f"show effective {KIWI_CONF_NAME} contents instead",
|
||||||
)
|
)
|
||||||
@pass_instance
|
@pass_instance
|
||||||
def cmd(ctx: Instance, output: Path, force: bool, show: bool):
|
def CMD(ctx: Instance, output: Path, force: bool, show: bool):
|
||||||
"""Initialize or reconfigure a kiwi-scp instance"""
|
"""Initialize or reconfigure a kiwi-scp instance"""
|
||||||
|
|
||||||
if output is not None:
|
if output is not None:
|
||||||
|
|
|
@ -1,63 +1,9 @@
|
||||||
import typing as t
|
|
||||||
from typing import Tuple
|
|
||||||
|
|
||||||
import click
|
import click
|
||||||
|
|
||||||
from .cli import pass_instance
|
from .cli import KiwiCommandType, KiwiCommand
|
||||||
|
from .decorators import kiwi_command
|
||||||
from ..config import ProjectConfig
|
from ..config import ProjectConfig
|
||||||
from ..instance import Instance, Services
|
from ..instance import Instance, Services
|
||||||
from ..misc import service_command
|
|
||||||
|
|
||||||
|
|
||||||
class KiwiCommand:
|
|
||||||
@classmethod
|
|
||||||
def run_for_instance(cls, instance: Instance, **kwargs):
|
|
||||||
for project in instance.config.projects:
|
|
||||||
cls.run_for_project(instance, project, **kwargs)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def run_for_project(cls, instance: Instance, project: ProjectConfig, **kwargs):
|
|
||||||
cls.run_for_services(instance, project, instance.get_services(project.name, None), **kwargs)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def run_for_services(cls, instance: Instance, project: ProjectConfig, services: Services, **kwargs):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
def kiwi_command(
|
|
||||||
name: str,
|
|
||||||
**kwargs,
|
|
||||||
) -> t.Callable:
|
|
||||||
def decorator(command_cls: t.Type[KiwiCommand]) -> t.Callable:
|
|
||||||
|
|
||||||
@click.command(name, **kwargs)
|
|
||||||
@pass_instance
|
|
||||||
@service_command
|
|
||||||
def cmd(ctx: Instance, project: t.Optional[str], services: Tuple[str], **cmd_kwargs) -> None:
|
|
||||||
print(f"{ctx.directory!r}: {project!r}, {services!r}")
|
|
||||||
if project is None:
|
|
||||||
# run for whole instance
|
|
||||||
print(f"for instance: {cmd_kwargs}")
|
|
||||||
command_cls.run_for_instance(ctx, **cmd_kwargs)
|
|
||||||
|
|
||||||
elif not services:
|
|
||||||
# run for one entire project
|
|
||||||
print(f"for project {project}: {cmd_kwargs}")
|
|
||||||
for project_cfg in ctx.config.projects:
|
|
||||||
if project_cfg.name == project:
|
|
||||||
command_cls.run_for_project(ctx, project_cfg, **cmd_kwargs)
|
|
||||||
|
|
||||||
else:
|
|
||||||
# run for some services
|
|
||||||
print(f"for services {project}.{services}: {cmd_kwargs}")
|
|
||||||
for project_cfg in ctx.config.projects:
|
|
||||||
if project_cfg.name == project:
|
|
||||||
services = ctx.get_services(project_cfg.name, services)
|
|
||||||
command_cls.run_for_services(ctx, project_cfg, services, **cmd_kwargs)
|
|
||||||
|
|
||||||
return cmd
|
|
||||||
|
|
||||||
return decorator
|
|
||||||
|
|
||||||
|
|
||||||
@click.option(
|
@click.option(
|
||||||
|
@ -67,9 +13,10 @@ def kiwi_command(
|
||||||
)
|
)
|
||||||
@kiwi_command(
|
@kiwi_command(
|
||||||
"list",
|
"list",
|
||||||
|
KiwiCommandType.PROJECT,
|
||||||
short_help="Inspect a kiwi-scp instance",
|
short_help="Inspect a kiwi-scp instance",
|
||||||
)
|
)
|
||||||
class cmd(KiwiCommand):
|
class CMD(KiwiCommand):
|
||||||
@classmethod
|
@classmethod
|
||||||
def run_for_instance(cls, instance: Instance, show: bool = None, **kwargs):
|
def run_for_instance(cls, instance: Instance, show: bool = None, **kwargs):
|
||||||
print(show)
|
print(show)
|
||||||
|
@ -80,17 +27,3 @@ class cmd(KiwiCommand):
|
||||||
**kwargs):
|
**kwargs):
|
||||||
print(show)
|
print(show)
|
||||||
print(services)
|
print(services)
|
||||||
|
|
||||||
# @click.command(
|
|
||||||
# "list",
|
|
||||||
# short_help="Inspect a kiwi-scp instance",
|
|
||||||
# )
|
|
||||||
# @pass_instance
|
|
||||||
# @service_command
|
|
||||||
# def cmd(ctx: Instance, project: str, services: Tuple[str]):
|
|
||||||
# """List projects in this instance, services inside a project or service(s) inside a project"""
|
|
||||||
# print(f"project: {project!r}, services: {services!r}")
|
|
||||||
# if project is not None:
|
|
||||||
# print(ctx.get_services(project, services))
|
|
||||||
# else:
|
|
||||||
# print(f"projects: {ctx.config.projects}")
|
|
||||||
|
|
61
kiwi_scp/commands/decorators.py
Normal file
61
kiwi_scp/commands/decorators.py
Normal file
|
@ -0,0 +1,61 @@
|
||||||
|
from typing import Callable, Type, Optional, Tuple
|
||||||
|
|
||||||
|
import click
|
||||||
|
|
||||||
|
from .cli import KiwiCommandType, KiwiCommand
|
||||||
|
from ..instance import Instance
|
||||||
|
|
||||||
|
_pass_instance = click.make_pass_decorator(
|
||||||
|
Instance,
|
||||||
|
ensure=True,
|
||||||
|
)
|
||||||
|
_project_arg = click.argument(
|
||||||
|
"project",
|
||||||
|
required=False,
|
||||||
|
type=str,
|
||||||
|
)
|
||||||
|
_services_arg = click.argument(
|
||||||
|
"services",
|
||||||
|
metavar="[SERVICE]...",
|
||||||
|
nargs=-1,
|
||||||
|
type=str,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def kiwi_command(
|
||||||
|
name: str,
|
||||||
|
command_type: KiwiCommandType,
|
||||||
|
**kwargs,
|
||||||
|
) -> Callable:
|
||||||
|
def decorator(command_cls: Type[KiwiCommand]) -> Callable:
|
||||||
|
|
||||||
|
@click.command(name, **kwargs)
|
||||||
|
@_pass_instance
|
||||||
|
def cmd(ctx: Instance, project: Optional[str] = None, services: Optional[Tuple[str]] = None,
|
||||||
|
**cmd_kwargs) -> None:
|
||||||
|
print(f"{ctx.directory!r}: {project!r}, {services!r}")
|
||||||
|
if project is None:
|
||||||
|
# run for whole instance
|
||||||
|
print(f"for instance: {cmd_kwargs}")
|
||||||
|
command_cls.run_for_instance(ctx, **cmd_kwargs)
|
||||||
|
|
||||||
|
elif not services:
|
||||||
|
# run for one entire project
|
||||||
|
print(f"for project {project}: {cmd_kwargs}")
|
||||||
|
command_cls.run_for_project(ctx, project, **cmd_kwargs)
|
||||||
|
|
||||||
|
else:
|
||||||
|
# run for some services
|
||||||
|
print(f"for services {project}.{services}: {cmd_kwargs}")
|
||||||
|
command_cls.run_for_services(ctx, project, list(services), **cmd_kwargs)
|
||||||
|
|
||||||
|
if command_type is KiwiCommandType.PROJECT:
|
||||||
|
cmd = _project_arg(cmd)
|
||||||
|
|
||||||
|
elif command_type is KiwiCommandType.SERVICE:
|
||||||
|
cmd = _project_arg(cmd)
|
||||||
|
cmd = _services_arg(cmd)
|
||||||
|
|
||||||
|
return cmd
|
||||||
|
|
||||||
|
return decorator
|
|
@ -1,44 +1,13 @@
|
||||||
import re
|
import re
|
||||||
from typing import Any, Type, List, Callable, Optional
|
from typing import Any, Type, Optional
|
||||||
|
|
||||||
import attr
|
|
||||||
import click
|
import click
|
||||||
import ruamel.yaml
|
import ruamel.yaml
|
||||||
import ruamel.yaml.compat
|
import ruamel.yaml.compat
|
||||||
from click.decorators import FC
|
|
||||||
|
|
||||||
from ._constants import HEADER_KIWI_CONF_NAME
|
from ._constants import HEADER_KIWI_CONF_NAME
|
||||||
|
|
||||||
|
|
||||||
@attr.s
|
|
||||||
class _MultiDecorator:
|
|
||||||
options: List[Callable[[FC], FC]] = attr.ib(factory=list)
|
|
||||||
|
|
||||||
def __call__(self, target: FC):
|
|
||||||
for option in reversed(self.options):
|
|
||||||
target = option(target)
|
|
||||||
|
|
||||||
return target
|
|
||||||
|
|
||||||
|
|
||||||
_project_arg = click.argument(
|
|
||||||
"project",
|
|
||||||
required=False,
|
|
||||||
type=str,
|
|
||||||
)
|
|
||||||
|
|
||||||
_services_arg = click.argument(
|
|
||||||
"services",
|
|
||||||
metavar="[SERVICE]...",
|
|
||||||
nargs=-1,
|
|
||||||
type=str,
|
|
||||||
)
|
|
||||||
|
|
||||||
instance_command = _MultiDecorator([])
|
|
||||||
project_command = _MultiDecorator([_project_arg])
|
|
||||||
service_command = _MultiDecorator([_project_arg, _services_arg])
|
|
||||||
|
|
||||||
|
|
||||||
def user_query(description: str, default: Any, cast_to: Type[Any] = str):
|
def user_query(description: str, default: Any, cast_to: Type[Any] = str):
|
||||||
# prompt user as per argument
|
# prompt user as per argument
|
||||||
while True:
|
while True:
|
||||||
|
|
Loading…
Reference in a new issue