sh-edraft.de/sh_cpl
3
0
Fork 0
Code Issues 4 Pull Requests 1 Actions Packages Projects 1 Releases 12 Wiki Activity

Improved argument handling

Browse Source
This commit is contained in:
Sven Heidemann
2022-05-19 19:37:32 +02:00
parent 4fe073580a
commit 0f5b1b7586
9 changed files with 34 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/cpl_cli/__init__.py
View File

@@ -22,8 +22,6 @@ from collections import namedtuple
# imports:
from .cli import CLI
from .command_abc import CommandABC
from .command_handler_service import CommandHandler
from .command_model import CommandModel
from .error import Error
from .main import main
from .startup import Startup

2
src/cpl_cli/cli.py
View File

@@ -26,7 +26,7 @@ class CLI(ApplicationABC):
:return:
"""
try:
self._configuration.parse_console_arguments()
self._configuration.parse_console_arguments(self._services)
except KeyboardInterrupt:
Console.write_line()
sys.exit()

9
src/cpl_cli/command/update_service.py
View File

@@ -38,6 +38,7 @@ class UpdateService(CommandABC):
self._build_settings = build_settings
self._project_settings = project_settings
self._cli_settings = cli_settings
self._is_simulation = False
@property
def help_message(self) -> str:
@@ -132,6 +133,9 @@ class UpdateService(CommandABC):
:param new_package:
:return:
"""
if self._is_simulation:
return
if old_package in self._project_settings.dependencies:
index = self._project_settings.dependencies.index(old_package)
if '/' in new_package:
@@ -158,6 +162,11 @@ class UpdateService(CommandABC):
:param args:
:return:
"""
if 'simulate' in args:
args.remove('simulate')
Console.write_line('Running in simulation mode:')
self._is_simulation = True
Pip.set_executable(self._project_settings.python_executable)
self._check_project_dependencies()
self._check_outdated()

137
src/cpl_cli/command_handler_service.py
View File

@@ -1,137 +0,0 @@
import os
from abc import ABC
from typing import Optional
from cpl_core.configuration.configuration_abc import ConfigurationABC
from cpl_core.console.console import Console
from cpl_core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl_core.utils.string import String
from cpl_cli.command.custom_script_service import CustomScriptService
from cpl_cli.configuration.workspace_settings import WorkspaceSettings
from cpl_cli.error import Error
from cpl_cli.command_model import CommandModel
class CommandHandler(ABC):
def __init__(self, config: ConfigurationABC, services: ServiceProviderABC):
"""
Service to handle incoming commands and args
:param config:
:param services:
"""
ABC.__init__(self)
self._config = config
self._env = self._config.environment
self._services = services
self._commands: list[CommandModel] = []
@property
def commands(self) -> list[CommandModel]:
return self._commands
@staticmethod
def _project_not_found():
Error.error(
'The command requires to be run in an CPL project, but a project could not be found.'
)
def add_command(self, cmd: CommandModel):
self._commands.append(cmd)
def remove_command(self, cmd: CommandModel):
self._commands.remove(cmd)
def handle(self, cmd: str, args: list[str]):
"""
Handles incoming commands and args
:param cmd:
:param args:
:return:
"""
for command in self._commands:
if cmd == command.name or cmd in command.aliases:
error = None
project_name: Optional[str] = None
workspace: Optional[WorkspaceSettings] = None
if os.path.isfile(os.path.join(self._env.working_directory, 'cpl-workspace.json')):
workspace = self._config.get_configuration(WorkspaceSettings)
if command.is_project_needed:
name = os.path.basename(self._env.working_directory)
for r, d, f in os.walk(self._env.working_directory):
for file in f:
if file.endswith('.json'):
f_name = file.split('.json')[0]
if f_name == name or \
String.convert_to_camel_case(f_name).lower() == String.convert_to_camel_case(
name).lower():
project_name = f_name
break
if not command.is_workspace_needed and project_name is None and workspace is None:
self._project_not_found()
return
elif command.is_workspace_needed or project_name is None:
if workspace is None:
Error.error(
'The command requires to be run in an CPL workspace or project, '
'but a workspace or project could not be found.'
)
return
if project_name is None:
project_name = workspace.default_project
self._config.add_configuration('ProjectName', project_name)
project_json = f'{project_name}.json'
if workspace is not None:
if project_name not in workspace.projects:
Error.error(
f'Project {project_name} not found.'
)
return
project_json = workspace.projects[project_name]
if not os.path.isfile(os.path.join(self._env.working_directory, project_json)):
self._project_not_found()
return
project_json = os.path.join(self._env.working_directory, project_json)
if command.change_cwd:
self._env.set_working_directory(
os.path.join(self._env.working_directory, os.path.dirname(project_json))
)
self._config.add_json_file(project_json, optional=True, output=False)
# pre scripts
Console.write('\n')
self._handle_pre_or_post_scripts(True, workspace, command)
self._services.get_service(command.command).run(args)
# post scripts
Console.write('\n\n')
self._handle_pre_or_post_scripts(False, workspace, command)
Console.write('\n')
def _handle_pre_or_post_scripts(self, pre: bool, workspace: WorkspaceSettings, command: CommandModel):
script_type = 'pre-' if pre else 'post-'
if workspace is not None and len(workspace.scripts) > 0:
for script in workspace.scripts:
if script_type in script and script.split(script_type)[1] == command.name:
script_name = script
script_cmd = workspace.scripts[script]
if script_cmd in workspace.scripts:
script_name = workspace.scripts[script]
css: CustomScriptService = self._services.get_service(CustomScriptService)
if css is None:
continue
css.run([script_name])

37
src/cpl_cli/command_model.py
View File

@@ -1,37 +0,0 @@
from cpl_cli.command_abc import CommandABC
class CommandModel:
def __init__(self, name: str, aliases: list[str], command: CommandABC, is_workspace_needed: bool,
is_project_needed: bool, change_cwd: bool):
self._name = name
self._aliases = aliases
self._command = command
self._is_workspace_needed = is_workspace_needed
self._is_project_needed = is_project_needed
self._change_cwd = change_cwd
@property
def name(self) -> str:
return self._name
@property
def aliases(self) -> list[str]:
return self._aliases
@property
def command(self) -> CommandABC:
return self._command
@property
def is_workspace_needed(self) -> bool:
return self._is_workspace_needed
@property
def is_project_needed(self) -> bool:
return self._is_project_needed
@property
def change_cwd(self) -> bool:
return self._change_cwd

4
src/cpl_cli/startup_argument_extension.py
View File

@@ -18,7 +18,6 @@ from cpl_cli.configuration.workspace_settings import WorkspaceSettings
from cpl_core.application import StartupExtensionABC
from cpl_core.configuration.argument_type_enum import ArgumentTypeEnum
from cpl_core.configuration.configuration_abc import ConfigurationABC
from cpl_core.console import Console
from cpl_core.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl_core.environment import ApplicationEnvironmentABC
from cpl_core.utils import String
@@ -83,7 +82,8 @@ class StartupArgumentExtension(StartupExtensionABC):
config.create_console_argument(ArgumentTypeEnum.Executable, '', 'uninstall', ['ui', 'UI'], UninstallService) \
.add_console_argument(ArgumentTypeEnum.Flag, '--', 'virtual', ['v', 'V']) \
.add_console_argument(ArgumentTypeEnum.Flag, '--', 'simulate', ['s', 'S'])
config.create_console_argument(ArgumentTypeEnum.Executable, '', 'update', ['u', 'U'], UpdateService)
config.create_console_argument(ArgumentTypeEnum.Executable, '', 'update', ['u', 'U'], UpdateService) \
.add_console_argument(ArgumentTypeEnum.Flag, '--', 'simulate', ['s', 'S'])
config.create_console_argument(ArgumentTypeEnum.Executable, '', 'version', ['v', 'V'], VersionService)
config.for_each_argument(

3
src/cpl_core/application/application_builder.py
View File

@@ -10,7 +10,7 @@ from cpl_core.dependency_injection.service_collection import ServiceCollection
class ApplicationBuilder(ApplicationBuilderABC):
r"""This is class is used to build a object of :class:`cpl_core.application.application_abc.ApplicationABC`
r"""This is class is used to build an object of :class:`cpl_core.application.application_abc.ApplicationABC`
Parameter
---------
@@ -54,7 +54,6 @@ class ApplicationBuilder(ApplicationBuilderABC):
config = self._configuration
services = self._services.build_service_provider()
config.resolve_runnable_argument_types(services)
for ex in self._app_extensions:
extension = ex()

31
src/cpl_core/configuration/configuration.py
View File

@@ -15,7 +15,7 @@ from cpl_core.configuration.flag_argument import FlagArgument
from cpl_core.configuration.variable_argument import VariableArgument
from cpl_core.console.console import Console
from cpl_core.console.foreground_color_enum import ForegroundColorEnum
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl_core.environment.application_environment import ApplicationEnvironment
from cpl_core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl_core.environment.environment_name_enum import EnvironmentNameEnum
@@ -53,6 +53,10 @@ class Configuration(ConfigurationABC):
def argument_error_function(self, argument_error_function: Callable):
self._argument_error_function = argument_error_function
@property
def arguments(self) -> list[ArgumentABC]:
return self._argument_types
@staticmethod
def _print_info(name: str, message: str):
r"""Prints an info message
@@ -147,7 +151,7 @@ class Configuration(ConfigurationABC):
self._print_error(__name__, f'Cannot load config file: {file}! -> {e}')
return {}
def _parse_arguments(self, call_stack: list[Callable], arg_list: list[str], args_types: list[ArgumentABC]):
def _parse_arguments(self, executables: list[ArgumentABC], arg_list: list[str], args_types: list[ArgumentABC]):
for i in range(0, len(arg_list)):
arg_str = arg_list[i]
for n in range(0, len(args_types)):
@@ -160,9 +164,9 @@ class Configuration(ConfigurationABC):
if isinstance(arg, ExecutableArgument):
if arg_str.startswith(arg.token) \
and arg_str_without_token == arg.name or arg_str_without_token in arg.aliases:
call_stack.append(arg.run)
executables.append(arg)
self._handled_args.append(arg_str)
self._parse_arguments(call_stack, arg_list[i + 1:], arg.console_arguments)
self._parse_arguments(executables, arg_list[i + 1:], arg.console_arguments)
# variables
elif isinstance(arg, VariableArgument):
@@ -178,7 +182,7 @@ class Configuration(ConfigurationABC):
value = arg_list[i + 1]
self._set_variable(arg.name, value)
self._handled_args.append(arg_str)
self._parse_arguments(call_stack, arg_list[i + 1:], arg.console_arguments)
self._parse_arguments(executables, arg_list[i + 1:], arg.console_arguments)
# flags
elif isinstance(arg, FlagArgument):
@@ -186,7 +190,7 @@ class Configuration(ConfigurationABC):
and arg_str_without_token == arg.name or arg_str_without_token in arg.aliases:
self._additional_arguments.append(arg.name)
self._handled_args.append(arg_str)
self._parse_arguments(call_stack, arg_list[i + 1:], arg.console_arguments)
self._parse_arguments(executables, arg_list[i + 1:], arg.console_arguments)
# add left over values to args
if arg_str not in self._additional_arguments and arg_str not in self._handled_args:
@@ -266,19 +270,16 @@ class Configuration(ConfigurationABC):
if config_model == search_type:
return self._config[config_model]
def parse_console_arguments(self, error: bool = None):
def parse_console_arguments(self, services: ServiceProviderABC, error: bool = None):
# sets environment variables as possible arguments as: --VAR=VALUE
for arg_name in ConfigurationVariableNameEnum.to_list():
self.add_console_argument(VariableArgument('--', str(arg_name).upper(), [str(arg_name).lower()], '='))
arg_list = sys.argv[1:]
call_stack = []
self._parse_arguments(call_stack, arg_list, self._argument_types)
executables: list[ExecutableArgument] = []
self._parse_arguments(executables, arg_list, self._argument_types)
for call in call_stack:
call(self._additional_arguments)
for exe in executables:
service: ExecutableArgument = services.get_service(exe.executable_type)
service.run(self._additional_arguments)
def resolve_runnable_argument_types(self, services: ServiceProviderABC):
for arg in self._argument_types:
if isinstance(arg, ExecutableArgument):
arg.set_executable(services.get_service(arg.executable_type))

17
src/cpl_core/configuration/configuration_abc.py
View File

@@ -31,6 +31,10 @@ class ConfigurationABC(ABC):
@abstractmethod
def argument_error_function(self, argument_error_function: Callable): pass
@property
@abstractmethod
def arguments(self) -> list[ArgumentABC]: pass
@abstractmethod
def add_environment_variables(self, prefix: str):
r"""Reads the environment variables
@@ -137,7 +141,7 @@ class ConfigurationABC(ABC):
pass
@abstractmethod
def parse_console_arguments(self, error: bool = None):
def parse_console_arguments(self, services: 'ServiceProviderABC', error: bool = None):
r"""Reads the console arguments
Parameter
@@ -146,14 +150,3 @@ class ConfigurationABC(ABC):
Defines is invalid argument error will be shown or not
"""
pass
@abstractmethod
def resolve_runnable_argument_types(self, services: 'ServiceProviderABC'):
r"""Gets all objects for given types of ConsoleArguments
Parameter
---------
services: :class:`cpl_core.dependency_injection.service_provider_abc.ServiceProviderABC`
Provides services
"""
pass