This commit is contained in:
2026-02-01 01:18:27 +03:00
parent 24aa75eb37
commit f859451069
7 changed files with 214 additions and 244 deletions
+2 -2
View File
@@ -1,9 +1,9 @@
from argenta import App, Orchestrator from argenta import App, Orchestrator, Command
from argenta.app import DynamicDividingLine from argenta.app import DynamicDividingLine
from .handlers import router 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() orchestrator = Orchestrator()
@@ -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
)
+128
View File
@@ -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 = _
+68 -167
View File
@@ -1,20 +1,19 @@
__all__ = ["App"] __all__ = ["App"]
from typing import Never, TypeAlias from typing import Never, TypeAlias
import difflib
from rich.console import Console from rich.console import Console
from argenta.app.autocompleter import AutoCompleter 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.presentation.renderers import PlainRenderer, RichRenderer, Renderer
from argenta.app.dividing_line.models import DynamicDividingLine, StaticDividingLine from argenta.app.dividing_line.models import DynamicDividingLine, StaticDividingLine
from argenta.app.presentation.viewers import Viewer from argenta.app.presentation.viewers import Viewer
from argenta.app.protocols import ( from argenta.app.protocols import Printer
DescriptionMessageGenerator,
EmptyCommandHandler,
NonStandardBehaviorHandler,
Printer,
)
from argenta.app.registered_routers.entity import RegisteredRouters from argenta.app.registered_routers.entity import RegisteredRouters
from argenta.command.exceptions import ( from argenta.command.exceptions import (
InputCommandException, InputCommandException,
@@ -30,7 +29,7 @@ from argenta.router import Router
Matches: TypeAlias = list[str] | list[Never] Matches: TypeAlias = list[str] | list[Never]
class BaseApp: class BaseApp(BehaviorHandlersSettersMixin):
def __init__( def __init__(
self, self,
*, *,
@@ -76,74 +75,20 @@ class BaseApp:
self._initial_message: str = self._renderer.render_initial_message(initial_message) self._initial_message: str = self._renderer.render_initial_message(initial_message)
self._farewell_message: str = self._renderer.render_farewell_message(farewell_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: super().__init__(
""" description_message_generator = self._handlers_fabric.generate_description_message_generator(),
Public. Sets the output pattern of the available commands incorrect_input_syntax_handler = self._handlers_fabric.generate_incorrect_input_syntax_handler(),
:param _: output pattern of the available commands repeated_input_flags_handler = self._handlers_fabric.generate_repeated_input_flags_handler(),
:return: None empty_input_command_handler = self._handlers_fabric.generate_empty_input_command_handler(),
""" unknown_command_handler = self._handlers_fabric.generate_unknown_command_handler(),
self._description_message_generator = _ exit_command_handler = self._handlers_fabric.generate_exit_command_handler(self._farewell_message)
)
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 = _
def _is_exit_command(self, command: InputCommand) -> bool: def _is_exit_command(self, command: InputCommand) -> bool:
""" if not self._system_router.command_handlers.get_command_handler_by_trigger(command.trigger.lower()):
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 return False
return True
def _is_unknown_command(self, input_command: InputCommand) -> bool: def _is_unknown_command(self, input_command: InputCommand) -> bool:
if not self.registered_routers.get_router_by_trigger(input_command.trigger.lower()): if not self.registered_routers.get_router_by_trigger(input_command.trigger.lower()):
@@ -151,12 +96,6 @@ class BaseApp:
return False return False
def _error_handler(self, error: InputCommandException, raw_command: str) -> None: 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): if isinstance(error, UnprocessedInputFlagException):
self._incorrect_input_syntax_handler(raw_command) self._incorrect_input_syntax_handler(raw_command)
elif isinstance(error, RepeatedInputFlagsException): elif isinstance(error, RepeatedInputFlagsException):
@@ -164,62 +103,33 @@ class BaseApp:
else: else:
self._empty_input_command_handler() 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: def _setup_system_router(self) -> None:
"""
Private. Sets up system router
:return: None
"""
@self._system_router.command(self._exit_command) @self._system_router.command(self._exit_command)
def _(response: Response) -> None: def _(response: Response) -> None:
self._exit_command_handler(response) self._exit_command_handler(response)
self.registered_routers.add_registered_router(self._system_router) 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: 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._setup_system_router()
self._validate_routers_for_collisions() self._validate_routers_for_collisions()
self._autocompleter.initial_setup(self.registered_routers.get_triggers()) 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 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 print() # pre-prompt gap
DEFAULT_PRINTER: Printer = Console().print raw_command: str = self._autocompleter.prompt(self._renderer.render_prompt(self._prompt))
DEFAULT_EXIT_COMMAND: Command = Command("q", description="Exit command") 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): class App(BaseApp):
@@ -254,13 +190,13 @@ class App(BaseApp):
prompt: str = ">>> ", prompt: str = ">>> ",
initial_message: str = "Argenta", initial_message: str = "Argenta",
farewell_message: str = "See you", 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:", system_router_title: str = "System points:",
dividing_line: AVAILABLE_DIVIDING_LINES = None, dividing_line: StaticDividingLine | DynamicDividingLine | None = None,
repeat_command_groups_printing: bool = False, repeat_command_groups_printing: bool = False,
override_system_messages: bool = False, override_system_messages: bool = False,
autocompleter: AutoCompleter | None = None, autocompleter: AutoCompleter | None = None,
printer: Printer = DEFAULT_PRINTER, printer: Printer = Console().print,
) -> None: ) -> None:
""" """
Public. The essence of the application itself. Public. The essence of the application itself.
@@ -290,41 +226,6 @@ class App(BaseApp):
printer=printer, 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: def include_router(self, router: Router) -> None:
""" """
Public. Registers the router in the application Public. Registers the router in the application
+1 -1
View File
@@ -115,7 +115,7 @@ class RichRenderer(Renderer):
) -> str: ) -> str:
return ( return (
f"[red]Unknown command:[/red] [blue]{command_trigger}[/blue]" 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 "") if most_similar_command_trigger else "")
) )
+13 -10
View File
@@ -1,7 +1,7 @@
__all__ = ["Command", "InputCommand"] __all__ = ["Command", "InputCommand"]
import shlex import shlex
from typing import Literal, Never, Self, cast from typing import Literal, Never, Self, cast, Iterable
from argenta.command.exceptions import ( from argenta.command.exceptions import (
EmptyInputCommandException, EmptyInputCommandException,
@@ -16,10 +16,6 @@ ParseResult = tuple[str, InputFlags]
MIN_FLAG_PREFIX: str = "-" MIN_FLAG_PREFIX: str = "-"
PREFIX_TYPE = Literal["-", "--", "---"] PREFIX_TYPE = Literal["-", "--", "---"]
DEFAULT_WITHOUT_FLAGS: Flags = Flags()
DEFAULT_WITHOUT_ALIASES: set[Never] = set()
DEFAULT_WITHOUT_INPUT_FLAGS: InputFlags = InputFlags()
class Command: class Command:
@@ -28,8 +24,8 @@ class Command:
trigger: str, trigger: str,
*, *,
description: str = "Some useful command", description: str = "Some useful command",
flags: Flag | Flags = DEFAULT_WITHOUT_FLAGS, flags: Flag | Flags | None = None,
aliases: set[str] | set[Never] = DEFAULT_WITHOUT_ALIASES, aliases: Iterable[str] | None = None,
): ):
""" """
Public. The command that can and should be registered in the Router Public. The command that can and should be registered in the Router
@@ -38,11 +34,16 @@ class Command:
:param flags: processed commands :param flags: processed commands
:param aliases: string synonyms for the main trigger :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.registered_flags: Flags = pretty_flags
self.trigger: str = trigger self.trigger: str = trigger
self.description: str = description 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] = { self._paired_string_entity_flag: dict[str, Flag] = {
flag.string_entity: flag for flag in pretty_flags flag.string_entity: flag for flag in pretty_flags
@@ -68,7 +69,7 @@ class InputCommand:
self, self,
trigger: str, 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 Private. The model of the input command, after parsing
@@ -81,6 +82,8 @@ class InputCommand:
input_flags input_flags
if isinstance(input_flags, InputFlags) if isinstance(input_flags, InputFlags)
else InputFlags([input_flags]) else InputFlags([input_flags])
if input_flags is not None
else InputFlags()
) )
@classmethod @classmethod
+1 -1
View File
@@ -39,4 +39,4 @@ class Orchestrator:
) )
setup_dishka(app, container, auto_inject=self._auto_inject_handlers) setup_dishka(app, container, auto_inject=self._auto_inject_handlers)
app.run_polling() app._run_polling()