11 Commits

Author SHA1 Message Date
kolo 83955aa046 first beta - adding hints for similar commands, now - feature freezing 2025-05-04 02:13:05 +03:00
kolo 5a17e916eb work on stable major version 2025-04-30 15:48:38 +03:00
kolo 1159dda16e work on Response model 2025-04-30 00:08:49 +03:00
kolo 315508a36e work on Response 2025-04-29 20:40:47 +03:00
kolo 9d6598c4e0 work on Response model 2025-04-29 00:07:32 +03:00
kolo eb43806da6 new model - Response 2025-04-28 02:21:34 +03:00
kolo e076dbf84f new img in docs 2025-04-27 23:43:14 +03:00
kolo 2f090b6b47 new img in docs 2025-04-27 23:42:56 +03:00
kolo c9dbf2bbae fix print framed text with static dividing line 2025-04-27 23:27:08 +03:00
kolo e768c1bd2c fix 2025-04-27 21:29:14 +03:00
kolo 408450ec12 fix 2025-04-27 21:20:44 +03:00
21 changed files with 351 additions and 269 deletions
+1 -1
View File
@@ -74,7 +74,7 @@
---
![preview](https://github.com/koloideal/Argenta/blob/kolo/imgs/mock_app_preview3.png?raw=True)
![preview](https://github.com/koloideal/Argenta/blob/kolo/imgs/mock_app_preview4.png?raw=True)
An example of the TUI appearance
---
Binary file not shown.

After

Width:  |  Height:  |  Size: 50 KiB

+22 -3
View File
@@ -1,7 +1,26 @@
from mock.mock_app.handlers.routers import work_router
from argenta.app import App
from argenta.app.defaults import PredefinedMessages
from argenta.app.autocompleter import AutoCompleter
from argenta.orchestrator import Orchestrator
from argenta.orchestrator.argparser import ArgParser
from argenta.orchestrator.argparser.arguments import BooleanArgument
app = App(repeat_command_groups=True)
orchestrator = Orchestrator()
orchestrator.start_polling(app)
arg_parser = ArgParser(processed_args=[BooleanArgument('repeat')])
app: App = App(autocompleter=AutoCompleter('.hist'))
orchestrator: Orchestrator = Orchestrator()
def main():
app.include_router(work_router)
app.add_message_on_startup(PredefinedMessages.USAGE)
app.add_message_on_startup(PredefinedMessages.AUTOCOMPLETE)
app.add_message_on_startup(PredefinedMessages.HELP)
orchestrator.start_polling(app)
if __name__ == "__main__":
main()
+15 -4
View File
@@ -1,6 +1,17 @@
from argenta.response import Response, Status
from argenta.app import App
from argenta.command.models import InputCommand
from argenta.app.dividing_line import StaticDividingLine, DynamicDividingLine
from argenta.app.autocompleter import AutoCompleter
from argenta.app.defaults import PredefinedMessages
from argenta.command import Command
from argenta.command.flags import Flags, InputFlags, InvalidValueInputFlags, UndefinedInputFlags, ValidInputFlags
from argenta.command.flag import Flag, InputFlag
from argenta.command.flag.defaults import PredefinedFlags
from argenta.router import Router
from argenta.orchestrator import Orchestrator
app = App()
app._all_registered_triggers_in_lower = ['fr', 'Tre', 'Pre']
print(app._is_unknown_command(InputCommand('fr')))
from argenta.command.models import InputCommand
from argenta.app.utils import most_similar_command
print(most_similar_command('case', ['case', 'tester', 'poster', 'caser']))
+14 -5
View File
@@ -1,6 +1,9 @@
from rich.console import Console
from argenta.command import Command
from argenta.command.flag.defaults import PredefinedFlags
from argenta.command.flags import Flags
from argenta.response import Response
from argenta.router import Router
@@ -9,14 +12,20 @@ work_router: Router = Router(title='Work points:')
console = Console()
'''@work_router.command(Command('get', 'Get Help', aliases=['help', 'Get_help']))
def command_help():
pass
@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())
print(response.valid_flags.get_flags())
print(response.invalid_value_flags.get_flags())
@work_router.command(Command('run', 'Run All'))
def command_start_solving():
pass'''
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())
+2 -1
View File
@@ -11,7 +11,8 @@ from argenta.orchestrator.argparser.arguments import BooleanArgument
arg_parser = ArgParser(processed_args=[BooleanArgument('repeat')])
app: App = App(dividing_line=DynamicDividingLine(),
autocompleter=AutoCompleter('./mock/.hist'))
autocompleter=AutoCompleter('./mock/.hist'),
repeat_command_groups=False,)
orchestrator: Orchestrator = Orchestrator(arg_parser)
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "argenta"
version = "1.0.0-alpha2"
version = "1.0.0-beta1"
description = "Python library for creating TUI"
authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }]
requires-python = ">=3.11, <4.0"
+2 -2
View File
@@ -14,7 +14,7 @@ class AutoCompleter:
self.autocomplete_button = autocomplete_button
self.matches: list[str] = []
def complete(self, text, state) -> str | None:
def _complete(self, text, state) -> str | None:
"""
Private. Auto-completion function
:param text: part of the command being entered
@@ -51,7 +51,7 @@ class AutoCompleter:
for line in all_commands:
readline.add_history(line)
readline.set_completer(self.complete)
readline.set_completer(self._complete)
readline.set_completer_delims(readline.get_completer_delims().replace(' ', ''))
readline.parse_and_bind(f'{self.autocomplete_button}: complete')
+50 -29
View File
@@ -16,6 +16,7 @@ from argenta.command.exceptions import (UnprocessedInputFlagException,
EmptyInputCommandException,
BaseInputCommandException)
from argenta.app.registered_routers.entity import RegisteredRouters
from argenta.response import Response
@@ -58,61 +59,61 @@ class BaseApp:
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[[], None] = lambda: print_func(self._farewell_message)
self._exit_command_handler: Callable[[Response], None] = lambda response: print_func(self._farewell_message)
def set_description_message_pattern(self, pattern: Callable[[str, str], str]) -> None:
def set_description_message_pattern(self, _: Callable[[str, str], str]) -> None:
"""
Public. Sets the output pattern of the available commands
:param pattern: output pattern of the available commands
:param _: output pattern of the available commands
:return: None
"""
self._description_message_gen: Callable[[str, str], str] = pattern
self._description_message_gen: Callable[[str, str], str] = _
def set_invalid_input_flags_handler(self, handler: Callable[[str], None]) -> None:
def set_invalid_input_flags_handler(self, _: Callable[[str], None]) -> None:
"""
Public. Sets the handler for incorrect flags when entering a command
:param handler: handler for incorrect flags when entering a command
:param _: handler for incorrect flags when entering a command
:return: None
"""
self._invalid_input_flags_handler = handler
self._invalid_input_flags_handler = _
def set_repeated_input_flags_handler(self, handler: Callable[[str], None]) -> None:
def set_repeated_input_flags_handler(self, _: Callable[[str], None]) -> None:
"""
Public. Sets the handler for repeated flags when entering a command
:param handler: handler for repeated flags when entering a command
:param _: handler for repeated flags when entering a command
:return: None
"""
self._repeated_input_flags_handler = handler
self._repeated_input_flags_handler = _
def set_unknown_command_handler(self, handler: Callable[[str], None]) -> None:
def set_unknown_command_handler(self, _: Callable[[str], None]) -> None:
"""
Public. Sets the handler for unknown commands when entering a command
:param handler: handler for unknown commands when entering a command
:param _: handler for unknown commands when entering a command
:return: None
"""
self._unknown_command_handler = handler
self._unknown_command_handler = _
def set_empty_command_handler(self, handler: Callable[[], None]) -> None:
def set_empty_command_handler(self, _: Callable[[], None]) -> None:
"""
Public. Sets the handler for empty commands when entering a command
:param handler: handler for empty commands when entering a command
:param _: handler for empty commands when entering a command
:return: None
"""
self._empty_input_command_handler = handler
self._empty_input_command_handler = _
def set_exit_command_handler(self, handler: Callable[[], None]) -> None:
def set_exit_command_handler(self, _: Callable[[], None]) -> None:
"""
Public. Sets the handler for exit command when entering a command
:param handler: handler for exit command when entering a command
:param _: handler for exit command when entering a command
:return: None
"""
self._exit_command_handler = handler
self._exit_command_handler = _
def _print_command_group_description(self) -> None:
@@ -138,7 +139,7 @@ class BaseApp:
"""
if isinstance(self._dividing_line, StaticDividingLine):
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
self._print_func(text)
print(text.strip('\n'))
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
elif isinstance(self._dividing_line, DynamicDividingLine):
@@ -210,20 +211,33 @@ class BaseApp:
system_router.set_title(self._system_router_title)
@system_router.command(self._exit_command)
def exit_command():
self._exit_command_handler()
def exit_command(response: Response) -> None:
self._exit_command_handler(response)
if system_router not in self._registered_routers.get_registered_routers():
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))
if not matches:
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:
return sorted(matches, key=lambda cmd: len(cmd))[0]
else:
return None
def _setup_default_view(self) -> None:
"""
Private. Sets up default app view
:return: None
"""
if not self._override_system_messages:
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\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')
@@ -233,7 +247,15 @@ class BaseApp:
self._invalid_input_flags_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._unknown_command_handler = lambda command: self._print_func(f"[red bold]Unknown command: {escape(command.get_trigger())}")
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 ''
self._print_func(first_part_of_text + second_part_of_text)
self._unknown_command_handler = unknown_command_handler
def _pre_cycle_setup(self) -> None:
@@ -241,7 +263,6 @@ class BaseApp:
Private. Configures various aspects of the application before the start of the cycle
:return: None
"""
self._setup_default_view()
self._setup_system_router()
for router_entity in self._registered_routers:
@@ -253,6 +274,9 @@ class BaseApp:
self._autocompleter.initial_setup(self._all_registered_triggers_in_lower)
if not self._override_system_messages:
self._setup_default_view()
self._print_func(self._initial_message)
for message in self._messages_on_startup:
@@ -267,7 +291,7 @@ class BaseApp:
class App(BaseApp):
def __init__(self,
prompt: str = '[italic dim bold]What do you want to do?\n',
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'),
@@ -346,9 +370,6 @@ class App(BaseApp):
res: str = f.getvalue()
self._print_framed_text(res)
if not self._repeat_command_groups_description:
self._print_func(self._prompt)
def include_router(self, router: Router) -> None:
"""
+4 -10
View File
@@ -1,3 +1,5 @@
from typing import Iterator
from argenta.router import Router
@@ -25,16 +27,8 @@ class RegisteredRouters:
"""
self._registered_routers.append(router)
def add_registered_routers(self, *routers: Router) -> None:
"""
Private. Adds new registered routers
:param routers: registered routers
:return: None
"""
self._registered_routers.extend(routers)
def __iter__(self):
def __iter__(self) -> Iterator[Router]:
return iter(self._registered_routers)
def __next__(self):
def __next__(self) -> Router:
return next(iter(self._registered_routers))
+1 -1
View File
@@ -1,4 +1,4 @@
from argenta.command.flag.models import InputFlag, Flag
from argenta.command.flag.models import Flag, InputFlag
class BaseInputCommandException(Exception):
+2 -2
View File
@@ -1,4 +1,4 @@
__all__ = ('InputFlags', 'InputFlag', 'Flag', 'Flags')
__all__ = ["Flag", "InputFlag"]
from argenta.command.flag.models import InputFlags, InputFlag, Flags, Flag
from argenta.command.flag.models import Flag, InputFlag
+23 -129
View File
@@ -1,8 +1,8 @@
from typing import Literal, Pattern
from abc import ABC, abstractmethod
class BaseFlag(ABC):
class BaseFlag:
def __init__(self, name: str,
prefix: Literal['-', '--', '---'] = '--') -> None:
"""
@@ -37,38 +37,6 @@ class BaseFlag(ABC):
return self._prefix
class InputFlag(BaseFlag):
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
:param prefix: the prefix of the input flag
:param value: the value of the input flag
:return: None
"""
super().__init__(name, prefix)
self._flag_value = value
def get_value(self) -> str | None:
"""
Public. Returns the value of the flag
:return: the value of the flag as str
"""
return self._flag_value
def set_value(self, value):
"""
Private. Sets the value of the flag
:param value: the fag value to set
:return: None
"""
self._flag_value = value
class Flag(BaseFlag):
def __init__(self, name: str,
prefix: Literal['-', '--', '---'] = '--',
@@ -113,106 +81,32 @@ class Flag(BaseFlag):
return True
class BaseFlags(ABC):
class InputFlag(BaseFlag):
def __init__(self, name: str,
prefix: Literal['-', '--', '---'] = '--',
value: str = None):
"""
Private. Base class for groups of flags
"""
__slots__ = ('_flags',)
@abstractmethod
def get_flags(self):
"""
Public. Returns a list of flags
:return: list of flags
"""
pass
@abstractmethod
def add_flag(self, flag: Flag | InputFlag):
"""
Public. Adds a flag to the list of flags
:param flag: flag to add
Public. The entity of the flag of the entered command
:param name: the name of the input flag
:param prefix: the prefix of the input flag
:param value: the value of the input flag
:return: None
"""
pass
super().__init__(name, prefix)
self._flag_value = value
@abstractmethod
def add_flags(self, flags: list[Flag] | list[InputFlag]):
def get_value(self) -> str | None:
"""
Public. Adds a list of flags to the list of flags
:param flags: list of flags to add
Public. Returns the value of the flag
:return: the value of the flag as str
"""
return self._flag_value
def set_value(self, value):
"""
Private. Sets the value of the flag
:param value: the fag value to set
:return: None
"""
pass
@abstractmethod
def get_flag(self, name: str):
"""
Public. Returns the flag entity by its name or None if not found
:param name: the name of the flag to get
:return: entity of the flag or None
"""
pass
def __iter__(self):
return iter(self._flags)
def __next__(self):
return next(iter(self))
def __getitem__(self, item):
return self._flags[item]
class Flags(BaseFlags, ABC):
def __init__(self, *flags: Flag):
"""
Public. A model that combines the registered flags
:param flags: the flags that will be registered
:return: None
"""
self._flags = flags if flags else []
def get_flags(self) -> list[Flag]:
return self._flags
def add_flag(self, flag: Flag):
self._flags.append(flag)
def add_flags(self, flags: list[Flag]):
self._flags.extend(flags)
def get_flag(self, name: str) -> Flag | None:
if name in [flag.get_name() for flag in self._flags]:
return list(filter(lambda flag: flag.get_name() == name, self._flags))[0]
else:
return None
class InputFlags(BaseFlags, ABC):
def __init__(self, *flags: InputFlag):
"""
Public. A model that combines the input flags of the input command
:param flags: all input flags
:return: None
"""
self._flags = flags if flags else []
def get_flags(self) -> list[InputFlag]:
return self._flags
def add_flag(self, flag: InputFlag):
self._flags.append(flag)
def add_flags(self, flags: list[InputFlag]):
self._flags.extend(flags)
def get_flag(self, name: str) -> InputFlag | None:
if name in [flag.get_name() for flag in self._flags]:
return list(filter(lambda flag: flag.get_name() == name, self._flags))[0]
else:
return None
self._flag_value = value
+10
View File
@@ -0,0 +1,10 @@
__all__ = ["Flags", "InputFlags",
"UndefinedInputFlags",
"InvalidValueInputFlags",
"ValidInputFlags"]
from argenta.command.flags.models import (Flags, InputFlags,
UndefinedInputFlags,
InvalidValueInputFlags,
ValidInputFlags)
+75
View File
@@ -0,0 +1,75 @@
from argenta.command.flag.models import InputFlag, Flag
from typing import Generic, TypeVar
FlagType = TypeVar('FlagType')
class BaseFlags(Generic[FlagType]):
def __init__(self, *flags: FlagType):
"""
Public. A model that combines the registered flags
:param flags: the flags that will be registered
:return: None
"""
self._flags = flags if flags else []
def get_flags(self) -> list[FlagType]:
"""
Public. Returns a list of flags
:return: list of flags as list[FlagType]
"""
return self._flags
def add_flag(self, flag: FlagType):
"""
Public. Adds a flag to the list of flags
:param flag: flag to add
:return: None
"""
self._flags.append(flag)
def add_flags(self, flags: list[FlagType]):
"""
Public. Adds a list of flags to the list of flags
:param flags: list of flags to add
:return: None
"""
self._flags.extend(flags)
def get_flag(self, name: str) -> FlagType | None:
"""
Public. Returns the flag entity by its name or None if not found
:param name: the name of the flag to get
:return: entity of the flag or None
"""
if name in [flag.get_name() for flag in self._flags]:
return list(filter(lambda flag: flag.get_name() == name, self._flags))[0]
else:
return None
def __iter__(self):
return iter(self._flags)
def __next__(self):
return next(iter(self))
def __getitem__(self, item):
return self._flags[item]
class Flags(BaseFlags[Flag]): pass
class InputFlags(BaseFlags[InputFlag]): pass
class ValidInputFlags(InputFlags): pass
class UndefinedInputFlags(InputFlags): pass
class InvalidValueInputFlags(InputFlags): pass
+16 -8
View File
@@ -1,4 +1,5 @@
from argenta.command.flag.models import Flag, InputFlag, Flags, InputFlags
from argenta.command.flag.models import Flag, InputFlag
from argenta.command.flags.models import InputFlags, Flags
from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException)
@@ -55,7 +56,7 @@ class Command(BaseCommand):
"""
return self._aliases
def validate_input_flag(self, flag: InputFlag) -> bool:
def validate_input_flag(self, flag: InputFlag) -> Literal['Undefined', 'Valid', 'Invalid']:
"""
Private. Validates the input flag
:param flag: input flag for validation
@@ -67,14 +68,21 @@ class Command(BaseCommand):
if registered_flags.get_string_entity() == flag.get_string_entity():
is_valid = registered_flags.validate_input_flag_value(flag.get_value())
if is_valid:
return True
return 'Valid'
else:
return 'Invalid'
else:
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())
if is_valid:
return True
return False
return 'Valid'
else:
return 'Invalid'
return 'Undefined'
return 'Undefined'
def get_description(self) -> str:
"""
@@ -131,11 +139,11 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
for k, _ in enumerate(list_of_tokens):
if _.startswith('-'):
if current_flag_name or len(_) < 2 or len(_[:_.rfind('-')]) > 3:
if len(_) < 2 or len(_[:_.rfind('-')]) > 3:
raise UnprocessedInputFlagException()
current_flag_name = _
else:
if not current_flag_name:
if not current_flag_name or current_flag_value:
raise UnprocessedInputFlagException()
current_flag_value = _
@@ -144,7 +152,7 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
if not list_of_tokens[k+1].startswith('-'):
continue
input_flag = InputFlag(name=current_flag_name[current_flag_name.rfind('-')+1:],
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)
+5
View File
@@ -0,0 +1,5 @@
__all__ = ["Response", "Status"]
from argenta.response.entity import Response
from argenta.response.status import Status
+20
View File
@@ -0,0 +1,20 @@
from argenta.command.flags.models import ValidInputFlags, UndefinedInputFlags, InvalidValueInputFlags
from argenta.response.status import Status
class Response:
def __init__(self, status: Status = None,
valid_flags: ValidInputFlags = ValidInputFlags(),
undefined_flags: UndefinedInputFlags = UndefinedInputFlags(),
invalid_value_flags: InvalidValueInputFlags = InvalidValueInputFlags()):
"""
Public. The entity of the user input sent to the handler
:param status: the status of the response
:param valid_flags: valid input flags
:param undefined_flags: undefined input flags
:param invalid_value_flags: input flags with invalid values
"""
self.status = status
self.valid_flags = valid_flags
self.undefined_flags = undefined_flags
self.invalid_value_flags = invalid_value_flags
+9
View File
@@ -0,0 +1,9 @@
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'
+10 -22
View File
@@ -1,12 +1,11 @@
from typing import Callable
from typing import Callable, Iterator
from argenta.command import Command
from argenta.command.flag import InputFlags
from argenta.response import Response
class CommandHandler:
def __init__(self, handler: Callable[[], None] | Callable[[InputFlags], None], handled_command: Command):
def __init__(self, handler: Callable[[Response], None], handled_command: Command):
"""
Private. Entity of the model linking the handler and the command being processed
:param handler: the handler being called
@@ -15,21 +14,18 @@ class CommandHandler:
self._handler = handler
self._handled_command = handled_command
def handling(self, input_flags: InputFlags = None) -> None:
def handling(self, response: Response) -> None:
"""
Private. Direct processing of an input command
:param input_flags: the flags of the input command
:param response: the entity of response: various groups of flags and status of response
:return: None
"""
if input_flags is not None:
self._handler(input_flags)
else:
self._handler()
self._handler(response)
def get_handler(self) -> Callable[[], None] | Callable[[InputFlags], None]:
def get_handler(self) -> Callable[[Response], None]:
"""
Private. Returns the handler being called
:return: the handler being called as Callable[[], None] or Callable[[InputFlags], None]
:return: the handler being called as Callable[[Response], None]
"""
return self._handler
@@ -64,16 +60,8 @@ class CommandHandlers:
"""
self.command_handlers.append(command_handler)
def add_handlers(self, *command_handlers: CommandHandler) -> None:
"""
Private. Extend a many CommandHandler to the list of CommandHandlers
:param command_handlers: many CommandHandler to be added
:return: None
"""
self.command_handlers.extend(command_handlers)
def __iter__(self):
def __iter__(self) -> Iterator[CommandHandler]:
return iter(self.command_handlers)
def __next__(self):
def __next__(self) -> CommandHandler:
return next(iter(self.command_handlers))
+58 -40
View File
@@ -1,9 +1,12 @@
from typing import Callable
from inspect import getfullargspec
from typing import Callable, Literal, Type
from inspect import getfullargspec, get_annotations
from rich.console import Console
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.flag.models import Flag, Flags, InputFlags
from argenta.command.flags.models import Flags, InputFlags, UndefinedInputFlags, ValidInputFlags, InvalidValueInputFlags
from argenta.router.exceptions import (RepeatedFlagNameException,
TooManyTransferredArgsException,
RequiredArgumentNotPassedException,
@@ -11,8 +14,7 @@ from argenta.router.exceptions import (RepeatedFlagNameException,
class Router:
def __init__(self,
title: str = None):
def __init__(self, title: str = None):
"""
Public. Directly configures and manages handlers
:param title: the title of the router, displayed when displaying the available commands
@@ -22,19 +24,18 @@ class Router:
self._command_handlers: CommandHandlers = CommandHandlers()
self._ignore_command_register: bool = False
self._not_valid_flag_handler: Callable[[Flag], None] = lambda flag: print(f"Undefined or incorrect input flag: {flag.get_string_entity()}{(' '+flag.get_value()) if flag.get_value() else ''}")
def command(self, command: Command) -> Callable:
"""
Public. Registers handler
:param command: Registered command
:return: decorated handler as Callable[[Any], Any]
:return: decorated handler as Callable
"""
self._validate_command(command)
def command_decorator(func):
Router._validate_func_args(command, func)
Router._validate_func_args(func)
self._command_handlers.add_handler(CommandHandler(func, command))
def wrapper(*args, **kwargs):
@@ -44,15 +45,6 @@ class Router:
return command_decorator
def set_invalid_input_flag_handler(self, func: Callable[[Flag], None]) -> None:
"""
Public. Registers handler for invalid input flag
:param func: registered handler
:return: None
"""
self._not_valid_flag_handler = func
def finds_appropriate_handler(self, input_command: InputCommand) -> None:
"""
Private. Finds the appropriate handler for given input command and passes control to it
@@ -78,36 +70,61 @@ class Router:
:return: None
"""
handle_command = command_handler.get_handled_command()
response: Response = Response()
if handle_command.get_registered_flags().get_flags():
if input_command_flags.get_flags():
if self._validate_input_flags(handle_command, input_command_flags):
command_handler.handling(input_command_flags)
return
flags, status = self._validate_input_flags(handle_command, input_command_flags)
response.valid_flags, response.undefined_flags, response.invalid_value_flags = flags
response.status = status
command_handler.handling(response)
else:
command_handler.handling(input_command_flags)
return
response.status = Status.ALL_FLAGS_VALID
command_handler.handling(response)
else:
if input_command_flags.get_flags():
self._not_valid_flag_handler(input_command_flags[0])
return
response.status = Status.UNDEFINED_FLAGS
response.undefined_flags = UndefinedInputFlags()
response.undefined_flags.add_flags(input_command_flags.get_flags())
command_handler.handling(response)
else:
command_handler.handling()
return
response.status = Status.ALL_FLAGS_VALID
command_handler.handling(response)
def _validate_input_flags(self, handled_command: Command, input_flags: InputFlags) -> bool:
@staticmethod
def _validate_input_flags(handled_command: Command, input_flags: InputFlags) -> tuple[tuple[ValidInputFlags,
UndefinedInputFlags,
InvalidValueInputFlags],
Status]:
"""
Private. Validates flags of input command
:param handled_command: entity of the handled command
:param input_flags:
:return: is flags of input command valid as bool
"""
valid_input_flags: ValidInputFlags = ValidInputFlags()
invalid_value_input_flags: InvalidValueInputFlags = InvalidValueInputFlags()
undefined_input_flags: UndefinedInputFlags = UndefinedInputFlags()
for flag in input_flags:
is_valid: bool = handled_command.validate_input_flag(flag)
if not is_valid:
self._not_valid_flag_handler(flag)
return False
return True
flag_status: Literal['Undefined', 'Valid', 'Invalid'] = handled_command.validate_input_flag(flag)
match flag_status:
case 'Valid':
valid_input_flags.add_flag(flag)
case 'Undefined':
undefined_input_flags.add_flag(flag)
case 'Invalid':
invalid_value_input_flags.add_flag(flag)
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():
status = Status.INVALID_VALUE_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 (valid_input_flags, undefined_input_flags, invalid_value_input_flags), status
@staticmethod
@@ -129,22 +146,23 @@ class Router:
@staticmethod
def _validate_func_args(command: Command, func: Callable) -> None:
def _validate_func_args(func: Callable) -> None:
"""
Private. Validates the arguments of the handler
:param command: registered command in handler
:param func: entity of the handler func
:return: None if func is valid else raise exception
"""
registered_args = command.get_registered_flags()
transferred_args = getfullargspec(func).args
if registered_args.get_flags() and transferred_args:
if len(transferred_args) != 1:
if len(transferred_args) > 1:
raise TooManyTransferredArgsException()
elif registered_args.get_flags() and not transferred_args:
elif len(transferred_args) == 0:
raise RequiredArgumentNotPassedException()
elif not registered_args.get_flags() and transferred_args:
raise TooManyTransferredArgsException()
arg_annotation: Type = get_annotations(func)[transferred_args[0]]
if not arg_annotation is Response:
Console().print(f'\n\n[b red]WARNING:[/b red] [i]The type of argument passed to the handler is [/i][blue]{Response}[/blue],'
f' [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]', highlight=False)
def set_command_register_ignore(self, _: bool) -> None: