diff --git a/metrics/__main__.py b/metrics/__main__.py index 803eced..1f7fc08 100644 --- a/metrics/__main__.py +++ b/metrics/__main__.py @@ -1,9 +1,9 @@ -from argenta import App, Orchestrator +from argenta import App, Orchestrator, Command from argenta.app import DynamicDividingLine from .handlers import router -app = App(initial_message="metrics", dividing_line=DynamicDividingLine('~'), override_system_messages=True) +app = App(initial_message="metrics", exit_command=Command('exit', aliases=['quit'])) orchestrator = Orchestrator() diff --git a/src/argenta/app/behavior_handlers/entity.py b/src/argenta/app/behavior_handlers/entity.py deleted file mode 100644 index db195e2..0000000 --- a/src/argenta/app/behavior_handlers/entity.py +++ /dev/null @@ -1,62 +0,0 @@ -from rich.markup import escape - -from argenta.response.entity import Response -from argenta.app.presentation.renderers import Renderer -from argenta.app.protocols import ( - NonStandardBehaviorHandler, - EmptyCommandHandler, - Printer, - MostSimilarCommandGetter, - DescriptionMessageGenerator, -) -from argenta.command import InputCommand - - -class BehaviorHandlersFabric: - def __init__( - self, - printer: Printer, - renderer: Renderer, - most_similar_command_getter: MostSimilarCommandGetter, - ) -> None: - self._printer = printer - self._renderer = renderer - self._most_similar_command_getter = most_similar_command_getter - - def generate_incorrect_input_syntax_handler(self) -> NonStandardBehaviorHandler[str]: - return lambda raw_command: self._printer( - self._renderer.render_text_for_incorrect_input_syntax_handler( - raw_command=escape(raw_command) - ) - ) - - def generate_repeated_input_flags_handler(self) -> NonStandardBehaviorHandler[str]: - return lambda raw_command: self._printer( - self._renderer.render_text_for_repeated_input_flags_handler( - raw_command=escape(raw_command) - ) - ) - - def generate_empty_input_command_handler(self) -> EmptyCommandHandler: - return lambda: self._printer(self._renderer.render_text_for_empty_input_command_handler()) - - def generate_unknown_command_handler(self) -> NonStandardBehaviorHandler[InputCommand]: - def unknown_command_handler(command: InputCommand) -> None: - command_trigger: str = command.trigger - most_similar_command_trigger: str | None = self._most_similar_command_getter(command_trigger) - self._printer( - self._renderer.render_text_for_unknown_command_handler( - command_trigger=command_trigger, - most_similar_command_trigger=most_similar_command_trigger - ) - ) - return unknown_command_handler - - def generate_exit_command_handler(self, farewell_message: str) -> NonStandardBehaviorHandler[Response]: - return lambda _: self._printer(farewell_message) - - def generate_description_message_generator(self) -> DescriptionMessageGenerator: - return lambda command, description: self._renderer.render_text_for_description_message_generator( - command=command, - description=description - ) diff --git a/src/argenta/app/behavior_handlers/models.py b/src/argenta/app/behavior_handlers/models.py new file mode 100644 index 0000000..7786df6 --- /dev/null +++ b/src/argenta/app/behavior_handlers/models.py @@ -0,0 +1,128 @@ +from rich.markup import escape + +from argenta.response.entity import Response +from argenta.app.presentation.renderers import Renderer +from argenta.app.protocols import ( + NonStandardBehaviorHandler, + EmptyCommandHandler, + Printer, + MostSimilarCommandGetter, + DescriptionMessageGenerator, +) +from argenta.command import InputCommand + + +class BehaviorHandlersFabric: + def __init__( + self, + printer: Printer, + renderer: Renderer, + most_similar_command_getter: MostSimilarCommandGetter, + ) -> None: + self._printer = printer + self._renderer = renderer + self._most_similar_command_getter = most_similar_command_getter + + def generate_incorrect_input_syntax_handler(self) -> NonStandardBehaviorHandler[str]: + return lambda raw_command: self._printer( + self._renderer.render_text_for_incorrect_input_syntax_handler( + raw_command=escape(raw_command) + ) + ) + + def generate_repeated_input_flags_handler(self) -> NonStandardBehaviorHandler[str]: + return lambda raw_command: self._printer( + self._renderer.render_text_for_repeated_input_flags_handler( + raw_command=escape(raw_command) + ) + ) + + def generate_empty_input_command_handler(self) -> EmptyCommandHandler: + return lambda: self._printer(self._renderer.render_text_for_empty_input_command_handler()) + + def generate_unknown_command_handler(self) -> NonStandardBehaviorHandler[InputCommand]: + def unknown_command_handler(command: InputCommand) -> None: + command_trigger: str = command.trigger + most_similar_command_trigger: str | None = self._most_similar_command_getter(command_trigger) + self._printer( + self._renderer.render_text_for_unknown_command_handler( + command_trigger=command_trigger, + most_similar_command_trigger=most_similar_command_trigger + ) + ) + return unknown_command_handler + + def generate_exit_command_handler(self, farewell_message: str) -> NonStandardBehaviorHandler[Response]: + return lambda _: self._printer(farewell_message) + + def generate_description_message_generator(self) -> DescriptionMessageGenerator: + return lambda command, description: self._renderer.render_text_for_description_message_generator( + command=command, + description=description + ) + + +class BehaviorHandlersSettersMixin: + def __init__( + self, + description_message_generator: DescriptionMessageGenerator, + incorrect_input_syntax_handler: NonStandardBehaviorHandler[str], + repeated_input_flags_handler: NonStandardBehaviorHandler[str], + empty_input_command_handler: EmptyCommandHandler, + unknown_command_handler: NonStandardBehaviorHandler[InputCommand], + exit_command_handler: NonStandardBehaviorHandler[Response] + ): + self._description_message_generator: DescriptionMessageGenerator = description_message_generator + self._incorrect_input_syntax_handler: NonStandardBehaviorHandler[str] = incorrect_input_syntax_handler + self._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = repeated_input_flags_handler + self._empty_input_command_handler: EmptyCommandHandler = empty_input_command_handler + self._unknown_command_handler: NonStandardBehaviorHandler[InputCommand] = unknown_command_handler + self._exit_command_handler: NonStandardBehaviorHandler[Response] = exit_command_handler + + def set_description_message_pattern(self, _: DescriptionMessageGenerator, /) -> None: + """ + Public. Sets the output pattern of the available commands + :param _: output pattern of the available commands + :return: None + """ + self._description_message_generator = _ + + def set_incorrect_input_syntax_handler(self, _: NonStandardBehaviorHandler[str], /) -> None: + """ + Public. Sets the handler for incorrect flags when entering a command + :param _: handler for incorrect flags when entering a command + :return: None + """ + self._incorrect_input_syntax_handler = _ + + def set_repeated_input_flags_handler(self, _: NonStandardBehaviorHandler[str], /) -> None: + """ + Public. Sets the handler for repeated flags when entering a command + :param _: handler for repeated flags when entering a command + :return: None + """ + self._repeated_input_flags_handler = _ + + def set_unknown_command_handler(self, _: NonStandardBehaviorHandler[InputCommand], /) -> None: + """ + Public. Sets the handler for unknown commands when entering a command + :param _: handler for unknown commands when entering a command + :return: None + """ + self._unknown_command_handler = _ + + def set_empty_command_handler(self, _: EmptyCommandHandler, /) -> None: + """ + Public. Sets the handler for empty commands when entering a command + :param _: handler for empty commands when entering a command + :return: None + """ + self._empty_input_command_handler = _ + + def set_exit_command_handler(self, _: NonStandardBehaviorHandler[Response], /) -> None: + """ + Public. Sets the handler for exit command when entering a command + :param _: handler for exit command when entering a command + :return: None + """ + self._exit_command_handler = _ diff --git a/src/argenta/app/models.py b/src/argenta/app/models.py index 13d79e8..b251e8c 100644 --- a/src/argenta/app/models.py +++ b/src/argenta/app/models.py @@ -1,20 +1,19 @@ __all__ = ["App"] from typing import Never, TypeAlias +import difflib from rich.console import Console from argenta.app.autocompleter import AutoCompleter -from argenta.app.behavior_handlers.entity import BehaviorHandlersFabric +from argenta.app.behavior_handlers.models import ( + BehaviorHandlersFabric, + BehaviorHandlersSettersMixin, +) from argenta.app.presentation.renderers import PlainRenderer, RichRenderer, Renderer from argenta.app.dividing_line.models import DynamicDividingLine, StaticDividingLine from argenta.app.presentation.viewers import Viewer -from argenta.app.protocols import ( - DescriptionMessageGenerator, - EmptyCommandHandler, - NonStandardBehaviorHandler, - Printer, -) +from argenta.app.protocols import Printer from argenta.app.registered_routers.entity import RegisteredRouters from argenta.command.exceptions import ( InputCommandException, @@ -30,7 +29,7 @@ from argenta.router import Router Matches: TypeAlias = list[str] | list[Never] -class BaseApp: +class BaseApp(BehaviorHandlersSettersMixin): def __init__( self, *, @@ -76,74 +75,20 @@ class BaseApp: self._initial_message: str = self._renderer.render_initial_message(initial_message) self._farewell_message: str = self._renderer.render_farewell_message(farewell_message) - self._description_message_generator: DescriptionMessageGenerator = self._handlers_fabric.generate_description_message_generator() - self._incorrect_input_syntax_handler: NonStandardBehaviorHandler[str] = self._handlers_fabric.generate_incorrect_input_syntax_handler() - self._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = self._handlers_fabric.generate_repeated_input_flags_handler() - self._empty_input_command_handler: EmptyCommandHandler = self._handlers_fabric.generate_empty_input_command_handler() - self._unknown_command_handler: NonStandardBehaviorHandler[InputCommand] = self._handlers_fabric.generate_unknown_command_handler() - self._exit_command_handler: NonStandardBehaviorHandler[Response] = self._handlers_fabric.generate_exit_command_handler(self._farewell_message) - def set_description_message_pattern(self, _: DescriptionMessageGenerator, /) -> None: - """ - Public. Sets the output pattern of the available commands - :param _: output pattern of the available commands - :return: None - """ - self._description_message_generator = _ - - def set_incorrect_input_syntax_handler(self, _: NonStandardBehaviorHandler[str], /) -> None: - """ - Public. Sets the handler for incorrect flags when entering a command - :param _: handler for incorrect flags when entering a command - :return: None - """ - self._incorrect_input_syntax_handler = _ - - def set_repeated_input_flags_handler(self, _: NonStandardBehaviorHandler[str], /) -> None: - """ - Public. Sets the handler for repeated flags when entering a command - :param _: handler for repeated flags when entering a command - :return: None - """ - self._repeated_input_flags_handler = _ - - def set_unknown_command_handler(self, _: NonStandardBehaviorHandler[InputCommand], /) -> None: - """ - Public. Sets the handler for unknown commands when entering a command - :param _: handler for unknown commands when entering a command - :return: None - """ - self._unknown_command_handler = _ - - def set_empty_command_handler(self, _: EmptyCommandHandler, /) -> None: - """ - Public. Sets the handler for empty commands when entering a command - :param _: handler for empty commands when entering a command - :return: None - """ - self._empty_input_command_handler = _ - - def set_exit_command_handler(self, _: NonStandardBehaviorHandler[Response], /) -> None: - """ - Public. Sets the handler for exit command when entering a command - :param _: handler for exit command when entering a command - :return: None - """ - self._exit_command_handler = _ + super().__init__( + description_message_generator = self._handlers_fabric.generate_description_message_generator(), + incorrect_input_syntax_handler = self._handlers_fabric.generate_incorrect_input_syntax_handler(), + repeated_input_flags_handler = self._handlers_fabric.generate_repeated_input_flags_handler(), + empty_input_command_handler = self._handlers_fabric.generate_empty_input_command_handler(), + unknown_command_handler = self._handlers_fabric.generate_unknown_command_handler(), + exit_command_handler = self._handlers_fabric.generate_exit_command_handler(self._farewell_message) + ) def _is_exit_command(self, command: InputCommand) -> bool: - """ - Private. Checks if the given command is an exit command - :param command: command to check - :return: is it an exit command or not as bool - """ - trigger = command.trigger - exit_trigger = self._exit_command.trigger - if trigger.lower() == exit_trigger.lower(): - return True - elif trigger.lower() in [x.lower() for x in self._exit_command.aliases]: - return True - return False + if not self._system_router.command_handlers.get_command_handler_by_trigger(command.trigger.lower()): + return False + return True def _is_unknown_command(self, input_command: InputCommand) -> bool: if not self.registered_routers.get_router_by_trigger(input_command.trigger.lower()): @@ -151,12 +96,6 @@ class BaseApp: return False def _error_handler(self, error: InputCommandException, raw_command: str) -> None: - """ - Private. Handles parsing errors of the entered command - :param error: error being handled - :param raw_command: the raw input command - :return: None - """ if isinstance(error, UnprocessedInputFlagException): self._incorrect_input_syntax_handler(raw_command) elif isinstance(error, RepeatedInputFlagsException): @@ -164,62 +103,33 @@ class BaseApp: else: self._empty_input_command_handler() + def _validate_routers_for_collisions(self) -> None: + seen_names: set[str] = set() + + for router_entity in self.registered_routers: + if not seen_names.isdisjoint(router_entity.triggers): + raise RepeatedTriggerNameException() + + alias_collisions = seen_names.intersection(router_entity.aliases) + if alias_collisions: + raise RepeatedAliasNameException(alias_collisions) + + seen_names.update(router_entity.triggers) + seen_names.update(router_entity.aliases) + + def _most_similar_command(self, unknown_command: str) -> str | None: + all_commands = self.registered_routers.get_triggers() + matches = difflib.get_close_matches(unknown_command, all_commands, n=1) + return matches[0] if matches else None + def _setup_system_router(self) -> None: - """ - Private. Sets up system router - :return: None - """ @self._system_router.command(self._exit_command) def _(response: Response) -> None: self._exit_command_handler(response) self.registered_routers.add_registered_router(self._system_router) - def _validate_routers_for_collisions(self) -> None: - """ - Private. Validates that there are no trigger/alias collisions between routers - :return: None - :raises: RepeatedTriggerNameException or RepeatedAliasNameException if collision detected - """ - - all_triggers: set[str] = set() - all_aliases: set[str] = set() - - for router_entity in self.registered_routers: - union_units: set[str] = all_triggers | all_aliases - trigger_collisions: set[str] = union_units & router_entity.triggers - if trigger_collisions: - raise RepeatedTriggerNameException() - - alias_collisions: set[str] = union_units & router_entity.aliases - if alias_collisions: - raise RepeatedAliasNameException(alias_collisions) - - all_triggers.update(router_entity.triggers) - all_aliases.update(router_entity.aliases) - - def _most_similar_command(self, unknown_command: str) -> str | None: - all_commands = self.registered_routers.get_triggers() - - matches_startswith_unknown_command: Matches = sorted( - cmd for cmd in all_commands if cmd.startswith(unknown_command) - ) - matches_startswith_cmd: Matches = sorted(cmd for cmd in all_commands if unknown_command.startswith(cmd)) - - matches: Matches = matches_startswith_unknown_command or matches_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 _pre_cycle_setup(self) -> None: - """ - Private. Configures various aspects of the application before the start of the cycle - :return: None - """ self._setup_system_router() self._validate_routers_for_collisions() self._autocompleter.initial_setup(self.registered_routers.get_triggers()) @@ -241,10 +151,36 @@ class BaseApp: is_stdout_redirected_by_router=processing_router.is_redirect_stdout_disabled ) + def _run_polling(self) -> None: + self._viewer.view_initial_message(self._initial_message) + self._pre_cycle_setup() + while True: + if self._repeat_command_groups_printing: + self._viewer.view_command_groups_description(self._description_message_generator, self.registered_routers) -AVAILABLE_DIVIDING_LINES: TypeAlias = StaticDividingLine | DynamicDividingLine | None -DEFAULT_PRINTER: Printer = Console().print -DEFAULT_EXIT_COMMAND: Command = Command("q", description="Exit command") + print() # pre-prompt gap + raw_command: str = self._autocompleter.prompt(self._renderer.render_prompt(self._prompt)) + print() # post-prompt gap + + try: + input_command: InputCommand = InputCommand.parse(raw_command=raw_command) + except InputCommandException as error: # noqa F841 + self._viewer.view_framed_text_from_generator( + output_text_generator=lambda: self._error_handler(error, raw_command) + ) + continue + + if self._is_unknown_command(input_command): + self._viewer.view_framed_text_from_generator( + output_text_generator=lambda: self._unknown_command_handler(input_command) + ) + continue + + if self._is_exit_command(input_command): + self._system_router.finds_appropriate_handler(input_command) + return + + self._process_exist_and_valid_command(input_command) class App(BaseApp): @@ -254,13 +190,13 @@ class App(BaseApp): prompt: str = ">>> ", initial_message: str = "Argenta", farewell_message: str = "See you", - exit_command: Command = DEFAULT_EXIT_COMMAND, + exit_command: Command = Command("q", description="Exit command"), system_router_title: str = "System points:", - dividing_line: AVAILABLE_DIVIDING_LINES = None, + dividing_line: StaticDividingLine | DynamicDividingLine | None = None, repeat_command_groups_printing: bool = False, override_system_messages: bool = False, autocompleter: AutoCompleter | None = None, - printer: Printer = DEFAULT_PRINTER, + printer: Printer = Console().print, ) -> None: """ Public. The essence of the application itself. @@ -290,41 +226,6 @@ class App(BaseApp): printer=printer, ) - def run_polling(self) -> None: - """ - Private. Starts the user input processing cycle - :return: None - """ - self._viewer.view_initial_message(self._initial_message) - self._pre_cycle_setup() - while True: - if self._repeat_command_groups_printing: - self._viewer.view_command_groups_description(self._description_message_generator, self.registered_routers) - - print() # pre-prompt gap - raw_command: str = self._autocompleter.prompt(self._renderer.render_prompt(self._prompt)) - print() # post-prompt gap - - try: - input_command: InputCommand = InputCommand.parse(raw_command=raw_command) - except InputCommandException as error: # noqa F841 - self._viewer.view_framed_text_from_generator( - output_text_generator=lambda: self._error_handler(error, raw_command) - ) - continue - - if self._is_exit_command(input_command): - self._system_router.finds_appropriate_handler(input_command) - return - - if self._is_unknown_command(input_command): - self._viewer.view_framed_text_from_generator( - output_text_generator=lambda: self._unknown_command_handler(input_command) - ) - continue - - self._process_exist_and_valid_command(input_command) - def include_router(self, router: Router) -> None: """ Public. Registers the router in the application diff --git a/src/argenta/app/presentation/renderers.py b/src/argenta/app/presentation/renderers.py index 47ff1de..379466d 100644 --- a/src/argenta/app/presentation/renderers.py +++ b/src/argenta/app/presentation/renderers.py @@ -115,7 +115,7 @@ class RichRenderer(Renderer): ) -> str: return ( f"[red]Unknown command:[/red] [blue]{command_trigger}[/blue]" - + (f"[red], most similar:[/red][blue]{most_similar_command_trigger}[/blue]" + + (f"[red], most similar:[/red] [blue]{most_similar_command_trigger}[/blue]" if most_similar_command_trigger else "") ) diff --git a/src/argenta/command/models.py b/src/argenta/command/models.py index 7f56c19..d608e65 100644 --- a/src/argenta/command/models.py +++ b/src/argenta/command/models.py @@ -1,7 +1,7 @@ __all__ = ["Command", "InputCommand"] import shlex -from typing import Literal, Never, Self, cast +from typing import Literal, Never, Self, cast, Iterable from argenta.command.exceptions import ( EmptyInputCommandException, @@ -16,10 +16,6 @@ ParseResult = tuple[str, InputFlags] MIN_FLAG_PREFIX: str = "-" PREFIX_TYPE = Literal["-", "--", "---"] -DEFAULT_WITHOUT_FLAGS: Flags = Flags() -DEFAULT_WITHOUT_ALIASES: set[Never] = set() - -DEFAULT_WITHOUT_INPUT_FLAGS: InputFlags = InputFlags() class Command: @@ -28,8 +24,8 @@ class Command: trigger: str, *, description: str = "Some useful command", - flags: Flag | Flags = DEFAULT_WITHOUT_FLAGS, - aliases: set[str] | set[Never] = DEFAULT_WITHOUT_ALIASES, + flags: Flag | Flags | None = None, + aliases: Iterable[str] | None = None, ): """ Public. The command that can and should be registered in the Router @@ -38,11 +34,16 @@ class Command: :param flags: processed commands :param aliases: string synonyms for the main trigger """ - pretty_flags = flags if isinstance(flags, Flags) else Flags([flags]) + pretty_flags: Flags = ( + flags if isinstance(flags, Flags) + else Flags([flags]) + if flags is not None + else Flags() + ) self.registered_flags: Flags = pretty_flags self.trigger: str = trigger self.description: str = description - self.aliases: set[str] | set[Never] = aliases + self.aliases: Iterable[str] | Iterable[Never] = aliases or set() self._paired_string_entity_flag: dict[str, Flag] = { flag.string_entity: flag for flag in pretty_flags @@ -68,7 +69,7 @@ class InputCommand: self, trigger: str, *, - input_flags: InputFlag | InputFlags = DEFAULT_WITHOUT_INPUT_FLAGS, + input_flags: InputFlag | InputFlags | None = None, ): """ Private. The model of the input command, after parsing @@ -81,6 +82,8 @@ class InputCommand: input_flags if isinstance(input_flags, InputFlags) else InputFlags([input_flags]) + if input_flags is not None + else InputFlags() ) @classmethod diff --git a/src/argenta/orchestrator/entity.py b/src/argenta/orchestrator/entity.py index c6d5193..ad38754 100644 --- a/src/argenta/orchestrator/entity.py +++ b/src/argenta/orchestrator/entity.py @@ -39,4 +39,4 @@ class Orchestrator: ) setup_dishka(app, container, auto_inject=self._auto_inject_handlers) - app.run_polling() + app._run_polling()