better perf

This commit is contained in:
2025-12-08 14:17:31 +03:00
parent cb6549452d
commit 75b1efb259
9 changed files with 190 additions and 88 deletions
+7 -16
View File
@@ -1,19 +1,10 @@
from argenta import App, DataBridge, Response, Router
from argenta.di import FromDishka
from argenta.di.integration import setup_dishka, _auto_inject_handlers
from argenta.di.providers import SystemProvider
from dishka import make_container
from argenta import Command, Response, Router
from argenta.command import InputCommand
container = make_container()
Response.patch_by_container(container)
app = App()
router = Router()
@router.command('command')
def handler(res: Response, data_bridge: FromDishka[DataBridge]):
print(data_bridge)
_auto_inject_handlers(app)
_auto_inject_handlers(app)
@router.command(Command('heLLo'))
def handler(_res: Response) -> None:
print("Hello World!")
router.finds_appropriate_handler(InputCommand('HellO'))
+69 -34
View File
@@ -26,7 +26,6 @@ from argenta.command.exceptions import (
from argenta.command.models import Command, InputCommand
from argenta.response import Response
from argenta.router import Router
from argenta.router.defaults import system_router
Matches: TypeAlias = list[str] | list[Never]
@@ -50,12 +49,12 @@ class BaseApp:
self._prompt: str = prompt
self._print_func: Printer = print_func
self._exit_command: Command = exit_command
self._system_router_title: str = system_router_title
self._dividing_line: StaticDividingLine | DynamicDividingLine = dividing_line
self._ignore_command_register: bool = ignore_command_register
self._repeat_command_groups_printing_description: bool = repeat_command_groups_printing
self._override_system_messages: bool = override_system_messages
self._autocompleter: AutoCompleter = autocompleter
self.system_router: Router = Router(title=system_router_title)
self._farewell_message: str = farewell_message
self._initial_message: str = initial_message
@@ -75,18 +74,20 @@ class BaseApp:
else self._matching_default_triggers_with_routers
)
self._incorrect_input_syntax_handler: NonStandardBehaviorHandler[str] = lambda _: print_func(
f"Incorrect flag syntax: {_}"
self._incorrect_input_syntax_handler: NonStandardBehaviorHandler[str] = (
lambda _: print_func(f"Incorrect flag syntax: {_}")
)
self._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = lambda _: print_func(
f"Repeated input flags: {_}"
self._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = (
lambda _: print_func(f"Repeated input flags: {_}")
)
self._empty_input_command_handler: EmptyCommandHandler = lambda: print_func("Empty input command")
self._unknown_command_handler: NonStandardBehaviorHandler[InputCommand] = lambda _: print_func(
f"Unknown command: {_.trigger}"
self._empty_input_command_handler: EmptyCommandHandler = lambda: print_func(
"Empty input command"
)
self._exit_command_handler: NonStandardBehaviorHandler[Response] = lambda _: print_func(
self._farewell_message
self._unknown_command_handler: NonStandardBehaviorHandler[InputCommand] = (
lambda _: print_func(f"Unknown command: {_.trigger}")
)
self._exit_command_handler: NonStandardBehaviorHandler[Response] = (
lambda _: print_func(self._farewell_message)
)
def set_description_message_pattern(self, _: DescriptionMessageGenerator, /) -> None:
@@ -97,7 +98,9 @@ class BaseApp:
"""
self._description_message_gen = _
def set_incorrect_input_syntax_handler(self, _: NonStandardBehaviorHandler[str], /) -> None:
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
@@ -105,7 +108,9 @@ class BaseApp:
"""
self._incorrect_input_syntax_handler = _
def set_repeated_input_flags_handler(self, _: NonStandardBehaviorHandler[str], /) -> None:
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
@@ -113,7 +118,9 @@ class BaseApp:
"""
self._repeated_input_flags_handler = _
def set_unknown_command_handler(self, _: NonStandardBehaviorHandler[InputCommand], /) -> None:
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
@@ -129,7 +136,9 @@ class BaseApp:
"""
self._empty_input_command_handler = _
def set_exit_command_handler(self, _: NonStandardBehaviorHandler[Response], /) -> None:
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
@@ -164,7 +173,11 @@ class BaseApp:
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
max_length_line
if 10 <= max_length_line <= 80
else 80
if max_length_line > 80
else 10
)
self._print_func(
@@ -181,11 +194,15 @@ class BaseApp:
elif isinstance(self._dividing_line, StaticDividingLine): # pyright: ignore[reportUnnecessaryIsInstance]
self._print_func(
self._dividing_line.get_full_static_line(is_override=self._override_system_messages)
self._dividing_line.get_full_static_line(
is_override=self._override_system_messages
)
)
print(text.strip("\n"))
self._print_func(
self._dividing_line.get_full_static_line(is_override=self._override_system_messages)
self._dividing_line.get_full_static_line(
is_override=self._override_system_messages
)
)
else:
@@ -219,10 +236,14 @@ class BaseApp:
"""
input_command_trigger = command.trigger
if self._ignore_command_register:
if input_command_trigger.lower() in list(self._current_matching_triggers_with_routers.keys()):
if input_command_trigger.lower() in list(
self._current_matching_triggers_with_routers.keys()
):
return False
else:
if input_command_trigger in list(self._current_matching_triggers_with_routers.keys()):
if input_command_trigger in list(
self._current_matching_triggers_with_routers.keys()
):
return False
return True
@@ -245,14 +266,13 @@ class BaseApp:
Private. Sets up system router
:return: None
"""
system_router.title = self._system_router_title
@system_router.command(self._exit_command)
@self.system_router.command(self._exit_command)
def _(response: Response) -> None:
self._exit_command_handler(response)
system_router.command_register_ignore = self._ignore_command_register
self.registered_routers.add_registered_router(system_router)
self.system_router.command_register_ignore = self._ignore_command_register
self.registered_routers.add_registered_router(self.system_router)
def _most_similar_command(self, unknown_command: str) -> str | None:
all_commands = list(self._current_matching_triggers_with_routers.keys())
@@ -279,7 +299,9 @@ class BaseApp:
:return: None
"""
self._prompt = f"[italic dim bold]{self._prompt}"
self._initial_message = "\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n"
self._initial_message = (
"\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n"
)
self._farewell_message = (
"[bold red]\n\n"
+ str(text2art(self._farewell_message, font="chanky")) # pyright: ignore[reportUnknownArgumentType]
@@ -297,14 +319,20 @@ class BaseApp:
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._empty_input_command_handler = lambda: self._print_func(
"[red bold]Empty input command"
)
def unknown_command_handler(command: InputCommand) -> None:
cmd_trg: str = command.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]"
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 ""
("[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)
@@ -324,9 +352,13 @@ class BaseApp:
for trigger in combined:
self._matching_default_triggers_with_routers[trigger] = router_entity
self._matching_lower_triggers_with_routers[trigger.lower()] = router_entity
self._matching_lower_triggers_with_routers[trigger.lower()] = (
router_entity
)
self._autocompleter.initial_setup(list(self._current_matching_triggers_with_routers.keys()))
self._autocompleter.initial_setup(
list(self._current_matching_triggers_with_routers.keys())
)
if not self._override_system_messages:
self._setup_default_view()
@@ -339,9 +371,11 @@ class BaseApp:
print("\n")
if not self._repeat_command_groups_printing_description:
self._print_command_group_description()
def _process_exist_and_valid_command(self, input_command: InputCommand) -> None:
processing_router = self._current_matching_triggers_with_routers[input_command.trigger.lower()]
processing_router = self._current_matching_triggers_with_routers[
input_command.trigger.lower()
]
if processing_router.disable_redirect_stdout:
dividing_line_unit_part: str = self._dividing_line.get_unit_part()
@@ -439,9 +473,10 @@ class App(BaseApp):
continue
if self._is_exit_command(input_command):
system_router.finds_appropriate_handler(input_command)
self.system_router.finds_appropriate_handler(input_command)
self._autocompleter.exit_setup(
list(self._current_matching_triggers_with_routers.keys()), self._ignore_command_register
list(self._current_matching_triggers_with_routers.keys()),
self._ignore_command_register,
)
return
+2 -1
View File
@@ -1,6 +1,7 @@
__all__ = ["ResponseStatus"]
from enum import Enum
from typing import Self
class ResponseStatus(Enum):
@@ -10,7 +11,7 @@ class ResponseStatus(Enum):
UNDEFINED_AND_INVALID_FLAGS = "UNDEFINED_AND_INVALID_FLAGS"
@classmethod
def from_flags(cls, *, has_invalid_value_flags: bool, has_undefined_flags: bool) -> "ResponseStatus":
def from_flags(cls, *, has_invalid_value_flags: bool, has_undefined_flags: bool) -> Self:
key = (has_invalid_value_flags, has_undefined_flags)
status_map: dict[tuple[bool, bool], ResponseStatus] = {
(True, True): cls.UNDEFINED_AND_INVALID_FLAGS,
+15 -4
View File
@@ -7,14 +7,17 @@ from argenta.command import Command
from argenta.response import Response
HandlerFunc = Callable[..., None]
class CommandHandler:
def __init__(self, handler_as_func: Callable[..., None], handled_command: Command):
def __init__(self, handler_as_func: HandlerFunc, handled_command: Command):
"""
Private. Entity of the model linking the handler and the command being processed
:param handler: the handler being called
:param handled_command: the command being processed
"""
self.handler_as_func: Callable[..., None] = handler_as_func
self.handler_as_func: HandlerFunc = handler_as_func
self.handled_command: Command = handled_command
def handling(self, response: Response) -> None:
@@ -27,12 +30,13 @@ class CommandHandler:
class CommandHandlers:
def __init__(self, command_handlers: list[CommandHandler] | None = None):
def __init__(self, command_handlers: tuple[CommandHandler] = tuple()):
"""
Private. The model that unites all CommandHandler of the routers
:param command_handlers: list of CommandHandlers for register
"""
self.command_handlers: list[CommandHandler] = command_handlers if command_handlers else []
self.command_handlers: list[CommandHandler] = list(command_handlers) if command_handlers else []
self.paired_command_handler_trigger: dict[str, CommandHandler] = {x.handled_command.trigger: x for x in command_handlers}
def add_handler(self, command_handler: CommandHandler) -> None:
"""
@@ -41,6 +45,13 @@ class CommandHandlers:
:return: None
"""
self.command_handlers.append(command_handler)
self.paired_command_handler_trigger[command_handler.handled_command.trigger.lower()] = command_handler
for alias in command_handler.handled_command.aliases:
self.paired_command_handler_trigger[alias.lower()] = command_handler
def get_command_handler_by_trigger(self, trigger: str):
print(self.paired_command_handler_trigger)
return self.paired_command_handler_trigger.get(trigger)
def __iter__(self) -> Iterator[CommandHandler]:
return iter(self.command_handlers)
-5
View File
@@ -1,5 +0,0 @@
__all__ = ["system_router"]
from argenta.router import Router
system_router = Router(title="System points:")
+25 -15
View File
@@ -11,7 +11,9 @@ from argenta.command.flag.flags import Flags, InputFlags
from argenta.response import Response, ResponseStatus
from argenta.router.command_handler.entity import CommandHandler, CommandHandlers
from argenta.router.exceptions import (
RepeatedAliasNameException,
RepeatedFlagNameException,
RepeatedTriggerNameException,
RequiredArgumentNotPassedException,
TriggerContainSpacesException,
)
@@ -57,13 +59,8 @@ class Router:
redefined_command = command
self._validate_command(redefined_command)
if overlapping := (self.aliases | self.triggers) & redefined_command.aliases:
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{overlapping}[/b blue]")
self._update_routing_keys(redefined_command)
self.aliases.update(redefined_command.aliases)
self.triggers.add(redefined_command.trigger)
def decorator(func: HandlerFunc) -> HandlerFunc:
_validate_func_args(func)
self.command_handlers.add_handler(CommandHandler(func, redefined_command))
@@ -80,10 +77,22 @@ class Router:
command_name: str = command.trigger
if command_name.find(" ") != -1:
raise TriggerContainSpacesException()
if command_name.lower() in self.triggers:
raise RepeatedTriggerNameException()
if overlapping := (self.aliases | self.triggers) & set(map(lambda x: x.lower(), command.aliases)):
raise RepeatedAliasNameException(overlapping)
flags: Flags = command.registered_flags
flags_name: list[str] = [flag.string_entity.lower() for flag in flags]
if len(set(flags_name)) < len(flags_name):
raise RepeatedFlagNameException()
def _update_routing_keys(self, registered_command: Command):
redefined_command_aliases_in_lower = set(map(lambda x: x.lower(), registered_command.aliases))
self.aliases.update(redefined_command_aliases_in_lower)
self.triggers.add(registered_command.trigger.lower())
def finds_appropriate_handler(self, input_command: InputCommand) -> None:
"""
@@ -91,15 +100,15 @@ class Router:
:param input_command: input command as InputCommand
:return: None
"""
input_command_name: str = input_command.trigger
input_command_name: str = input_command.trigger.lower()
input_command_flags: InputFlags = input_command.input_flags
for command_handler in self.command_handlers:
handle_command = command_handler.handled_command
if input_command_name.lower() == handle_command.trigger.lower():
self.process_input_command(input_command_flags, command_handler)
if input_command_name.lower() in handle_command.aliases:
self.process_input_command(input_command_flags, command_handler)
command_handler = self.command_handlers.get_command_handler_by_trigger(input_command_name)
if not command_handler:
raise RuntimeError(f"Handler for '{input_command.trigger}' command not found!")
else:
self.process_input_command(input_command_flags, command_handler)
def process_input_command(self, input_command_flags: InputFlags, command_handler: CommandHandler) -> None:
"""
@@ -147,13 +156,14 @@ def _structuring_input_flags(handled_command: Command, input_flags: InputFlags)
undefined_flags = True
status = ResponseStatus.from_flags(
has_invalid_value_flags=invalid_value_flags, has_undefined_flags=undefined_flags
has_invalid_value_flags=invalid_value_flags,
has_undefined_flags=undefined_flags
)
return Response(status=status, input_flags=input_flags)
def _validate_func_args(func: Callable[..., None]) -> None:
def _validate_func_args(func: HandlerFunc) -> None:
"""
Private. Validates the arguments of the handler
:param func: entity of the handler func
+24
View File
@@ -11,7 +11,31 @@ class RepeatedFlagNameException(Exception):
@override
def __str__(self) -> str:
return "Repeated registered flag names in register command"
class RepeatedTriggerNameException(Exception):
"""
Private. Raised when a repeated trigger name is registered
"""
@override
def __str__(self) -> str:
return "Repeated trigger name in registered commands"
class RepeatedAliasNameException(Exception):
"""
Private. Raised when a repeated alias name is registered
"""
@override
def __init__(self, repeated_aliases: set[str]) -> None:
self.repeated_aliases = repeated_aliases
super().__init__()
@override
def __str__(self) -> str:
return f"Repeated aliases names: {self.repeated_aliases}"
class RequiredArgumentNotPassedException(Exception):
"""
+6 -12
View File
@@ -1,3 +1,4 @@
from argenta.router.exceptions import RepeatedAliasNameException
import pytest
from pytest import CaptureFixture
@@ -207,24 +208,17 @@ def test_include_routers_registers_multiple_routers() -> None:
assert app.registered_routers.registered_routers == [router, router2]
def test_overlapping_aliases_prints_warning(capsys: CaptureFixture[str]) -> None:
app = App(override_system_messages=True)
def test_overlapping_aliases_raises_exception() -> None:
router = Router()
@router.command(Command('test', aliases={'alias'}))
def handler(_res: Response) -> None:
pass
@router.command(Command('test2', aliases={'alias'}))
def handler2(_res: Response) -> None:
pass
app.include_routers(router)
app._pre_cycle_setup()
captured = capsys.readouterr()
assert "Overlapping" in captured.out
with pytest.raises(RepeatedAliasNameException):
@router.command(Command('test2', aliases={'alias'}))
def handler2(_res: Response) -> None:
pass
# ============================================================================
+42 -1
View File
@@ -12,6 +12,7 @@ from argenta.router import Router
from argenta.router.entity import _structuring_input_flags, _validate_func_args # pyright: ignore[reportPrivateUsage]
from argenta.router.exceptions import (
RepeatedFlagNameException,
RepeatedTriggerNameException,
RequiredArgumentNotPassedException,
TriggerContainSpacesException,
)
@@ -26,7 +27,20 @@ def test_validate_command_raises_error_for_trigger_with_spaces() -> None:
router = Router()
with pytest.raises(TriggerContainSpacesException):
router._validate_command(Command(trigger='command with spaces'))
def test_validate_command_raises_error_for_same_trigger() -> None:
router = Router()
@router.command('comm')
def handler(res: Response):
pass
with pytest.raises(RepeatedTriggerNameException):
@router.command('comm')
def handler2(res: Response):
pass
def test_validate_command_raises_error_for_repeated_flag_names() -> None:
router = Router()
@@ -192,6 +206,33 @@ def test_finds_appropriate_handler_executes_handler_by_alias(capsys: CaptureFixt
output = capsys.readouterr()
assert "Hello World!" in output.out
def test_finds_appropriate_handler_executes_handler_by_alias_with_differrent_register(capsys: CaptureFixture[str]) -> None:
router = Router()
@router.command(Command('hello', aliases={'hI'}))
def handler(_res: Response) -> None:
print("Hello World!")
router.finds_appropriate_handler(InputCommand('HI'))
output = capsys.readouterr()
assert "Hello World!" in output.out
def test_finds_appropriate_handler_executes_handler_by_trigger_with_differrent_register(capsys: CaptureFixture[str]) -> None:
router = Router()
@router.command(Command('heLLo'))
def handler(_res: Response) -> None:
print("Hello World!")
router.finds_appropriate_handler(InputCommand('HellO'))
output = capsys.readouterr()
assert "Hello World!" in output.out
def test_finds_appropriate_handler_executes_handler_with_flags_by_alias(capsys: CaptureFixture[str]) -> None: