mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8fbf651223 | |||
| 459c16ec87 | |||
| d9c74310c3 | |||
| 5e6cdc342e | |||
| a378163431 | |||
| fd4f2e1570 |
+53
-46
@@ -1,18 +1,18 @@
|
||||
from typing import Callable
|
||||
from inspect import getfullargspec
|
||||
import re
|
||||
|
||||
from ..command.entity import Command
|
||||
from ..router.entity import Router
|
||||
from ..command.input_comand.entity import InputCommand
|
||||
from ..command.input_comand.exceptions import (IncorrectInputFlagException,
|
||||
InvalidInputFlagsHandlerHasBeenAlreadyCreatedException,
|
||||
IncorrectNumberArgsHandlerException,
|
||||
UnknownCommandHandlerHasBeenAlreadyCreatedException)
|
||||
from ..command.exceptions import (UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
from .exceptions import (InvalidRouterInstanceException,
|
||||
InvalidDescriptionMessagePatternException,
|
||||
NoRegisteredRoutersException,
|
||||
NoRegisteredHandlersException,
|
||||
RepeatedCommandInDifferentRoutersException)
|
||||
RepeatedCommandInDifferentRoutersException,
|
||||
IncorrectNumberOfHandlerArgsException)
|
||||
|
||||
|
||||
class App:
|
||||
@@ -45,11 +45,12 @@ class App:
|
||||
self.repeat_command_groups = repeat_command_groups
|
||||
|
||||
self._routers: list[Router] = []
|
||||
self._invalid_input_flags_handler: Callable[[str], None] | None = None
|
||||
self._unknown_command_handler: Callable[[Command], None] | None = None
|
||||
self._registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | Command]] | Router]] = []
|
||||
self._app_main_router: Router | None = None
|
||||
self._description_message_pattern: str = '[{command}] *=*=* {description}'
|
||||
self._registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | Command]] | Router]] = []
|
||||
self._invalid_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Incorrect flag syntax: "{raw_command}"')
|
||||
self._repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Repeated input flags: "{raw_command}"')
|
||||
self._empty_input_command_handler: Callable[[], None] = lambda: print_func(f'Empty input command')
|
||||
self._unknown_command_handler: Callable[[Command], None] = lambda command: print_func(f"Unknown command: {command.get_string_entity()}")
|
||||
|
||||
|
||||
def start_polling(self) -> None:
|
||||
@@ -71,24 +72,42 @@ class App:
|
||||
raw_command: str = input()
|
||||
|
||||
try:
|
||||
input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
|
||||
except IncorrectInputFlagException:
|
||||
input_command: Command = Command.parse_input_command(raw_command=raw_command)
|
||||
except UnprocessedInputFlagException:
|
||||
self.print_func(self.line_separate)
|
||||
if self._invalid_input_flags_handler:
|
||||
self._invalid_input_flags_handler(raw_command)
|
||||
else:
|
||||
self.print_func(f'Incorrect flag syntax: "{raw_command}"')
|
||||
self.print_func(self.line_separate)
|
||||
|
||||
if not self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
continue
|
||||
|
||||
self._checking_command_for_exit_command(input_command.get_string_entity())
|
||||
except RepeatedInputFlagsException:
|
||||
self.print_func(self.line_separate)
|
||||
self._repeated_input_flags_handler(raw_command)
|
||||
self.print_func(self.line_separate)
|
||||
|
||||
if not self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
continue
|
||||
|
||||
except EmptyInputCommandException:
|
||||
self.print_func(self.line_separate)
|
||||
self._empty_input_command_handler()
|
||||
self.print_func(self.line_separate)
|
||||
|
||||
if not self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
continue
|
||||
|
||||
self._check_command_for_exit_command(input_command.get_string_entity())
|
||||
|
||||
self.print_func(self.line_separate)
|
||||
is_unknown_command: bool = self._check_is_command_unknown(input_command)
|
||||
|
||||
if is_unknown_command:
|
||||
self.print_func(self.line_separate)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
if not self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
continue
|
||||
@@ -111,44 +130,39 @@ class App:
|
||||
|
||||
|
||||
def set_description_message_pattern(self, pattern: str) -> None:
|
||||
try:
|
||||
pattern.format(command='command',
|
||||
description='description')
|
||||
except KeyError:
|
||||
raise InvalidDescriptionMessagePatternException(pattern)
|
||||
first_check = re.match(r'.*{command}.*', pattern)
|
||||
second_check = re.match(r'.*{description}.*', pattern)
|
||||
|
||||
if bool(first_check) and bool(second_check):
|
||||
self._description_message_pattern: str = pattern
|
||||
else:
|
||||
raise InvalidDescriptionMessagePatternException(pattern)
|
||||
|
||||
|
||||
def set_invalid_input_flags_handler(self, handler: Callable[[str], None]) -> None:
|
||||
if self._invalid_input_flags_handler:
|
||||
raise InvalidInputFlagsHandlerHasBeenAlreadyCreatedException()
|
||||
else:
|
||||
args = getfullargspec(handler).args
|
||||
if len(args) != 1:
|
||||
raise IncorrectNumberArgsHandlerException()
|
||||
raise IncorrectNumberOfHandlerArgsException()
|
||||
else:
|
||||
self._invalid_input_flags_handler = handler
|
||||
|
||||
|
||||
def set_unknown_command_handler(self, handler: Callable[[str], None]) -> None:
|
||||
if self._unknown_command_handler:
|
||||
raise UnknownCommandHandlerHasBeenAlreadyCreatedException()
|
||||
else:
|
||||
def set_repeated_input_flags_handler(self, handler: Callable[[str], None]) -> None:
|
||||
args = getfullargspec(handler).args
|
||||
if len(args) != 1:
|
||||
raise IncorrectNumberArgsHandlerException()
|
||||
raise IncorrectNumberOfHandlerArgsException()
|
||||
else:
|
||||
self._repeated_input_flags_handler = handler
|
||||
|
||||
|
||||
def set_unknown_command_handler(self, handler: Callable[[str], None]) -> None:
|
||||
args = getfullargspec(handler).args
|
||||
if len(args) != 1:
|
||||
raise IncorrectNumberOfHandlerArgsException()
|
||||
else:
|
||||
self._unknown_command_handler = handler
|
||||
|
||||
|
||||
def get_all_app_commands(self) -> list[str]:
|
||||
all_commands: list[str] = []
|
||||
for router in self._routers:
|
||||
all_commands.extend(router.get_all_commands())
|
||||
|
||||
return all_commands
|
||||
|
||||
|
||||
def include_router(self, router: Router) -> None:
|
||||
if not isinstance(router, Router):
|
||||
raise InvalidRouterInstanceException()
|
||||
@@ -190,7 +204,7 @@ class App:
|
||||
raise RepeatedCommandInDifferentRoutersException()
|
||||
|
||||
|
||||
def _checking_command_for_exit_command(self, command: str):
|
||||
def _check_command_for_exit_command(self, command: str):
|
||||
if command.lower() == self.exit_command.lower():
|
||||
if self.ignore_exit_command_register:
|
||||
self.print_func(self.farewell_message)
|
||||
@@ -211,14 +225,7 @@ class App:
|
||||
else:
|
||||
if command_entity['command'].get_string_entity() == command.get_string_entity():
|
||||
return False
|
||||
|
||||
if self._unknown_command_handler:
|
||||
self._unknown_command_handler(command)
|
||||
else:
|
||||
print(f"Unknown command: {command.get_string_entity()}")
|
||||
|
||||
self.print_func(self.line_separate)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
return True
|
||||
|
||||
|
||||
|
||||
@@ -28,3 +28,8 @@ class NoRegisteredHandlersException(Exception):
|
||||
class RepeatedCommandInDifferentRoutersException(Exception):
|
||||
def __str__(self):
|
||||
return "Commands in different handlers cannot be repeated"
|
||||
|
||||
|
||||
class IncorrectNumberOfHandlerArgsException(Exception):
|
||||
def __str__(self):
|
||||
return "Incorrect Input Flags Handler has incorrect number of arguments"
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
from .entity import Command
|
||||
+79
-18
@@ -2,46 +2,51 @@ from .params.flag.entity import Flag
|
||||
from .params.flag.flags_group.entity import FlagsGroup
|
||||
from .exceptions import (InvalidCommandInstanceException,
|
||||
InvalidDescriptionInstanceException,
|
||||
InvalidFlagsInstanceException)
|
||||
from .params.flag.input_flag.entity import InputFlag
|
||||
InvalidFlagsInstanceException,
|
||||
UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
|
||||
class Command:
|
||||
T = TypeVar('T')
|
||||
|
||||
|
||||
class Command(Generic[T]):
|
||||
def __init__(self, command: str,
|
||||
description: str | None = None,
|
||||
description: str = None,
|
||||
flags: Flag | FlagsGroup | None = None):
|
||||
self._command = command
|
||||
self._description = description
|
||||
self._flags: FlagsGroup | None = flags if isinstance(flags, FlagsGroup) else FlagsGroup([flags]) if isinstance(flags, Flag) else flags
|
||||
self._description = f'description for "{self._command}" command' if not description else description
|
||||
self._registered_flags: FlagsGroup | None = flags if isinstance(flags, FlagsGroup) else FlagsGroup([flags]) if isinstance(flags, Flag) else flags
|
||||
|
||||
self._input_flags: FlagsGroup | None = None
|
||||
|
||||
self._input_flags: InputFlag | FlagsGroup | None = None
|
||||
|
||||
def get_string_entity(self):
|
||||
return self._command
|
||||
|
||||
|
||||
def get_description(self):
|
||||
if not self._description:
|
||||
description = f'description for "{self._command}" command'
|
||||
return description
|
||||
else:
|
||||
return self._description
|
||||
|
||||
def get_flags(self):
|
||||
return self._flags
|
||||
|
||||
def set_command(self, command: str):
|
||||
self._command = command
|
||||
def get_registered_flags(self):
|
||||
return self._registered_flags
|
||||
|
||||
|
||||
def validate_commands_params(self):
|
||||
if not isinstance(self._command, str):
|
||||
raise InvalidCommandInstanceException(self._command)
|
||||
if not isinstance(self._description, str):
|
||||
raise InvalidDescriptionInstanceException()
|
||||
if not any([(isinstance(self._flags, Flag), isinstance(self._flags, FlagsGroup)), not self._flags]):
|
||||
if not any([(isinstance(self._registered_flags, FlagsGroup)), not self._registered_flags]):
|
||||
raise InvalidFlagsInstanceException
|
||||
|
||||
def validate_input_flag(self, flag: InputFlag):
|
||||
registered_flags: FlagsGroup | Flag | None = self._flags
|
||||
|
||||
def validate_input_flag(self, flag: Flag):
|
||||
registered_flags: FlagsGroup | None = self.get_registered_flags()
|
||||
if registered_flags:
|
||||
if isinstance(registered_flags, Flag):
|
||||
if registered_flags.get_string_entity() == flag.get_string_entity():
|
||||
@@ -57,4 +62,60 @@ class Command:
|
||||
return False
|
||||
|
||||
|
||||
def set_input_flags(self, input_flags: FlagsGroup):
|
||||
self._input_flags = input_flags
|
||||
|
||||
def get_input_flags(self) -> FlagsGroup:
|
||||
return self._input_flags
|
||||
|
||||
@staticmethod
|
||||
def parse_input_command(raw_command: str) -> 'Command[T]':
|
||||
if not raw_command:
|
||||
raise EmptyInputCommandException()
|
||||
list_of_tokens = raw_command.split()
|
||||
command = list_of_tokens[0]
|
||||
list_of_tokens.pop(0)
|
||||
|
||||
flags: FlagsGroup = FlagsGroup()
|
||||
current_flag_name = None
|
||||
current_flag_value = None
|
||||
for _ in list_of_tokens:
|
||||
if _.startswith('-'):
|
||||
flag_prefix_last_symbol_index = _.rfind('-')
|
||||
if current_flag_name or len(_) < 2 or len(_[:flag_prefix_last_symbol_index]) > 3:
|
||||
raise UnprocessedInputFlagException()
|
||||
else:
|
||||
current_flag_name = _
|
||||
else:
|
||||
if not current_flag_name:
|
||||
raise UnprocessedInputFlagException()
|
||||
else:
|
||||
current_flag_value = _
|
||||
if current_flag_name and current_flag_value:
|
||||
flag_prefix_last_symbol_index = current_flag_name.rfind('-')
|
||||
flag_prefix = current_flag_name[:flag_prefix_last_symbol_index]
|
||||
flag_name = current_flag_name[flag_prefix_last_symbol_index:]
|
||||
|
||||
input_flag = Flag(flag_name=flag_name,
|
||||
flag_prefix=flag_prefix)
|
||||
input_flag.set_value(current_flag_value)
|
||||
|
||||
all_flags = [x.get_string_entity() for x in flags.get_flags()]
|
||||
if input_flag.get_string_entity() not in all_flags:
|
||||
flags.add_flag(input_flag)
|
||||
else:
|
||||
raise RepeatedInputFlagsException(input_flag)
|
||||
|
||||
current_flag_name = None
|
||||
current_flag_value = None
|
||||
if any([current_flag_name, current_flag_value]):
|
||||
raise UnprocessedInputFlagException()
|
||||
if len(flags.get_flags()) == 0:
|
||||
return Command(command=command)
|
||||
else:
|
||||
input_command = Command(command=command)
|
||||
input_command.set_input_flags(flags)
|
||||
return input_command
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
from .params.flag.entity import Flag
|
||||
|
||||
|
||||
class InvalidCommandInstanceException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Command Instance"
|
||||
@@ -11,3 +14,21 @@ class InvalidDescriptionInstanceException(Exception):
|
||||
class InvalidFlagsInstanceException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Flags Instance"
|
||||
|
||||
|
||||
class UnprocessedInputFlagException(Exception):
|
||||
def __str__(self):
|
||||
return "Unprocessed Input Flags"
|
||||
|
||||
|
||||
class RepeatedInputFlagsException(Exception):
|
||||
def __init__(self, flag: Flag):
|
||||
self.flag = flag
|
||||
def __str__(self):
|
||||
return ("Repeated Input Flags\n"
|
||||
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'")
|
||||
|
||||
|
||||
class EmptyInputCommandException(Exception):
|
||||
def __str__(self):
|
||||
return "Input Command is empty"
|
||||
@@ -1,66 +0,0 @@
|
||||
from ..input_comand.exceptions import IncorrectInputFlagException, RepeatedInputFlagsException
|
||||
from ..entity import Command
|
||||
from ..params.flag.flags_group.entity import FlagsGroup
|
||||
from ..params.flag.input_flag.entity import InputFlag
|
||||
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
|
||||
T = TypeVar('T')
|
||||
|
||||
|
||||
class InputCommand(Command, Generic[T]):
|
||||
def set_input_flags(self, input_flags: FlagsGroup):
|
||||
self._input_flags = input_flags
|
||||
|
||||
def get_input_flags(self) -> FlagsGroup:
|
||||
return self._input_flags
|
||||
|
||||
@staticmethod
|
||||
def parse(raw_command: str) -> 'InputCommand[T]':
|
||||
list_of_tokens = raw_command.split()
|
||||
command = list_of_tokens[0]
|
||||
list_of_tokens.pop(0)
|
||||
|
||||
flags: FlagsGroup = FlagsGroup()
|
||||
current_flag_name = None
|
||||
current_flag_value = None
|
||||
for _ in list_of_tokens:
|
||||
if _.startswith('-'):
|
||||
flag_prefix_last_symbol_index = _.rfind('-')
|
||||
if current_flag_name or len(_) < 2 or len(_[:flag_prefix_last_symbol_index]) > 3:
|
||||
raise IncorrectInputFlagException()
|
||||
else:
|
||||
current_flag_name = _
|
||||
else:
|
||||
if not current_flag_name:
|
||||
raise IncorrectInputFlagException()
|
||||
else:
|
||||
current_flag_value = _
|
||||
if current_flag_name and current_flag_value:
|
||||
flag_prefix_last_symbol_index = current_flag_name.rfind('-')
|
||||
flag_prefix = current_flag_name[:flag_prefix_last_symbol_index]
|
||||
flag_name = current_flag_name[flag_prefix_last_symbol_index:]
|
||||
|
||||
input_flag = InputFlag(flag_name=flag_name,
|
||||
flag_prefix=flag_prefix)
|
||||
input_flag.set_value(current_flag_value)
|
||||
|
||||
all_flags = [x.get_string_entity() for x in flags.get_flags()]
|
||||
if input_flag.get_string_entity() not in all_flags:
|
||||
flags.add_flag(input_flag)
|
||||
else:
|
||||
raise RepeatedInputFlagsException(input_flag)
|
||||
|
||||
current_flag_name = None
|
||||
current_flag_value = None
|
||||
if any([current_flag_name, current_flag_value]):
|
||||
raise IncorrectInputFlagException()
|
||||
if len(flags.get_flags()) == 0:
|
||||
return InputCommand(command=command)
|
||||
else:
|
||||
input_command = InputCommand(command=command)
|
||||
input_command.set_input_flags(flags)
|
||||
return input_command
|
||||
|
||||
|
||||
@@ -1,39 +0,0 @@
|
||||
from ..params.flag.input_flag.entity import InputFlag
|
||||
|
||||
|
||||
class InvalidInputFlagException(Exception):
|
||||
def __init__(self, flag: InputFlag):
|
||||
self.flag = flag
|
||||
def __str__(self):
|
||||
return ("Invalid Input Flags\n"
|
||||
f"Unknown or invalid input flag: '{self.flag.get_string_entity()} {self.flag.get_value()}'")
|
||||
|
||||
class IncorrectInputFlagException(Exception):
|
||||
def __str__(self):
|
||||
return "Incorrect Input Flags"
|
||||
|
||||
|
||||
class RepeatedInputFlagsException(Exception):
|
||||
def __init__(self, flag: InputFlag):
|
||||
self.flag = flag
|
||||
def __str__(self):
|
||||
return ("Repeated Input Flags\n"
|
||||
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'")
|
||||
|
||||
|
||||
class InvalidInputFlagsHandlerHasBeenAlreadyCreatedException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Input Flags Handler has already been created"
|
||||
|
||||
|
||||
class UnknownCommandHandlerHasBeenAlreadyCreatedException(Exception):
|
||||
def __str__(self):
|
||||
return "Unknown Command Handler has already been created"
|
||||
|
||||
|
||||
class IncorrectNumberArgsHandlerException(Exception):
|
||||
def __str__(self):
|
||||
return "Incorrect Input Flags Handler has incorrect number of arguments"
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
from .entity import Flag
|
||||
from .flags_group.entity import FlagsGroup
|
||||
@@ -1,11 +1,11 @@
|
||||
from typing import Literal
|
||||
from typing import Literal, Pattern
|
||||
|
||||
|
||||
class Flag:
|
||||
def __init__(self, flag_name: str,
|
||||
flag_prefix: Literal['-', '--', '---'] = '-',
|
||||
ignore_flag_value_register: bool = False,
|
||||
possible_flag_values: list[str] = False):
|
||||
possible_flag_values: list[str] | Pattern[str] = False):
|
||||
self._flag_name = flag_name
|
||||
self._flag_prefix = flag_prefix
|
||||
self.possible_flag_values = possible_flag_values
|
||||
@@ -30,7 +30,14 @@ class Flag:
|
||||
self._value = value
|
||||
|
||||
def validate_input_flag_value(self, input_flag_value: str):
|
||||
if self.possible_flag_values:
|
||||
if isinstance(self.possible_flag_values, Pattern):
|
||||
is_valid = bool(self.possible_flag_values.match(input_flag_value))
|
||||
if bool(is_valid):
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
if isinstance(self.possible_flag_values, list):
|
||||
if self.ignore_flag_value_register:
|
||||
if input_flag_value.lower() in [x.lower() for x in self.possible_flag_values]:
|
||||
return True
|
||||
@@ -41,5 +48,4 @@ class Flag:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
@@ -1,18 +1,17 @@
|
||||
from argenta.command.params.flag.entity import Flag
|
||||
from argenta.command.params.flag.input_flag.entity import InputFlag
|
||||
|
||||
|
||||
class FlagsGroup:
|
||||
def __init__(self, flags: list[Flag | InputFlag] = None):
|
||||
self._flags: list[Flag | InputFlag] = [] if not flags else flags
|
||||
def __init__(self, flags: list[Flag] = None):
|
||||
self._flags: list[Flag] = [] if not flags else flags
|
||||
|
||||
def get_flags(self):
|
||||
return self._flags
|
||||
|
||||
def add_flag(self, flag: Flag | InputFlag):
|
||||
def add_flag(self, flag: Flag):
|
||||
self._flags.append(flag)
|
||||
|
||||
def add_flags(self, flags: list[Flag | InputFlag]):
|
||||
def add_flags(self, flags: list[Flag]):
|
||||
self._flags.extend(flags)
|
||||
|
||||
def __iter__(self):
|
||||
@@ -20,3 +19,6 @@ class FlagsGroup:
|
||||
|
||||
def __next__(self):
|
||||
return next(iter(self))
|
||||
|
||||
def __getitem__(self, item):
|
||||
return self._flags[item]
|
||||
|
||||
@@ -1,11 +0,0 @@
|
||||
from ...flag.entity import Flag
|
||||
|
||||
|
||||
class InputFlag(Flag):
|
||||
def set_value(self, value: str):
|
||||
self._value = value
|
||||
|
||||
def get_value(self) -> str:
|
||||
return self._value
|
||||
|
||||
|
||||
+25
-36
@@ -1,13 +1,13 @@
|
||||
from typing import Callable, Any
|
||||
from inspect import getfullargspec
|
||||
|
||||
from ..command.entity import Command
|
||||
from ..command.input_comand.entity import InputCommand
|
||||
from ..command.input_comand.exceptions import InvalidInputFlagException
|
||||
from ..command.params.flag.entity import Flag
|
||||
from ..command.params.flag.flags_group.entity import FlagsGroup
|
||||
from ..router.exceptions import (RepeatedCommandException, RepeatedFlagNameException,
|
||||
CurrentCommandDoesNotProcessFlagsException,
|
||||
TooManyTransferredArgsException,
|
||||
RequiredArgumentNotPassedException)
|
||||
RequiredArgumentNotPassedException,
|
||||
IncorrectNumberOfHandlerArgsException)
|
||||
|
||||
|
||||
class Router:
|
||||
@@ -21,8 +21,7 @@ class Router:
|
||||
self._command_entities: list[dict[str, Callable[[], None] | Command]] = []
|
||||
self._ignore_command_register: bool = False
|
||||
|
||||
self._unknown_command_handler: Callable[[str], None] | None = None
|
||||
self._not_valid_flag_handler: Callable[[str], None] | None = None
|
||||
self._not_valid_flag_handler: Callable[[Flag], None] = lambda flag: print(f"Undefined or incorrect input flag: '{flag.get_string_entity()} {flag.get_value()}'")
|
||||
|
||||
|
||||
def command(self, command: Command) -> Callable[[Any], Any]:
|
||||
@@ -39,23 +38,33 @@ class Router:
|
||||
|
||||
return command_decorator
|
||||
|
||||
def set_invalid_input_flag_handler(self, func):
|
||||
processed_args = getfullargspec(func).args
|
||||
if len(processed_args) != 1:
|
||||
raise IncorrectNumberOfHandlerArgsException()
|
||||
else:
|
||||
self._not_valid_flag_handler = func
|
||||
|
||||
def input_command_handler(self, input_command: InputCommand):
|
||||
|
||||
def input_command_handler(self, input_command: Command):
|
||||
input_command_name: str = input_command.get_string_entity()
|
||||
input_command_flags: FlagsGroup = input_command.get_input_flags()
|
||||
for command_entity in self._command_entities:
|
||||
if input_command_name.lower() == command_entity['command'].get_string_entity().lower():
|
||||
if command_entity['command'].get_flags():
|
||||
if input_command.get_input_flags():
|
||||
for flag in input_command.get_input_flags():
|
||||
if command_entity['command'].get_registered_flags():
|
||||
if input_command_flags:
|
||||
for flag in input_command_flags:
|
||||
is_valid = command_entity['command'].validate_input_flag(flag)
|
||||
if not is_valid:
|
||||
raise InvalidInputFlagException(flag)
|
||||
return command_entity['handler_func'](input_command.get_input_flags())
|
||||
self._not_valid_flag_handler(flag)
|
||||
return
|
||||
return command_entity['handler_func'](input_command_flags)
|
||||
else:
|
||||
return command_entity['handler_func'](FlagsGroup(None))
|
||||
else:
|
||||
if input_command.get_input_flags():
|
||||
raise CurrentCommandDoesNotProcessFlagsException()
|
||||
if input_command_flags:
|
||||
self._not_valid_flag_handler(input_command_flags[0])
|
||||
return
|
||||
else:
|
||||
return command_entity['handler_func']()
|
||||
|
||||
@@ -68,7 +77,7 @@ class Router:
|
||||
if command_name.lower() in [x.lower() for x in self.get_all_commands()]:
|
||||
raise RepeatedCommandException()
|
||||
|
||||
flags: FlagsGroup = command.get_flags()
|
||||
flags: FlagsGroup = command.get_registered_flags()
|
||||
if flags:
|
||||
flags_name: list = [x.get_string_entity().lower() for x in flags]
|
||||
if len(set(flags_name)) < len(flags_name):
|
||||
@@ -77,7 +86,7 @@ class Router:
|
||||
|
||||
@staticmethod
|
||||
def _validate_func_args(command: Command, func: Callable):
|
||||
registered_args = command.get_flags()
|
||||
registered_args = command.get_registered_flags()
|
||||
transferred_args = getfullargspec(func).args
|
||||
if registered_args and transferred_args:
|
||||
if len(transferred_args) != 1:
|
||||
@@ -104,29 +113,9 @@ class Router:
|
||||
return self.title
|
||||
|
||||
|
||||
def get_router_info(self) -> dict:
|
||||
return {
|
||||
'title': self.title,
|
||||
'name': self.name,
|
||||
'ignore_command_register': self._ignore_command_register,
|
||||
'attributes': {
|
||||
'command_entities': self._command_entities,
|
||||
'unknown_command_func': self._unknown_command_handler
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
def get_all_commands(self) -> list[str]:
|
||||
all_commands: list[str] = []
|
||||
for command_entity in self._command_entities:
|
||||
all_commands.append(command_entity['command'].get_string_entity())
|
||||
|
||||
return all_commands
|
||||
|
||||
def get_all_flags(self) -> list[FlagsGroup]:
|
||||
all_flags: list[FlagsGroup] = []
|
||||
for command_entity in self._command_entities:
|
||||
all_flags.append(command_entity['command'].get_flags())
|
||||
|
||||
return all_flags
|
||||
|
||||
@@ -13,11 +13,6 @@ class RepeatedFlagNameException(Exception):
|
||||
return "Repeated flag name in register command"
|
||||
|
||||
|
||||
class CurrentCommandDoesNotProcessFlagsException(Exception):
|
||||
def __str__(self):
|
||||
return "Current command does not process flags"
|
||||
|
||||
|
||||
class TooManyTransferredArgsException(Exception):
|
||||
def __str__(self):
|
||||
return "Too many transferred arguments"
|
||||
@@ -26,3 +21,13 @@ class TooManyTransferredArgsException(Exception):
|
||||
class RequiredArgumentNotPassedException(Exception):
|
||||
def __str__(self):
|
||||
return "Required argument not passed"
|
||||
|
||||
|
||||
class NotValidInputFlagHandlerHasBeenAlreadyCreatedException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Input Flag Handler has already been created"
|
||||
|
||||
|
||||
class IncorrectNumberOfHandlerArgsException(Exception):
|
||||
def __str__(self):
|
||||
return "Incorrect Input Flags Handler has incorrect number of arguments"
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
import re
|
||||
|
||||
|
||||
def set_description_message_pattern(pattern: str) -> None:
|
||||
first_check = re.match(r'.*command.*', pattern)
|
||||
second_check = re.match(r'.*{description}.*', pattern)
|
||||
|
||||
|
||||
set_description_message_pattern('Invalid des{ommand}cription pattern')
|
||||
@@ -1,3 +1,5 @@
|
||||
import re
|
||||
|
||||
from rich.console import Console
|
||||
|
||||
from argenta.command.entity import Command
|
||||
@@ -5,16 +7,16 @@ from argenta.command.params.flag.entity import Flag
|
||||
from argenta.command.params.flag.flags_group.entity import FlagsGroup
|
||||
from argenta.router import Router
|
||||
|
||||
from ..handlers.handlers_implementation.help_command import help_command
|
||||
|
||||
work_router: Router = Router(title='Work points:')
|
||||
work_router: Router = Router(title='Work nts:')
|
||||
settings_router: Router = Router(title='Settings points:')
|
||||
|
||||
console = Console()
|
||||
|
||||
flagi = FlagsGroup(flags=[
|
||||
Flag(flag_name='host',
|
||||
flag_prefix='--', ),
|
||||
flag_prefix='--',
|
||||
possible_flag_values=re.compile(r'^192.168.\d{1,3}.\d{1,3}$')),
|
||||
Flag(flag_name='port',
|
||||
flag_prefix='--', )
|
||||
])
|
||||
@@ -42,3 +44,13 @@ def command_start_solving(argrrtrts: FlagsGroup | None):
|
||||
def command_update():
|
||||
print('uefi')
|
||||
# upgrade_command()
|
||||
|
||||
|
||||
def invalid_input_flag(flag: Flag):
|
||||
print(f'Invalid inpuuuuuuuuuuuuuuuuuuuuuuuut flag: "{flag.get_string_entity()} {flag.get_value()}"')
|
||||
|
||||
|
||||
work_router.set_invalid_input_flag_handler(invalid_input_flag)
|
||||
|
||||
|
||||
|
||||
@@ -1,14 +1,18 @@
|
||||
from tests.mock_app.handlers.routers import work_router, settings_router
|
||||
from argenta.app.entity import App
|
||||
from art import text2art
|
||||
from rich.console import Console
|
||||
|
||||
from argenta.app import App
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
from argenta.command.params.flag import Flag, FlagsGroup
|
||||
|
||||
|
||||
app: App = App(prompt='[italic white bold]What do you want to do(enter number of action)?',
|
||||
line_separate='[bold green]\n---------------------------------------------\n',
|
||||
line_separate=f'\n{"[bold green]-[/bold green][bold red]-[/bold red]"*25}\n',
|
||||
print_func=Console().print,
|
||||
command_group_description_separate='',
|
||||
repeat_command_groups=False)
|
||||
repeat_command_groups=True)
|
||||
|
||||
|
||||
def main():
|
||||
@@ -25,6 +29,8 @@ def main():
|
||||
app.set_farewell_message(goodbye_message)
|
||||
|
||||
app.set_invalid_input_flags_handler(lambda raw_command: print(f"Invalid input flags: {raw_command}"))
|
||||
app.set_unknown_command_handler(lambda command: print(f"Unknown command: {command.get_string_entity()}"))
|
||||
app.set_repeated_input_flags_handler(lambda raw_command: print(f"Repeated input flags: {raw_command}"))
|
||||
|
||||
app.set_description_message_pattern('[bold red][{command}][/bold red] [blue]*=*=*[/blue] [bold yellow italic]{description}')
|
||||
|
||||
+2
-4
@@ -1,9 +1,9 @@
|
||||
[project]
|
||||
name = "argenta"
|
||||
version = "0.3.1"
|
||||
version = "0.3.5"
|
||||
description = "python library for creating custom shells"
|
||||
authors = [
|
||||
{name = "kolo",email = "kolo.is.main@gmail.com"}
|
||||
{name = "kolo", email = "kolo.is.main@gmail.com"}
|
||||
]
|
||||
license = {text = "MIT"}
|
||||
readme = "README.md"
|
||||
@@ -33,6 +33,4 @@ numpy = "^2.2.2"
|
||||
word2number = "^1.1"
|
||||
numexpr = "^2.10.2"
|
||||
requests = "^2.32.3"
|
||||
tqdm = "^4.67.1"
|
||||
setuptools = "^75.8.0"
|
||||
|
||||
|
||||
@@ -1,29 +0,0 @@
|
||||
from pprint import pprint
|
||||
from rich.console import Console
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
work_router: Router = Router(name='work')
|
||||
settings_router: Router = Router(name='settings')
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
@work_router.command(command='a')
|
||||
def command_help():
|
||||
console.print('[bold red]command help [/bold red]')
|
||||
|
||||
|
||||
@work_router.command(command='B', description='tester')
|
||||
def command_start_solving():
|
||||
console.print('[bold red]command start [/bold red]')
|
||||
|
||||
|
||||
@settings_router.command(command='b')
|
||||
def command_settings():
|
||||
console.print('[bold red]command settings [/bold red]')
|
||||
|
||||
|
||||
@work_router.unknown_command
|
||||
def command_unknown_command(command):
|
||||
console.print(f'[bold red]Unknown command: [/bold red]{command}')
|
||||
@@ -1,17 +0,0 @@
|
||||
from pprint import pprint
|
||||
from tests.mock_default_app.handlers.routers import work_router, settings_router
|
||||
from argenta.app.entity import App
|
||||
|
||||
|
||||
app: App = App(ignore_command_register=False,
|
||||
line_separate='\n-------------------------------\n')
|
||||
|
||||
|
||||
def main():
|
||||
app.include_router(work_router, is_main=True)
|
||||
app.include_router(settings_router)
|
||||
|
||||
app.start_polling()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,9 +0,0 @@
|
||||
class Entity:
|
||||
pass
|
||||
|
||||
class Person(Entity):
|
||||
pass
|
||||
|
||||
a: Entity = Entity()
|
||||
print(a)
|
||||
a = Person()
|
||||
@@ -0,0 +1,20 @@
|
||||
from argenta.app import App
|
||||
from argenta.app.exceptions import (InvalidDescriptionMessagePatternException,
|
||||
NoRegisteredRoutersException)
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestApp(unittest.TestCase):
|
||||
def test_set_invalid_description_message_pattern(self):
|
||||
with self.assertRaises(InvalidDescriptionMessagePatternException):
|
||||
App().set_description_message_pattern('Invalid description pattern')
|
||||
|
||||
def test_set_invalid_description_message_pattern2(self):
|
||||
with self.assertRaises(InvalidDescriptionMessagePatternException):
|
||||
App().set_description_message_pattern('Invalid {desription} description {comand} pattern')
|
||||
|
||||
def test_no_registered_router(self):
|
||||
with self.assertRaises(NoRegisteredRoutersException):
|
||||
App()._validate_number_of_routers()
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
from argenta.command import Command
|
||||
from argenta.command.params.flag import Flag, FlagsGroup
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestCommand(unittest.TestCase):
|
||||
def test_parse_correct_raw_command(self):
|
||||
self.assertEqual(Command.parse_input_command('ssh --host 192.168.0.3').get_string_entity(), 'ssh')
|
||||
|
||||
def test_parse_raw_command_with_flag_name_without_value(self):
|
||||
with self.assertRaises(UnprocessedInputFlagException):
|
||||
Command.parse_input_command('ssh --host')
|
||||
|
||||
def test_parse_raw_command_without_flag_name_with_value(self):
|
||||
with self.assertRaises(UnprocessedInputFlagException):
|
||||
Command.parse_input_command('ssh 192.168.0.3')
|
||||
|
||||
def test_parse_raw_command_with_repeated_flag_name(self):
|
||||
with self.assertRaises(RepeatedInputFlagsException):
|
||||
Command.parse_input_command('ssh --host 192.168.0.3 --host 172.198.0.43')
|
||||
|
||||
def test_parse_empty_raw_command(self):
|
||||
with self.assertRaises(EmptyInputCommandException):
|
||||
Command.parse_input_command('')
|
||||
|
||||
def test_get_command_description(self):
|
||||
self.assertEqual(Command(command='test', description='test description').get_description(), 'test description')
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
from argenta.command.params.flag import Flag
|
||||
|
||||
import unittest
|
||||
import re
|
||||
|
||||
|
||||
class TestFlag(unittest.TestCase):
|
||||
def test_get_string_entity(self):
|
||||
self.assertEqual(Flag(flag_name='test').get_string_entity(),
|
||||
'-test')
|
||||
|
||||
def test_get_string_entity2(self):
|
||||
self.assertEqual(Flag(flag_name='test',
|
||||
flag_prefix='---').get_string_entity(),
|
||||
'---test')
|
||||
|
||||
def test_get_flag_name(self):
|
||||
self.assertEqual(Flag(flag_name='test').get_flag_name(),
|
||||
'test')
|
||||
|
||||
def test_get_flag_prefix(self):
|
||||
self.assertEqual(Flag(flag_name='test').get_flag_prefix(),
|
||||
'-')
|
||||
|
||||
def test_get_flag_prefix2(self):
|
||||
self.assertEqual(Flag(flag_name='test',
|
||||
flag_prefix='--').get_flag_prefix(),
|
||||
'--')
|
||||
|
||||
def test_get_flag_value_without_set(self):
|
||||
self.assertEqual(Flag(flag_name='test').get_value(),
|
||||
None)
|
||||
|
||||
def test_get_flag_value_with_set(self):
|
||||
flag = Flag(flag_name='test')
|
||||
flag.set_value('example')
|
||||
self.assertEqual(flag.get_value(), 'example')
|
||||
|
||||
def test_validate_incorrect_flag_value_with_list_of_possible_flag_values(self):
|
||||
flag = Flag(flag_name='test', possible_flag_values=['1', '2', '3'])
|
||||
self.assertEqual(flag.validate_input_flag_value('bad value'), False)
|
||||
|
||||
def test_validate_correct_flag_value_with_list_of_possible_flag_values(self):
|
||||
flag = Flag(flag_name='test', possible_flag_values=['1', '2', '3'])
|
||||
self.assertEqual(flag.validate_input_flag_value('1'), True)
|
||||
|
||||
def test_validate_incorrect_flag_value_with_pattern_of_possible_flag_values(self):
|
||||
flag = Flag(flag_name='test', possible_flag_values=re.compile(r'192.168.\d+.\d+'))
|
||||
self.assertEqual(flag.validate_input_flag_value('152.123.9.8'), False)
|
||||
|
||||
def test_validate_correct_flag_value_with_pattern_of_possible_flag_values(self):
|
||||
flag = Flag(flag_name='test', possible_flag_values=re.compile(r'192.168.\d+.\d+'))
|
||||
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
from argenta.command.params.flag import Flag, FlagsGroup
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestFlagsGroup(unittest.TestCase):
|
||||
def test_get_flags(self):
|
||||
flags = FlagsGroup()
|
||||
list_of_flags = [
|
||||
Flag('test1'),
|
||||
Flag('test2'),
|
||||
Flag('test3'),
|
||||
]
|
||||
flags.add_flags(list_of_flags)
|
||||
self.assertEqual(flags.get_flags(),
|
||||
list_of_flags)
|
||||
|
||||
def test_add_flag(self):
|
||||
flags = FlagsGroup()
|
||||
flags.add_flag(Flag('test'))
|
||||
self.assertEqual(len(flags.get_flags()), 1)
|
||||
|
||||
def test_add_flags(self):
|
||||
flags = FlagsGroup()
|
||||
flags.add_flags([Flag('test'), Flag('test2')])
|
||||
self.assertEqual(len(flags.get_flags()), 2)
|
||||
@@ -0,0 +1,34 @@
|
||||
from argenta.command.params.flag import FlagsGroup, Flag
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestRouter(unittest.TestCase):
|
||||
def test_get_router_name(self):
|
||||
self.assertEqual(Router(name='test name').get_name(), 'test name')
|
||||
|
||||
def test_get_router_title(self):
|
||||
self.assertEqual(Router(title='test title').get_title(), 'test title')
|
||||
|
||||
def test_input_correct_command(self):
|
||||
router = Router()
|
||||
@router.command(Command(command='test'))
|
||||
def test():
|
||||
return 'correct result'
|
||||
|
||||
self.assertEqual(router.input_command_handler(Command(command='test')), 'correct result')
|
||||
|
||||
def test_input_command_with_invalid_flag(self):
|
||||
router = Router()
|
||||
router.set_invalid_input_flag_handler(lambda x: x)
|
||||
|
||||
@router.command(Command(command='test'))
|
||||
def test():
|
||||
return 'correct result'
|
||||
|
||||
input_command = Command(command='test')
|
||||
input_command.set_input_flags(FlagsGroup([Flag('host')]))
|
||||
|
||||
self.assertEqual(router.input_command_handler(input_command), None)
|
||||
Reference in New Issue
Block a user