mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 10:05:28 +03:00
ruff format
This commit is contained in:
@@ -8,8 +8,8 @@ from argenta.orchestrator.argparser import ArgParser
|
||||
from argenta.orchestrator.argparser.arguments import BooleanArgument
|
||||
|
||||
|
||||
arg_parser = ArgParser(processed_args=[BooleanArgument('repeat')])
|
||||
app: App = App(autocompleter=AutoCompleter('.hist'))
|
||||
arg_parser = ArgParser(processed_args=[BooleanArgument("repeat")])
|
||||
app: App = App(autocompleter=AutoCompleter(".hist"))
|
||||
orchestrator: Orchestrator = Orchestrator()
|
||||
|
||||
|
||||
@@ -22,5 +22,6 @@ def main():
|
||||
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
+10
-1
@@ -14,12 +14,14 @@ import socket
|
||||
# Маршрутизатор для работы с файлами
|
||||
file_router = Router("Файловые операции")
|
||||
|
||||
|
||||
@file_router.command(Command("list", "Список файлов"))
|
||||
def list_files(response: Response):
|
||||
files = os.listdir()
|
||||
for file in files:
|
||||
print(file)
|
||||
|
||||
|
||||
@file_router.command(Command("size", "Размер файла"))
|
||||
def file_size(response: Response):
|
||||
file_name = input("Введите имя файла: ")
|
||||
@@ -29,9 +31,11 @@ def file_size(response: Response):
|
||||
else:
|
||||
print(f"Файл {file_name} не найден")
|
||||
|
||||
|
||||
# Маршрутизатор для системных операций
|
||||
system_router = Router("Системные операции")
|
||||
|
||||
|
||||
@system_router.command(Command("info", "Информация о системе"))
|
||||
def system_info(response: Response):
|
||||
print(f"Система: {platform.system()}")
|
||||
@@ -39,6 +43,7 @@ def system_info(response: Response):
|
||||
print(f"Архитектура: {platform.architecture()}")
|
||||
print(f"Процессор: {platform.processor()}")
|
||||
|
||||
|
||||
@system_router.command(Command("memory", "Информация о памяти"))
|
||||
def memory_info(response: Response):
|
||||
memory = psutil.virtual_memory()
|
||||
@@ -46,27 +51,31 @@ def memory_info(response: Response):
|
||||
print(f"Доступно: {memory.available / (1024**3):.2f} ГБ")
|
||||
print(f"Использовано: {memory.used / (1024**3):.2f} ГБ ({memory.percent}%)")
|
||||
|
||||
|
||||
# Маршрутизатор для сетевых операций
|
||||
network_router = Router("Сетевые операции")
|
||||
|
||||
|
||||
@network_router.command(Command("ping", "Проверка доступности хоста"))
|
||||
def ping_host(response: Response):
|
||||
host = input("Введите имя хоста: ")
|
||||
print(f"Пингую {host}...")
|
||||
subprocess.run(["ping", "-c", "4", host])
|
||||
|
||||
|
||||
@network_router.command(Command("ip", "Показать IP-адреса"))
|
||||
def show_ip(response: Response):
|
||||
hostname = socket.gethostname()
|
||||
print(f"Имя хоста: {hostname}")
|
||||
print(f"IP-адрес: {socket.gethostbyname(hostname)}")
|
||||
|
||||
|
||||
# Создание приложения и регистрация маршрутизаторов
|
||||
app = App(
|
||||
prompt="System> ",
|
||||
initial_message="Pingator",
|
||||
dividing_line=DynamicDividingLine("*"),
|
||||
autocompleter=AutoCompleter('.hist', 'e'),
|
||||
autocompleter=AutoCompleter(".hist", "e"),
|
||||
)
|
||||
|
||||
# Добавляем все маршрутизаторы
|
||||
|
||||
@@ -5,6 +5,8 @@ console = Console()
|
||||
|
||||
|
||||
def help_command():
|
||||
console.print("[italic bold]The main functionality of the script is to convert an expression from a string "
|
||||
console.print(
|
||||
"[italic bold]The main functionality of the script is to convert an expression from a string "
|
||||
"to a mathematical one and then calculate this expression. "
|
||||
"Project GitHub: https://github.com/koloideal/WordMath[/italic bold]")
|
||||
"Project GitHub: https://github.com/koloideal/WordMath[/italic bold]"
|
||||
)
|
||||
|
||||
@@ -7,12 +7,19 @@ from argenta.response import Response
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
work_router: Router = Router(title='Work points:')
|
||||
work_router: Router = Router(title="Work points:")
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
@work_router.command(Command('get', 'Get Help', aliases=['help', 'Get_help'], flags=Flags(PredefinedFlags.PORT, PredefinedFlags.HOST)))
|
||||
@work_router.command(
|
||||
Command(
|
||||
"get",
|
||||
"Get Help",
|
||||
aliases=["help", "Get_help"],
|
||||
flags=Flags(PredefinedFlags.PORT, PredefinedFlags.HOST),
|
||||
)
|
||||
)
|
||||
def command_help(response: Response):
|
||||
print(response.status)
|
||||
print(response.undefined_flags.get_flags())
|
||||
@@ -20,12 +27,9 @@ def command_help(response: Response):
|
||||
print(response.invalid_value_flags.get_flags())
|
||||
|
||||
|
||||
@work_router.command('run')
|
||||
@work_router.command("run")
|
||||
def command_start_solving(response: Response):
|
||||
print(response.status)
|
||||
print(response.undefined_flags.get_flags())
|
||||
print(response.valid_flags.get_flags())
|
||||
print(response.invalid_value_flags.get_flags())
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -9,10 +9,12 @@ from argenta.orchestrator.argparser import ArgParser
|
||||
from argenta.orchestrator.argparser.arguments import BooleanArgument
|
||||
|
||||
|
||||
arg_parser = ArgParser(processed_args=[BooleanArgument('repeat')])
|
||||
app: App = App(dividing_line=DynamicDividingLine(),
|
||||
arg_parser = ArgParser(processed_args=[BooleanArgument("repeat")])
|
||||
app: App = App(
|
||||
dividing_line=DynamicDividingLine(),
|
||||
autocompleter=AutoCompleter(),
|
||||
repeat_command_groups=False,)
|
||||
repeat_command_groups=False,
|
||||
)
|
||||
orchestrator: Orchestrator = Orchestrator(arg_parser)
|
||||
|
||||
|
||||
@@ -25,5 +27,6 @@ def main():
|
||||
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -4,7 +4,9 @@ from typing import Never
|
||||
|
||||
|
||||
class AutoCompleter:
|
||||
def __init__(self, history_filename: str = False, autocomplete_button: str = 'tab') -> None:
|
||||
def __init__(
|
||||
self, history_filename: str = False, autocomplete_button: str = "tab"
|
||||
) -> None:
|
||||
"""
|
||||
Public. Configures and implements auto-completion of input command
|
||||
:param history_filename: the name of the file for saving the history of the autocompleter
|
||||
@@ -21,16 +23,22 @@ class AutoCompleter:
|
||||
:param state: the current cursor position is relative to the beginning of the line
|
||||
:return: the desired candidate as str or None
|
||||
"""
|
||||
matches: list[str] = sorted(cmd for cmd in self.get_history_items() if cmd.startswith(text))
|
||||
matches: list[str] = sorted(
|
||||
cmd for cmd in self.get_history_items() if cmd.startswith(text)
|
||||
)
|
||||
if len(matches) > 1:
|
||||
common_prefix = matches[0]
|
||||
for match in matches[1:]:
|
||||
i = 0
|
||||
while i < len(common_prefix) and i < len(match) and common_prefix[i] == match[i]:
|
||||
while (
|
||||
i < len(common_prefix)
|
||||
and i < len(match)
|
||||
and common_prefix[i] == match[i]
|
||||
):
|
||||
i += 1
|
||||
common_prefix = common_prefix[:i]
|
||||
if state == 0:
|
||||
readline.insert_text(common_prefix[len(text):])
|
||||
readline.insert_text(common_prefix[len(text) :])
|
||||
readline.redisplay()
|
||||
return None
|
||||
elif len(matches) == 1:
|
||||
@@ -52,8 +60,8 @@ class AutoCompleter:
|
||||
readline.add_history(line)
|
||||
|
||||
readline.set_completer(self._complete)
|
||||
readline.set_completer_delims(readline.get_completer_delims().replace(' ', ''))
|
||||
readline.parse_and_bind(f'{self.autocomplete_button}: complete')
|
||||
readline.set_completer_delims(readline.get_completer_delims().replace(" ", ""))
|
||||
readline.parse_and_bind(f"{self.autocomplete_button}: complete")
|
||||
|
||||
def exit_setup(self, all_commands: list[str]) -> None:
|
||||
"""
|
||||
@@ -62,14 +70,14 @@ class AutoCompleter:
|
||||
"""
|
||||
if self.history_filename:
|
||||
readline.write_history_file(self.history_filename)
|
||||
with open(self.history_filename, 'r') as history_file:
|
||||
with open(self.history_filename, "r") as history_file:
|
||||
raw_history = history_file.read()
|
||||
pretty_history: list[str] = []
|
||||
for line in set(raw_history.strip().split('\n')):
|
||||
for line in set(raw_history.strip().split("\n")):
|
||||
if line.split()[0] in all_commands:
|
||||
pretty_history.append(line)
|
||||
with open(self.history_filename, 'w') as history_file:
|
||||
history_file.write('\n'.join(pretty_history))
|
||||
with open(self.history_filename, "w") as history_file:
|
||||
history_file.write("\n".join(pretty_history))
|
||||
|
||||
@staticmethod
|
||||
def get_history_items() -> list[str] | list[Never]:
|
||||
@@ -77,4 +85,7 @@ class AutoCompleter:
|
||||
Private. Returns a list of all commands entered by the user
|
||||
:return: all commands entered by the user as list[str] | list[Never]
|
||||
"""
|
||||
return [readline.get_history_item(i) for i in range(1, readline.get_current_history_length() + 1)]
|
||||
return [
|
||||
readline.get_history_item(i)
|
||||
for i in range(1, readline.get_current_history_length() + 1)
|
||||
]
|
||||
|
||||
@@ -6,7 +6,7 @@ class PredefinedMessages:
|
||||
"""
|
||||
Public. A dataclass with predetermined messages for quick use
|
||||
"""
|
||||
USAGE = '[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]'
|
||||
HELP = '[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]'
|
||||
AUTOCOMPLETE = '[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>'
|
||||
|
||||
USAGE = "[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]"
|
||||
HELP = "[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]"
|
||||
AUTOCOMPLETE = "[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>"
|
||||
|
||||
@@ -2,7 +2,7 @@ from abc import ABC
|
||||
|
||||
|
||||
class BaseDividingLine(ABC):
|
||||
def __init__(self, unit_part: str = '-') -> None:
|
||||
def __init__(self, unit_part: str = "-") -> None:
|
||||
"""
|
||||
Private. The basic dividing line
|
||||
:param unit_part: the single part of the dividing line
|
||||
@@ -16,13 +16,13 @@ class BaseDividingLine(ABC):
|
||||
:return: unit_part of dividing line as str
|
||||
"""
|
||||
if len(self._unit_part) == 0:
|
||||
return ' '
|
||||
return " "
|
||||
else:
|
||||
return self._unit_part[0]
|
||||
|
||||
|
||||
class StaticDividingLine(BaseDividingLine):
|
||||
def __init__(self, unit_part: str = '-', length: int = 25) -> None:
|
||||
def __init__(self, unit_part: str = "-", length: int = 25) -> None:
|
||||
"""
|
||||
Public. The static dividing line
|
||||
:param unit_part: the single part of the dividing line
|
||||
@@ -39,13 +39,13 @@ class StaticDividingLine(BaseDividingLine):
|
||||
:return: full line of dividing line as str
|
||||
"""
|
||||
if is_override:
|
||||
return f'\n{self.length * self.get_unit_part()}\n'
|
||||
return f"\n{self.length * self.get_unit_part()}\n"
|
||||
else:
|
||||
return f'\n[dim]{self.length * self.get_unit_part()}[/dim]\n'
|
||||
return f"\n[dim]{self.length * self.get_unit_part()}[/dim]\n"
|
||||
|
||||
|
||||
class DynamicDividingLine(BaseDividingLine):
|
||||
def __init__(self, unit_part: str = '-') -> None:
|
||||
def __init__(self, unit_part: str = "-") -> None:
|
||||
"""
|
||||
Public. The dynamic dividing line
|
||||
:param unit_part: the single part of the dividing line
|
||||
@@ -61,8 +61,6 @@ class DynamicDividingLine(BaseDividingLine):
|
||||
:return: full line of dividing line as str
|
||||
"""
|
||||
if is_override:
|
||||
return f'\n{length * self.get_unit_part()}\n'
|
||||
return f"\n{length * self.get_unit_part()}\n"
|
||||
else:
|
||||
return f'\n[dim]{self.get_unit_part() * length}[/dim]\n'
|
||||
|
||||
|
||||
return f"\n[dim]{self.get_unit_part() * length}[/dim]\n"
|
||||
|
||||
+146
-81
@@ -11,17 +11,20 @@ from argenta.router import Router
|
||||
from argenta.router.defaults import system_router
|
||||
from argenta.app.autocompleter import AutoCompleter
|
||||
from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
from argenta.command.exceptions import (
|
||||
UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException,
|
||||
BaseInputCommandException)
|
||||
BaseInputCommandException,
|
||||
)
|
||||
from argenta.app.registered_routers.entity import RegisteredRouters
|
||||
from argenta.response import Response
|
||||
|
||||
|
||||
|
||||
class BaseApp:
|
||||
def __init__(self, prompt: str,
|
||||
def __init__(
|
||||
self,
|
||||
prompt: str,
|
||||
initial_message: str,
|
||||
farewell_message: str,
|
||||
exit_command: Command,
|
||||
@@ -31,8 +34,8 @@ class BaseApp:
|
||||
repeat_command_groups: bool,
|
||||
override_system_messages: bool,
|
||||
autocompleter: AutoCompleter,
|
||||
print_func: Callable[[str], None]) -> None:
|
||||
|
||||
print_func: Callable[[str], None],
|
||||
) -> None:
|
||||
self._prompt = prompt
|
||||
self._print_func = print_func
|
||||
self._exit_command = exit_command
|
||||
@@ -46,20 +49,30 @@ class BaseApp:
|
||||
self._farewell_message = farewell_message
|
||||
self._initial_message = initial_message
|
||||
|
||||
|
||||
self._description_message_gen: Callable[[str, str], str] = lambda command, description: f'[{command}] *=*=* {description}'
|
||||
self._description_message_gen: Callable[[str, str], str] = (
|
||||
lambda command, description: f"[{command}] *=*=* {description}"
|
||||
)
|
||||
self._registered_routers: RegisteredRouters = RegisteredRouters()
|
||||
self._messages_on_startup: list[str] = []
|
||||
|
||||
self._all_registered_triggers_in_lower: list[str] = []
|
||||
self._all_registered_triggers_in_default_case: list[str] = []
|
||||
|
||||
self._incorrect_input_syntax_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('Empty input command')
|
||||
self._unknown_command_handler: Callable[[InputCommand], None] = lambda command: print_func(f"Unknown command: {command.get_trigger()}")
|
||||
self._exit_command_handler: Callable[[Response], None] = lambda response: print_func(self._farewell_message)
|
||||
|
||||
self._incorrect_input_syntax_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(
|
||||
"Empty input command"
|
||||
)
|
||||
self._unknown_command_handler: Callable[[InputCommand], None] = (
|
||||
lambda command: print_func(f"Unknown command: {command.get_trigger()}")
|
||||
)
|
||||
self._exit_command_handler: Callable[[Response], None] = (
|
||||
lambda response: print_func(self._farewell_message)
|
||||
)
|
||||
|
||||
def set_description_message_pattern(self, _: Callable[[str, str], str]) -> None:
|
||||
"""
|
||||
@@ -69,7 +82,6 @@ class BaseApp:
|
||||
"""
|
||||
self._description_message_gen: Callable[[str, str], str] = _
|
||||
|
||||
|
||||
def set_incorrect_input_syntax_handler(self, _: Callable[[str], None]) -> None:
|
||||
"""
|
||||
Public. Sets the handler for incorrect flags when entering a command
|
||||
@@ -78,7 +90,6 @@ class BaseApp:
|
||||
"""
|
||||
self._incorrect_input_syntax_handler = _
|
||||
|
||||
|
||||
def set_repeated_input_flags_handler(self, _: Callable[[str], None]) -> None:
|
||||
"""
|
||||
Public. Sets the handler for repeated flags when entering a command
|
||||
@@ -87,7 +98,6 @@ class BaseApp:
|
||||
"""
|
||||
self._repeated_input_flags_handler = _
|
||||
|
||||
|
||||
def set_unknown_command_handler(self, _: Callable[[str], None]) -> None:
|
||||
"""
|
||||
Public. Sets the handler for unknown commands when entering a command
|
||||
@@ -96,7 +106,6 @@ class BaseApp:
|
||||
"""
|
||||
self._unknown_command_handler = _
|
||||
|
||||
|
||||
def set_empty_command_handler(self, _: Callable[[], None]) -> None:
|
||||
"""
|
||||
Public. Sets the handler for empty commands when entering a command
|
||||
@@ -105,7 +114,6 @@ class BaseApp:
|
||||
"""
|
||||
self._empty_input_command_handler = _
|
||||
|
||||
|
||||
def set_exit_command_handler(self, _: Callable[[], None]) -> None:
|
||||
"""
|
||||
Public. Sets the handler for exit command when entering a command
|
||||
@@ -114,7 +122,6 @@ class BaseApp:
|
||||
"""
|
||||
self._exit_command_handler = _
|
||||
|
||||
|
||||
def _print_command_group_description(self) -> None:
|
||||
"""
|
||||
Private. Prints the description of the available commands
|
||||
@@ -124,11 +131,13 @@ class BaseApp:
|
||||
if registered_router.title:
|
||||
self._print_func(registered_router.title)
|
||||
for command_handler in registered_router.get_command_handlers():
|
||||
self._print_func(self._description_message_gen(
|
||||
self._print_func(
|
||||
self._description_message_gen(
|
||||
command_handler.get_handled_command().get_trigger(),
|
||||
command_handler.get_handled_command().get_description()))
|
||||
self._print_func('')
|
||||
|
||||
command_handler.get_handled_command().get_description(),
|
||||
)
|
||||
)
|
||||
self._print_func("")
|
||||
|
||||
def _print_framed_text(self, text: str) -> None:
|
||||
"""
|
||||
@@ -137,19 +146,36 @@ class BaseApp:
|
||||
:return: None
|
||||
"""
|
||||
if isinstance(self._dividing_line, StaticDividingLine):
|
||||
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
|
||||
print(text.strip('\n'))
|
||||
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
|
||||
self._print_func(
|
||||
self._dividing_line.get_full_static_line(self._override_system_messages)
|
||||
)
|
||||
print(text.strip("\n"))
|
||||
self._print_func(
|
||||
self._dividing_line.get_full_static_line(self._override_system_messages)
|
||||
)
|
||||
|
||||
elif isinstance(self._dividing_line, DynamicDividingLine):
|
||||
clear_text = re.sub(r'\u001b\[[0-9;]*m', '', text)
|
||||
max_length_line = max([len(line) for line in clear_text.split('\n')])
|
||||
max_length_line = max_length_line if 10 <= max_length_line <= 80 else 80 if max_length_line > 80 else 10
|
||||
|
||||
self._print_func(self._dividing_line.get_full_dynamic_line(max_length_line, self._override_system_messages))
|
||||
print(text.strip('\n'))
|
||||
self._print_func(self._dividing_line.get_full_dynamic_line(max_length_line, self._override_system_messages))
|
||||
clear_text = re.sub(r"\u001b\[[0-9;]*m", "", text)
|
||||
max_length_line = max([len(line) for line in clear_text.split("\n")])
|
||||
max_length_line = (
|
||||
max_length_line
|
||||
if 10 <= max_length_line <= 80
|
||||
else 80
|
||||
if max_length_line > 80
|
||||
else 10
|
||||
)
|
||||
|
||||
self._print_func(
|
||||
self._dividing_line.get_full_dynamic_line(
|
||||
max_length_line, self._override_system_messages
|
||||
)
|
||||
)
|
||||
print(text.strip("\n"))
|
||||
self._print_func(
|
||||
self._dividing_line.get_full_dynamic_line(
|
||||
max_length_line, self._override_system_messages
|
||||
)
|
||||
)
|
||||
|
||||
def _is_exit_command(self, command: InputCommand) -> bool:
|
||||
"""
|
||||
@@ -158,9 +184,14 @@ class BaseApp:
|
||||
:return: is it an exit command or not as bool
|
||||
"""
|
||||
if self._ignore_command_register:
|
||||
if command.get_trigger().lower() == self._exit_command.get_trigger().lower():
|
||||
if (
|
||||
command.get_trigger().lower()
|
||||
== self._exit_command.get_trigger().lower()
|
||||
):
|
||||
return True
|
||||
elif command.get_trigger().lower() in [x.lower() for x in self._exit_command.get_aliases()]:
|
||||
elif command.get_trigger().lower() in [
|
||||
x.lower() for x in self._exit_command.get_aliases()
|
||||
]:
|
||||
return True
|
||||
else:
|
||||
if command.get_trigger() == self._exit_command.get_trigger():
|
||||
@@ -169,7 +200,6 @@ class BaseApp:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_unknown_command(self, command: InputCommand) -> bool:
|
||||
"""
|
||||
Private. Checks if the given command is an unknown command
|
||||
@@ -185,8 +215,9 @@ class BaseApp:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _error_handler(self, error: BaseInputCommandException, raw_command: str) -> None:
|
||||
def _error_handler(
|
||||
self, error: BaseInputCommandException, raw_command: str
|
||||
) -> None:
|
||||
"""
|
||||
Private. Handles parsing errors of the entered command
|
||||
:param error: error being handled
|
||||
@@ -201,7 +232,6 @@ class BaseApp:
|
||||
case EmptyInputCommandException():
|
||||
self._empty_input_command_handler()
|
||||
|
||||
|
||||
def _setup_system_router(self) -> None:
|
||||
"""
|
||||
Private. Sets up system router
|
||||
@@ -217,12 +247,19 @@ class BaseApp:
|
||||
system_router.set_command_register_ignore(self._ignore_command_register)
|
||||
self._registered_routers.add_registered_router(system_router)
|
||||
|
||||
|
||||
def _most_similar_command(self, unknown_command: str) -> str | None:
|
||||
all_commands = self._all_registered_triggers_in_lower if self._ignore_command_register else self._all_registered_triggers_in_default_case
|
||||
matches: list[str] | list = sorted(cmd for cmd in all_commands if cmd.startswith(unknown_command))
|
||||
all_commands = (
|
||||
self._all_registered_triggers_in_lower
|
||||
if self._ignore_command_register
|
||||
else self._all_registered_triggers_in_default_case
|
||||
)
|
||||
matches: list[str] | list = sorted(
|
||||
cmd for cmd in all_commands if cmd.startswith(unknown_command)
|
||||
)
|
||||
if not matches:
|
||||
matches: list[str] | list = sorted(cmd for cmd in all_commands if unknown_command.startswith(cmd))
|
||||
matches: list[str] | list = sorted(
|
||||
cmd for cmd in all_commands if unknown_command.startswith(cmd)
|
||||
)
|
||||
if len(matches) == 1:
|
||||
return matches[0]
|
||||
elif len(matches) > 1:
|
||||
@@ -230,33 +267,49 @@ class BaseApp:
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
def _setup_default_view(self) -> None:
|
||||
"""
|
||||
Private. Sets up default app view
|
||||
:return: None
|
||||
"""
|
||||
self._prompt = '[italic dim bold]What do you want to do?\n'
|
||||
self._initial_message = f'\n[bold red]{text2art(self._initial_message, font="tarty1")}\n'
|
||||
self._farewell_message = (f'[bold red]\n{text2art(f"\n{self._farewell_message}\n", font="chanky")}[/bold red]\n'
|
||||
f'[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n')
|
||||
self._description_message_gen = lambda command, description: (f'[bold red]{escape("[" + command + "]")}[/bold red] '
|
||||
f'[blue dim]*=*=*[/blue dim] '
|
||||
f'[bold yellow italic]{escape(description)}')
|
||||
self._incorrect_input_syntax_handler = lambda raw_command: self._print_func(f'[red bold]Incorrect flag syntax: {escape(raw_command)}')
|
||||
self._repeated_input_flags_handler = lambda raw_command: self._print_func(f'[red bold]Repeated input flags: {escape(raw_command)}')
|
||||
self._empty_input_command_handler = lambda: self._print_func('[red bold]Empty input command')
|
||||
self._prompt = "[italic dim bold]What do you want to do?\n"
|
||||
self._initial_message = (
|
||||
f"\n[bold red]{text2art(self._initial_message, font='tarty1')}\n"
|
||||
)
|
||||
self._farewell_message = (
|
||||
f"[bold red]\n{text2art(f'\n{self._farewell_message}\n', font='chanky')}[/bold red]\n"
|
||||
f"[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n"
|
||||
)
|
||||
self._description_message_gen = lambda command, description: (
|
||||
f"[bold red]{escape('[' + command + ']')}[/bold red] "
|
||||
f"[blue dim]*=*=*[/blue dim] "
|
||||
f"[bold yellow italic]{escape(description)}"
|
||||
)
|
||||
self._incorrect_input_syntax_handler = lambda raw_command: self._print_func(
|
||||
f"[red bold]Incorrect flag syntax: {escape(raw_command)}"
|
||||
)
|
||||
self._repeated_input_flags_handler = lambda raw_command: self._print_func(
|
||||
f"[red bold]Repeated input flags: {escape(raw_command)}"
|
||||
)
|
||||
self._empty_input_command_handler = lambda: self._print_func(
|
||||
"[red bold]Empty input command"
|
||||
)
|
||||
|
||||
def unknown_command_handler(command: InputCommand) -> None:
|
||||
cmd_trg: str = command.get_trigger()
|
||||
mst_sim_cmd: str | None = self._most_similar_command(cmd_trg)
|
||||
first_part_of_text = f"[red]Unknown command:[/red] [blue]{escape(cmd_trg)}[/blue]"
|
||||
second_part_of_text = ('[red], most similar:[/red] ' + ('[blue]' + mst_sim_cmd + '[/blue]')) if mst_sim_cmd else ''
|
||||
first_part_of_text = (
|
||||
f"[red]Unknown command:[/red] [blue]{escape(cmd_trg)}[/blue]"
|
||||
)
|
||||
second_part_of_text = (
|
||||
("[red], most similar:[/red] " + ("[blue]" + mst_sim_cmd + "[/blue]"))
|
||||
if mst_sim_cmd
|
||||
else ""
|
||||
)
|
||||
self._print_func(first_part_of_text + second_part_of_text)
|
||||
|
||||
self._unknown_command_handler = unknown_command_handler
|
||||
|
||||
|
||||
def _pre_cycle_setup(self) -> None:
|
||||
"""
|
||||
Private. Configures various aspects of the application before the start of the cycle
|
||||
@@ -265,11 +318,19 @@ class BaseApp:
|
||||
self._setup_system_router()
|
||||
|
||||
for router_entity in self._registered_routers:
|
||||
self._all_registered_triggers_in_default_case.extend(router_entity.get_triggers())
|
||||
self._all_registered_triggers_in_default_case.extend(router_entity.get_aliases())
|
||||
self._all_registered_triggers_in_default_case.extend(
|
||||
router_entity.get_triggers()
|
||||
)
|
||||
self._all_registered_triggers_in_default_case.extend(
|
||||
router_entity.get_aliases()
|
||||
)
|
||||
|
||||
self._all_registered_triggers_in_lower.extend([x.lower() for x in router_entity.get_triggers()])
|
||||
self._all_registered_triggers_in_lower.extend([x.lower() for x in router_entity.get_aliases()])
|
||||
self._all_registered_triggers_in_lower.extend(
|
||||
[x.lower() for x in router_entity.get_triggers()]
|
||||
)
|
||||
self._all_registered_triggers_in_lower.extend(
|
||||
[x.lower() for x in router_entity.get_aliases()]
|
||||
)
|
||||
|
||||
self._autocompleter.initial_setup(self._all_registered_triggers_in_lower)
|
||||
|
||||
@@ -281,26 +342,27 @@ class BaseApp:
|
||||
for message in self._messages_on_startup:
|
||||
self._print_func(message)
|
||||
if self._messages_on_startup:
|
||||
print('\n')
|
||||
print("\n")
|
||||
|
||||
if not self._repeat_command_groups_description:
|
||||
self._print_command_group_description()
|
||||
|
||||
|
||||
|
||||
class App(BaseApp):
|
||||
def __init__(self,
|
||||
prompt: str = 'What do you want to do?\n',
|
||||
initial_message: str = '\nArgenta\n',
|
||||
farewell_message: str = '\nSee you\n',
|
||||
exit_command: Command = Command('Q', 'Exit command'),
|
||||
system_router_title: str | None = 'System points:',
|
||||
def __init__(
|
||||
self,
|
||||
prompt: str = "What do you want to do?\n",
|
||||
initial_message: str = "\nArgenta\n",
|
||||
farewell_message: str = "\nSee you\n",
|
||||
exit_command: Command = Command("Q", "Exit command"),
|
||||
system_router_title: str | None = "System points:",
|
||||
ignore_command_register: bool = True,
|
||||
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
|
||||
repeat_command_groups: bool = True,
|
||||
override_system_messages: bool = False,
|
||||
autocompleter: AutoCompleter = AutoCompleter(),
|
||||
print_func: Callable[[str], None] = Console().print) -> None:
|
||||
print_func: Callable[[str], None] = Console().print,
|
||||
) -> None:
|
||||
"""
|
||||
Public. The essence of the application itself.
|
||||
Configures and manages all aspects of the behavior and presentation of the user interacting with the user
|
||||
@@ -317,7 +379,8 @@ class App(BaseApp):
|
||||
:param print_func: system messages text output function
|
||||
:return: None
|
||||
"""
|
||||
super().__init__(prompt=prompt,
|
||||
super().__init__(
|
||||
prompt=prompt,
|
||||
initial_message=initial_message,
|
||||
farewell_message=farewell_message,
|
||||
exit_command=exit_command,
|
||||
@@ -327,8 +390,8 @@ class App(BaseApp):
|
||||
repeat_command_groups=repeat_command_groups,
|
||||
override_system_messages=override_system_messages,
|
||||
autocompleter=autocompleter,
|
||||
print_func=print_func)
|
||||
|
||||
print_func=print_func,
|
||||
)
|
||||
|
||||
def run_polling(self) -> None:
|
||||
"""
|
||||
@@ -343,7 +406,9 @@ class App(BaseApp):
|
||||
raw_command: str = Console().input(self._prompt)
|
||||
|
||||
try:
|
||||
input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
|
||||
input_command: InputCommand = InputCommand.parse(
|
||||
raw_command=raw_command
|
||||
)
|
||||
except BaseInputCommandException as error:
|
||||
with redirect_stdout(io.StringIO()) as f:
|
||||
self._error_handler(error, raw_command)
|
||||
@@ -354,9 +419,13 @@ class App(BaseApp):
|
||||
if self._is_exit_command(input_command):
|
||||
system_router.finds_appropriate_handler(input_command)
|
||||
if self._ignore_command_register:
|
||||
self._autocompleter.exit_setup(self._all_registered_triggers_in_lower)
|
||||
self._autocompleter.exit_setup(
|
||||
self._all_registered_triggers_in_lower
|
||||
)
|
||||
else:
|
||||
self._autocompleter.exit_setup(self._all_registered_triggers_in_default_case)
|
||||
self._autocompleter.exit_setup(
|
||||
self._all_registered_triggers_in_default_case
|
||||
)
|
||||
return
|
||||
|
||||
if self._is_unknown_command(input_command):
|
||||
@@ -372,7 +441,6 @@ class App(BaseApp):
|
||||
res: str = f.getvalue()
|
||||
self._print_framed_text(res)
|
||||
|
||||
|
||||
def include_router(self, router: Router) -> None:
|
||||
"""
|
||||
Public. Registers the router in the application
|
||||
@@ -382,7 +450,6 @@ class App(BaseApp):
|
||||
router.set_command_register_ignore(self._ignore_command_register)
|
||||
self._registered_routers.add_registered_router(router)
|
||||
|
||||
|
||||
def include_routers(self, *routers: Router) -> None:
|
||||
"""
|
||||
Public. Registers the routers in the application
|
||||
@@ -392,7 +459,6 @@ class App(BaseApp):
|
||||
for router in routers:
|
||||
self.include_router(router)
|
||||
|
||||
|
||||
def add_message_on_startup(self, message: str) -> None:
|
||||
"""
|
||||
Public. Adds a message that will be displayed when the application is launched
|
||||
@@ -400,4 +466,3 @@ class App(BaseApp):
|
||||
:return: None
|
||||
"""
|
||||
self._messages_on_startup.append(message)
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ class BaseInputCommandException(Exception):
|
||||
"""
|
||||
Private. Base exception class for all exceptions raised when parse input command
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
@@ -12,6 +13,7 @@ class UnprocessedInputFlagException(BaseInputCommandException):
|
||||
"""
|
||||
Private. Raised when an unprocessed input flag is detected
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Unprocessed Input Flags"
|
||||
|
||||
@@ -20,16 +22,21 @@ class RepeatedInputFlagsException(BaseInputCommandException):
|
||||
"""
|
||||
Private. Raised when repeated input flags are detected
|
||||
"""
|
||||
|
||||
def __init__(self, flag: 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()}'")
|
||||
return (
|
||||
"Repeated Input Flags\n"
|
||||
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'"
|
||||
)
|
||||
|
||||
|
||||
class EmptyInputCommandException(BaseInputCommandException):
|
||||
"""
|
||||
Private. Raised when an empty input command is detected
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Input Command is empty"
|
||||
|
||||
@@ -8,17 +8,24 @@ class PredefinedFlags:
|
||||
"""
|
||||
Public. A dataclass with predefined flags and most frequently used flags for quick use
|
||||
"""
|
||||
HELP = Flag(name='help', possible_values=False)
|
||||
SHORT_HELP = Flag(name='H', prefix='-', possible_values=False)
|
||||
|
||||
INFO = Flag(name='info', possible_values=False)
|
||||
SHORT_INFO = Flag(name='I', prefix='-', possible_values=False)
|
||||
HELP = Flag(name="help", possible_values=False)
|
||||
SHORT_HELP = Flag(name="H", prefix="-", possible_values=False)
|
||||
|
||||
ALL = Flag(name='all', possible_values=False)
|
||||
SHORT_ALL = Flag(name='A', prefix='-', possible_values=False)
|
||||
INFO = Flag(name="info", possible_values=False)
|
||||
SHORT_INFO = Flag(name="I", prefix="-", possible_values=False)
|
||||
|
||||
HOST = Flag(name='host', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
|
||||
SHORT_HOST = Flag(name='H', prefix='-', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
|
||||
ALL = Flag(name="all", possible_values=False)
|
||||
SHORT_ALL = Flag(name="A", prefix="-", possible_values=False)
|
||||
|
||||
PORT = Flag(name='port', possible_values=re.compile(r'^\d{1,5}$'))
|
||||
SHORT_PORT = Flag(name='P', prefix='-', possible_values=re.compile(r'^\d{1,5}$'))
|
||||
HOST = Flag(
|
||||
name="host", possible_values=re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
|
||||
)
|
||||
SHORT_HOST = Flag(
|
||||
name="H",
|
||||
prefix="-",
|
||||
possible_values=re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$"),
|
||||
)
|
||||
|
||||
PORT = Flag(name="port", possible_values=re.compile(r"^\d{1,5}$"))
|
||||
SHORT_PORT = Flag(name="P", prefix="-", possible_values=re.compile(r"^\d{1,5}$"))
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
from typing import Literal, Pattern
|
||||
|
||||
|
||||
|
||||
class BaseFlag:
|
||||
def __init__(self, name: str,
|
||||
prefix: Literal['-', '--', '---'] = '--') -> None:
|
||||
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--") -> None:
|
||||
"""
|
||||
Private. Base class for flags
|
||||
:param name: the name of the flag
|
||||
@@ -41,9 +39,12 @@ class BaseFlag:
|
||||
|
||||
|
||||
class Flag(BaseFlag):
|
||||
def __init__(self, name: str,
|
||||
prefix: Literal['-', '--', '---'] = '--',
|
||||
possible_values: list[str] | Pattern[str] | False = True) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
prefix: Literal["-", "--", "---"] = "--",
|
||||
possible_values: list[str] | Pattern[str] | False = True,
|
||||
) -> None:
|
||||
"""
|
||||
Public. The entity of the flag being registered for subsequent processing
|
||||
:param name: The name of the flag
|
||||
@@ -85,9 +86,9 @@ class Flag(BaseFlag):
|
||||
|
||||
|
||||
class InputFlag(BaseFlag):
|
||||
def __init__(self, name: str,
|
||||
prefix: Literal['-', '--', '---'] = '--',
|
||||
value: str = None):
|
||||
def __init__(
|
||||
self, name: str, prefix: Literal["-", "--", "---"] = "--", value: str = None
|
||||
):
|
||||
"""
|
||||
Public. The entity of the flag of the entered command
|
||||
:param name: the name of the input flag
|
||||
@@ -114,5 +115,7 @@ class InputFlag(BaseFlag):
|
||||
self._flag_value = value
|
||||
|
||||
def __eq__(self, other) -> bool:
|
||||
return self.get_string_entity() == other.get_string_entity() and self.get_value() == other.get_value()
|
||||
|
||||
return (
|
||||
self.get_string_entity() == other.get_string_entity()
|
||||
and self.get_value() == other.get_value()
|
||||
)
|
||||
|
||||
@@ -1,10 +1,16 @@
|
||||
__all__ = ["Flags", "InputFlags",
|
||||
__all__ = [
|
||||
"Flags",
|
||||
"InputFlags",
|
||||
"UndefinedInputFlags",
|
||||
"InvalidValueInputFlags",
|
||||
"ValidInputFlags"]
|
||||
"ValidInputFlags",
|
||||
]
|
||||
|
||||
|
||||
from argenta.command.flags.models import (Flags, InputFlags,
|
||||
from argenta.command.flags.models import (
|
||||
Flags,
|
||||
InputFlags,
|
||||
UndefinedInputFlags,
|
||||
InvalidValueInputFlags,
|
||||
ValidInputFlags)
|
||||
ValidInputFlags,
|
||||
)
|
||||
|
||||
@@ -2,8 +2,7 @@ from argenta.command.flag.models import InputFlag, Flag
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
|
||||
|
||||
FlagType = TypeVar('FlagType')
|
||||
FlagType = TypeVar("FlagType")
|
||||
|
||||
|
||||
class BaseFlags(Generic[FlagType]):
|
||||
@@ -89,4 +88,3 @@ class UndefinedInputFlags(InputFlags):
|
||||
|
||||
class InvalidValueInputFlags(InputFlags):
|
||||
pass
|
||||
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
from argenta.command.flag.models import Flag, InputFlag
|
||||
from argenta.command.flags.models import InputFlags, Flags
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
from argenta.command.exceptions import (
|
||||
UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
EmptyInputCommandException,
|
||||
)
|
||||
from typing import Generic, TypeVar, cast, Literal
|
||||
|
||||
|
||||
InputCommandType = TypeVar('InputCommandType')
|
||||
InputCommandType = TypeVar("InputCommandType")
|
||||
|
||||
|
||||
class BaseCommand:
|
||||
@@ -26,10 +28,13 @@ class BaseCommand:
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
def __init__(self, trigger: str,
|
||||
def __init__(
|
||||
self,
|
||||
trigger: str,
|
||||
description: str = None,
|
||||
flags: Flag | Flags = None,
|
||||
aliases: list[str] = None):
|
||||
aliases: list[str] = None,
|
||||
):
|
||||
"""
|
||||
Public. The command that can and should be registered in the Router
|
||||
:param trigger: A string trigger, which, when entered by the user, indicates that the input corresponds to the command
|
||||
@@ -38,8 +43,14 @@ class Command(BaseCommand):
|
||||
:param aliases: string synonyms for the main trigger
|
||||
"""
|
||||
super().__init__(trigger)
|
||||
self._registered_flags: Flags = flags if isinstance(flags, Flags) else Flags(flags) if isinstance(flags, Flag) else Flags()
|
||||
self._description = 'Very useful command' if not description else description
|
||||
self._registered_flags: Flags = (
|
||||
flags
|
||||
if isinstance(flags, Flags)
|
||||
else Flags(flags)
|
||||
if isinstance(flags, Flag)
|
||||
else Flags()
|
||||
)
|
||||
self._description = "Very useful command" if not description else description
|
||||
self._aliases = aliases if isinstance(aliases, list) else []
|
||||
|
||||
def get_registered_flags(self) -> Flags:
|
||||
@@ -56,7 +67,9 @@ class Command(BaseCommand):
|
||||
"""
|
||||
return self._aliases
|
||||
|
||||
def validate_input_flag(self, flag: InputFlag) -> Literal['Undefined', 'Valid', 'Invalid']:
|
||||
def validate_input_flag(
|
||||
self, flag: InputFlag
|
||||
) -> Literal["Undefined", "Valid", "Invalid"]:
|
||||
"""
|
||||
Private. Validates the input flag
|
||||
:param flag: input flag for validation
|
||||
@@ -66,23 +79,27 @@ class Command(BaseCommand):
|
||||
if registered_flags:
|
||||
if isinstance(registered_flags, Flag):
|
||||
if registered_flags.get_string_entity() == flag.get_string_entity():
|
||||
is_valid = registered_flags.validate_input_flag_value(flag.get_value())
|
||||
is_valid = registered_flags.validate_input_flag_value(
|
||||
flag.get_value()
|
||||
)
|
||||
if is_valid:
|
||||
return 'Valid'
|
||||
return "Valid"
|
||||
else:
|
||||
return 'Invalid'
|
||||
return "Invalid"
|
||||
else:
|
||||
return 'Undefined'
|
||||
return "Undefined"
|
||||
else:
|
||||
for registered_flag in registered_flags:
|
||||
if registered_flag.get_string_entity() == flag.get_string_entity():
|
||||
is_valid = registered_flag.validate_input_flag_value(flag.get_value())
|
||||
is_valid = registered_flag.validate_input_flag_value(
|
||||
flag.get_value()
|
||||
)
|
||||
if is_valid:
|
||||
return 'Valid'
|
||||
return "Valid"
|
||||
else:
|
||||
return 'Invalid'
|
||||
return 'Undefined'
|
||||
return 'Undefined'
|
||||
return "Invalid"
|
||||
return "Undefined"
|
||||
return "Undefined"
|
||||
|
||||
def get_description(self) -> str:
|
||||
"""
|
||||
@@ -92,10 +109,8 @@ class Command(BaseCommand):
|
||||
return self._description
|
||||
|
||||
|
||||
|
||||
class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
def __init__(self, trigger: str,
|
||||
input_flags: InputFlag | InputFlags = None):
|
||||
def __init__(self, trigger: str, input_flags: InputFlag | InputFlags = None):
|
||||
"""
|
||||
Private. The model of the input command, after parsing
|
||||
:param trigger:the trigger of the command
|
||||
@@ -103,7 +118,13 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
:return: None
|
||||
"""
|
||||
super().__init__(trigger)
|
||||
self._input_flags: InputFlags = input_flags if isinstance(input_flags, InputFlags) else InputFlags(input_flags) if isinstance(input_flags, InputFlag) else InputFlags()
|
||||
self._input_flags: InputFlags = (
|
||||
input_flags
|
||||
if isinstance(input_flags, InputFlags)
|
||||
else InputFlags(input_flags)
|
||||
if isinstance(input_flags, InputFlag)
|
||||
else InputFlags()
|
||||
)
|
||||
|
||||
def _set_input_flags(self, input_flags: InputFlags) -> None:
|
||||
"""
|
||||
@@ -120,7 +141,6 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
"""
|
||||
return self._input_flags
|
||||
|
||||
|
||||
@staticmethod
|
||||
def parse(raw_command: str) -> InputCommandType:
|
||||
"""
|
||||
@@ -138,8 +158,8 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
current_flag_name, current_flag_value = None, None
|
||||
|
||||
for k, _ in enumerate(list_of_tokens):
|
||||
if _.startswith('-'):
|
||||
if len(_) < 2 or len(_[:_.rfind('-')]) > 3:
|
||||
if _.startswith("-"):
|
||||
if len(_) < 2 or len(_[: _.rfind("-")]) > 3:
|
||||
raise UnprocessedInputFlagException()
|
||||
current_flag_name = _
|
||||
else:
|
||||
@@ -148,16 +168,22 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
current_flag_value = _
|
||||
|
||||
if current_flag_name:
|
||||
if not len(list_of_tokens) == k+1:
|
||||
if not list_of_tokens[k+1].startswith('-'):
|
||||
if not len(list_of_tokens) == k + 1:
|
||||
if not list_of_tokens[k + 1].startswith("-"):
|
||||
continue
|
||||
|
||||
input_flag = InputFlag(name=current_flag_name[current_flag_name.rfind('-') + 1:],
|
||||
prefix=cast(Literal['-', '--', '---'],
|
||||
current_flag_name[:current_flag_name.rfind('-')+1]),
|
||||
value=current_flag_value)
|
||||
input_flag = InputFlag(
|
||||
name=current_flag_name[current_flag_name.rfind("-") + 1 :],
|
||||
prefix=cast(
|
||||
Literal["-", "--", "---"],
|
||||
current_flag_name[: current_flag_name.rfind("-") + 1],
|
||||
),
|
||||
value=current_flag_value,
|
||||
)
|
||||
|
||||
all_flags = [flag.get_string_entity() for flag in input_flags.get_flags()]
|
||||
all_flags = [
|
||||
flag.get_string_entity() for flag in input_flags.get_flags()
|
||||
]
|
||||
if input_flag.get_string_entity() not in all_flags:
|
||||
input_flags.add_flag(input_flag)
|
||||
else:
|
||||
@@ -169,4 +195,3 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
raise UnprocessedInputFlagException()
|
||||
else:
|
||||
return InputCommand(trigger=command, input_flags=input_flags)
|
||||
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
__all__ = ["BooleanArgument", "PositionalArgument", "OptionalArgument"]
|
||||
|
||||
|
||||
from argenta.orchestrator.argparser.arguments.models import (BooleanArgument,
|
||||
from argenta.orchestrator.argparser.arguments.models import (
|
||||
BooleanArgument,
|
||||
PositionalArgument,
|
||||
OptionalArgument)
|
||||
OptionalArgument,
|
||||
)
|
||||
|
||||
@@ -6,6 +6,7 @@ class BaseArgument(ABC):
|
||||
"""
|
||||
Private. Base class for all arguments
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get_string_entity(self) -> str:
|
||||
"""
|
||||
@@ -28,7 +29,7 @@ class PositionalArgument(BaseArgument):
|
||||
|
||||
|
||||
class OptionalArgument(BaseArgument):
|
||||
def __init__(self, name: str, prefix: Literal['-', '--', '---'] = '--'):
|
||||
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--"):
|
||||
"""
|
||||
Public. Optional argument, must have the value
|
||||
:param name: name of the argument
|
||||
@@ -42,7 +43,7 @@ class OptionalArgument(BaseArgument):
|
||||
|
||||
|
||||
class BooleanArgument(BaseArgument):
|
||||
def __init__(self, name: str, prefix: Literal['-', '--', '---'] = '--'):
|
||||
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--"):
|
||||
"""
|
||||
Public. Boolean argument, does not require a value
|
||||
:param name: name of the argument
|
||||
|
||||
@@ -1,16 +1,20 @@
|
||||
from argparse import ArgumentParser
|
||||
|
||||
from argenta.orchestrator.argparser.arguments.models import (BooleanArgument,
|
||||
from argenta.orchestrator.argparser.arguments.models import (
|
||||
BooleanArgument,
|
||||
OptionalArgument,
|
||||
PositionalArgument)
|
||||
PositionalArgument,
|
||||
)
|
||||
|
||||
|
||||
class ArgParser:
|
||||
def __init__(self,
|
||||
def __init__(
|
||||
self,
|
||||
processed_args: list[PositionalArgument | OptionalArgument | BooleanArgument],
|
||||
name: str = 'Argenta',
|
||||
description: str = 'Argenta available arguments',
|
||||
epilog: str = 'github.com/koloideal/Argenta | made by kolo') -> None:
|
||||
name: str = "Argenta",
|
||||
description: str = "Argenta available arguments",
|
||||
epilog: str = "github.com/koloideal/Argenta | made by kolo",
|
||||
) -> None:
|
||||
"""
|
||||
Public. Cmd argument parser and configurator at startup
|
||||
:param name: the name of the ArgParse instance
|
||||
@@ -22,10 +26,16 @@ class ArgParser:
|
||||
self.description = description
|
||||
self.epilog = epilog
|
||||
|
||||
self.entity: ArgumentParser = ArgumentParser(prog=name, description=description, epilog=epilog)
|
||||
self.args: list[PositionalArgument | OptionalArgument | BooleanArgument] | None = processed_args
|
||||
self.entity: ArgumentParser = ArgumentParser(
|
||||
prog=name, description=description, epilog=epilog
|
||||
)
|
||||
self.args: (
|
||||
list[PositionalArgument | OptionalArgument | BooleanArgument] | None
|
||||
) = processed_args
|
||||
|
||||
def set_args(self, *args: PositionalArgument | OptionalArgument | BooleanArgument) -> None:
|
||||
def set_args(
|
||||
self, *args: PositionalArgument | OptionalArgument | BooleanArgument
|
||||
) -> None:
|
||||
"""
|
||||
Public. Sets the arguments to be processed
|
||||
:param args: processed arguments
|
||||
@@ -46,4 +56,4 @@ class ArgParser:
|
||||
elif type(arg) is OptionalArgument:
|
||||
self.entity.add_argument(arg.get_string_entity())
|
||||
elif type(arg) is BooleanArgument:
|
||||
self.entity.add_argument(arg.get_string_entity(), action='store_true')
|
||||
self.entity.add_argument(arg.get_string_entity(), action="store_true")
|
||||
|
||||
@@ -33,4 +33,3 @@ class Orchestrator:
|
||||
return self.arg_parser.entity.parse_args()
|
||||
else:
|
||||
return None
|
||||
|
||||
@@ -1,19 +1,21 @@
|
||||
from argenta.response.status import Status
|
||||
from argenta.command.flags import (ValidInputFlags,
|
||||
from argenta.command.flags import (
|
||||
ValidInputFlags,
|
||||
UndefinedInputFlags,
|
||||
InvalidValueInputFlags)
|
||||
InvalidValueInputFlags,
|
||||
)
|
||||
|
||||
|
||||
class Response:
|
||||
__slots__ = ('status',
|
||||
'valid_flags',
|
||||
'undefined_flags',
|
||||
'invalid_value_flags')
|
||||
__slots__ = ("status", "valid_flags", "undefined_flags", "invalid_value_flags")
|
||||
|
||||
def __init__(self, status: Status = None,
|
||||
def __init__(
|
||||
self,
|
||||
status: Status = None,
|
||||
valid_flags: ValidInputFlags = ValidInputFlags(),
|
||||
undefined_flags: UndefinedInputFlags = UndefinedInputFlags(),
|
||||
invalid_value_flags: InvalidValueInputFlags = InvalidValueInputFlags()):
|
||||
invalid_value_flags: InvalidValueInputFlags = InvalidValueInputFlags(),
|
||||
):
|
||||
"""
|
||||
Public. The entity of the user input sent to the handler
|
||||
:param status: the status of the response
|
||||
|
||||
@@ -2,8 +2,7 @@ from enum import Enum
|
||||
|
||||
|
||||
class Status(Enum):
|
||||
ALL_FLAGS_VALID = 'ALL_FLAGS_VALID'
|
||||
UNDEFINED_FLAGS = 'UNDEFINED_FLAGS'
|
||||
INVALID_VALUE_FLAGS = 'INVALID_VALUE_FLAGS'
|
||||
UNDEFINED_AND_INVALID_FLAGS = 'UNDEFINED_AND_INVALID_FLAGS'
|
||||
|
||||
ALL_FLAGS_VALID = "ALL_FLAGS_VALID"
|
||||
UNDEFINED_FLAGS = "UNDEFINED_FLAGS"
|
||||
INVALID_VALUE_FLAGS = "INVALID_VALUE_FLAGS"
|
||||
UNDEFINED_AND_INVALID_FLAGS = "UNDEFINED_AND_INVALID_FLAGS"
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
system_router = Router(title='System points:')
|
||||
system_router = Router(title="System points:")
|
||||
|
||||
@@ -6,18 +6,23 @@ from argenta.command import Command
|
||||
from argenta.command.models import InputCommand
|
||||
from argenta.response import Response, Status
|
||||
from argenta.router.command_handler.entity import CommandHandlers, CommandHandler
|
||||
from argenta.command.flags.models import (Flags, InputFlags,
|
||||
from argenta.command.flags.models import (
|
||||
Flags,
|
||||
InputFlags,
|
||||
UndefinedInputFlags,
|
||||
ValidInputFlags,
|
||||
InvalidValueInputFlags)
|
||||
from argenta.router.exceptions import (RepeatedFlagNameException,
|
||||
InvalidValueInputFlags,
|
||||
)
|
||||
from argenta.router.exceptions import (
|
||||
RepeatedFlagNameException,
|
||||
TooManyTransferredArgsException,
|
||||
RequiredArgumentNotPassedException,
|
||||
TriggerContainSpacesException)
|
||||
TriggerContainSpacesException,
|
||||
)
|
||||
|
||||
|
||||
class Router:
|
||||
def __init__(self, title: str | None = 'Awesome title'):
|
||||
def __init__(self, title: str | None = "Awesome title"):
|
||||
"""
|
||||
Public. Directly configures and manages handlers
|
||||
:param title: the title of the router, displayed when displaying the available commands
|
||||
@@ -28,7 +33,6 @@ class Router:
|
||||
self._command_handlers: CommandHandlers = CommandHandlers()
|
||||
self._ignore_command_register: bool = False
|
||||
|
||||
|
||||
def command(self, command: Command | str) -> Callable:
|
||||
"""
|
||||
Public. Registers handler
|
||||
@@ -45,11 +49,11 @@ class Router:
|
||||
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return command_decorator
|
||||
|
||||
|
||||
def finds_appropriate_handler(self, input_command: InputCommand) -> None:
|
||||
"""
|
||||
Private. Finds the appropriate handler for given input command and passes control to it
|
||||
@@ -66,8 +70,9 @@ class Router:
|
||||
if input_command_name.lower() in handle_command.get_aliases():
|
||||
self.process_input_command(input_command_flags, command_handler)
|
||||
|
||||
|
||||
def process_input_command(self, input_command_flags: InputFlags, command_handler: CommandHandler) -> None:
|
||||
def process_input_command(
|
||||
self, input_command_flags: InputFlags, command_handler: CommandHandler
|
||||
) -> None:
|
||||
"""
|
||||
Private. Processes input command with the appropriate handler
|
||||
:param input_command_flags: input command flags as InputFlags
|
||||
@@ -78,7 +83,9 @@ class Router:
|
||||
response: Response = Response()
|
||||
if handle_command.get_registered_flags().get_flags():
|
||||
if input_command_flags.get_flags():
|
||||
response: Response = self._structuring_input_flags(handle_command, input_command_flags)
|
||||
response: Response = self._structuring_input_flags(
|
||||
handle_command, input_command_flags
|
||||
)
|
||||
command_handler.handling(response)
|
||||
else:
|
||||
response.status = Status.ALL_FLAGS_VALID
|
||||
@@ -93,9 +100,10 @@ class Router:
|
||||
response.status = Status.ALL_FLAGS_VALID
|
||||
command_handler.handling(response)
|
||||
|
||||
|
||||
@staticmethod
|
||||
def _structuring_input_flags(handled_command: Command, input_flags: InputFlags) -> Response:
|
||||
def _structuring_input_flags(
|
||||
handled_command: Command, input_flags: InputFlags
|
||||
) -> Response:
|
||||
"""
|
||||
Private. Validates flags of input command
|
||||
:param handled_command: entity of the handled command
|
||||
@@ -106,29 +114,41 @@ class Router:
|
||||
invalid_value_input_flags: InvalidValueInputFlags = InvalidValueInputFlags()
|
||||
undefined_input_flags: UndefinedInputFlags = UndefinedInputFlags()
|
||||
for flag in input_flags:
|
||||
flag_status: Literal['Undefined', 'Valid', 'Invalid'] = handled_command.validate_input_flag(flag)
|
||||
flag_status: Literal["Undefined", "Valid", "Invalid"] = (
|
||||
handled_command.validate_input_flag(flag)
|
||||
)
|
||||
match flag_status:
|
||||
case 'Valid':
|
||||
case "Valid":
|
||||
valid_input_flags.add_flag(flag)
|
||||
case 'Undefined':
|
||||
case "Undefined":
|
||||
undefined_input_flags.add_flag(flag)
|
||||
case 'Invalid':
|
||||
case "Invalid":
|
||||
invalid_value_input_flags.add_flag(flag)
|
||||
|
||||
if not invalid_value_input_flags.get_flags() and not undefined_input_flags.get_flags():
|
||||
if (
|
||||
not invalid_value_input_flags.get_flags()
|
||||
and not undefined_input_flags.get_flags()
|
||||
):
|
||||
status = Status.ALL_FLAGS_VALID
|
||||
elif invalid_value_input_flags.get_flags() and not undefined_input_flags.get_flags():
|
||||
elif (
|
||||
invalid_value_input_flags.get_flags()
|
||||
and not undefined_input_flags.get_flags()
|
||||
):
|
||||
status = Status.INVALID_VALUE_FLAGS
|
||||
elif not invalid_value_input_flags.get_flags() and undefined_input_flags.get_flags():
|
||||
elif (
|
||||
not invalid_value_input_flags.get_flags()
|
||||
and undefined_input_flags.get_flags()
|
||||
):
|
||||
status = Status.UNDEFINED_FLAGS
|
||||
else:
|
||||
status = Status.UNDEFINED_AND_INVALID_FLAGS
|
||||
|
||||
return Response(invalid_value_flags=invalid_value_input_flags,
|
||||
return Response(
|
||||
invalid_value_flags=invalid_value_input_flags,
|
||||
valid_flags=valid_input_flags,
|
||||
status=status,
|
||||
undefined_flags=undefined_input_flags)
|
||||
|
||||
undefined_flags=undefined_input_flags,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _validate_command(command: Command | str) -> None:
|
||||
@@ -138,20 +158,19 @@ class Router:
|
||||
:return: None if command is valid else raise exception
|
||||
"""
|
||||
match type(command).__name__:
|
||||
case 'Command':
|
||||
case "Command":
|
||||
command_name: str = command.get_trigger()
|
||||
if command_name.find(' ') != -1:
|
||||
if command_name.find(" ") != -1:
|
||||
raise TriggerContainSpacesException()
|
||||
flags: Flags = 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):
|
||||
raise RepeatedFlagNameException()
|
||||
case 'str':
|
||||
if command.find(' ') != -1:
|
||||
case "str":
|
||||
if command.find(" ") != -1:
|
||||
raise TriggerContainSpacesException()
|
||||
|
||||
|
||||
@staticmethod
|
||||
def _validate_func_args(func: Callable) -> None:
|
||||
"""
|
||||
@@ -173,13 +192,14 @@ class Router:
|
||||
pass
|
||||
else:
|
||||
file_path: str = getsourcefile(func)
|
||||
source_line: int = getsourcelines(func)[1]+1
|
||||
source_line: int = getsourcelines(func)[1] + 1
|
||||
fprint = Console().print
|
||||
fprint(f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
|
||||
f'of argument([green]{transferred_arg}[/green]) passed to the handler is [/i][bold blue]{Response}[/bold blue],'
|
||||
f' [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]\n', highlight=False)
|
||||
|
||||
|
||||
fprint(
|
||||
f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
|
||||
f"of argument([green]{transferred_arg}[/green]) passed to the handler is [/i][bold blue]{Response}[/bold blue],"
|
||||
f" [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]\n",
|
||||
highlight=False,
|
||||
)
|
||||
|
||||
def set_command_register_ignore(self, _: bool) -> None:
|
||||
"""
|
||||
@@ -189,7 +209,6 @@ class Router:
|
||||
"""
|
||||
self._ignore_command_register = _
|
||||
|
||||
|
||||
def get_triggers(self) -> list[str]:
|
||||
"""
|
||||
Public. Gets registered triggers
|
||||
@@ -200,7 +219,6 @@ class Router:
|
||||
all_triggers.append(command_handler.get_handled_command().get_trigger())
|
||||
return all_triggers
|
||||
|
||||
|
||||
def get_aliases(self) -> list[str]:
|
||||
"""
|
||||
Public. Gets registered aliases
|
||||
@@ -212,7 +230,6 @@ class Router:
|
||||
all_aliases.extend(command_handler.get_handled_command().get_aliases())
|
||||
return all_aliases
|
||||
|
||||
|
||||
def get_command_handlers(self) -> CommandHandlers:
|
||||
"""
|
||||
Private. Gets registered command handlers
|
||||
|
||||
@@ -2,6 +2,7 @@ class RepeatedFlagNameException(Exception):
|
||||
"""
|
||||
Private. Raised when a repeated flag name is registered
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Repeated registered flag names in register command"
|
||||
|
||||
@@ -10,6 +11,7 @@ class TooManyTransferredArgsException(Exception):
|
||||
"""
|
||||
Private. Raised when too many arguments are passed
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Too many transferred arguments"
|
||||
|
||||
@@ -18,6 +20,7 @@ class RequiredArgumentNotPassedException(Exception):
|
||||
"""
|
||||
Private. Raised when a required argument is not passed
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Required argument not passed"
|
||||
|
||||
@@ -26,5 +29,6 @@ class TriggerContainSpacesException(Exception):
|
||||
"""
|
||||
Private. Raised when there is a space in the trigger being registered
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Command trigger cannot contain spaces"
|
||||
|
||||
Reference in New Issue
Block a user