This commit is contained in:
2026-02-01 00:26:25 +03:00
parent 5f6b3368e1
commit 24aa75eb37
11 changed files with 187 additions and 186 deletions
+1 -1
View File
@@ -19,7 +19,7 @@ def test_input_incorrect_command(capsys: CaptureFixture[str]):
def test(response: Response) -> None: def test(response: Response) -> None:
print('test command') print('test command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
app.set_unknown_command_handler( app.set_unknown_command_handler(
lambda command: print(f'Unknown command: {command.trigger}') lambda command: print(f'Unknown command: {command.trigger}')
+2 -1
View File
@@ -1,8 +1,9 @@
from argenta import App, Orchestrator from argenta import App, Orchestrator
from argenta.app import DynamicDividingLine
from .handlers import router from .handlers import router
app = App(initial_message="metrics", override_system_messages=False) app = App(initial_message="metrics", dividing_line=DynamicDividingLine('~'), override_system_messages=True)
orchestrator = Orchestrator() orchestrator = Orchestrator()
+3
View File
@@ -0,0 +1,3 @@
from rich.console import Console
Console().print('[red]hi[/red]')
+2 -2
View File
@@ -1,6 +1,6 @@
from rich.markup import escape from rich.markup import escape
from argenta import Response from argenta.response.entity import Response
from argenta.app.presentation.renderers import Renderer from argenta.app.presentation.renderers import Renderer
from argenta.app.protocols import ( from argenta.app.protocols import (
NonStandardBehaviorHandler, NonStandardBehaviorHandler,
@@ -53,7 +53,7 @@ class BehaviorHandlersFabric:
return unknown_command_handler return unknown_command_handler
def generate_exit_command_handler(self, farewell_message: str) -> NonStandardBehaviorHandler[Response]: def generate_exit_command_handler(self, farewell_message: str) -> NonStandardBehaviorHandler[Response]:
return lambda _: self._printer(self._renderer.render_farewell_message(farewell_message)) return lambda _: self._printer(farewell_message)
def generate_description_message_generator(self) -> DescriptionMessageGenerator: def generate_description_message_generator(self) -> DescriptionMessageGenerator:
return lambda command, description: self._renderer.render_text_for_description_message_generator( return lambda command, description: self._renderer.render_text_for_description_message_generator(
+4 -4
View File
@@ -41,9 +41,9 @@ class StaticDividingLine(BaseDividingLine):
:return: full line of dividing line as str :return: full line of dividing line as str
""" """
if is_override: if is_override:
return f"\n{self.length * self.get_unit_part()}\n" return self.length * self.get_unit_part()
else: else:
return f"\n[dim]{self.length * self.get_unit_part()}[/dim]\n" return f"[dim]{self.length * self.get_unit_part()}[/dim]"
class DynamicDividingLine(BaseDividingLine): class DynamicDividingLine(BaseDividingLine):
@@ -63,6 +63,6 @@ class DynamicDividingLine(BaseDividingLine):
:return: full line of dividing line as str :return: full line of dividing line as str
""" """
if is_override: if is_override:
return f"\n{length * self.get_unit_part()}\n" return length * self.get_unit_part()
else: else:
return f"\n[dim]{self.get_unit_part() * length}[/dim]\n" return f"[dim]{self.get_unit_part() * length}[/dim]"
+28 -100
View File
@@ -1,9 +1,6 @@
__all__ = ["App"] __all__ = ["App"]
import io from typing import Never, TypeAlias
import re
from contextlib import redirect_stdout
from typing import Callable, Never, TypeAlias
from rich.console import Console from rich.console import Console
@@ -31,7 +28,6 @@ from argenta.router import Router
Matches: TypeAlias = list[str] | list[Never] Matches: TypeAlias = list[str] | list[Never]
_ANSI_ESCAPE_RE: re.Pattern[str] = re.compile(r"\u001b\[[0-9;]*m")
class BaseApp: class BaseApp:
@@ -47,10 +43,10 @@ class BaseApp:
repeat_command_groups_printing: bool, repeat_command_groups_printing: bool,
override_system_messages: bool, override_system_messages: bool,
autocompleter: AutoCompleter, autocompleter: AutoCompleter,
print_func: Printer, printer: Printer,
) -> None: ) -> None:
self._prompt: str = prompt self._prompt: str = prompt
self._print_func: Printer = print_func self._printer: Printer = printer
self._exit_command: Command = exit_command self._exit_command: Command = exit_command
self._dividing_line: StaticDividingLine | DynamicDividingLine | None = dividing_line self._dividing_line: StaticDividingLine | DynamicDividingLine | None = dividing_line
self._repeat_command_groups_printing: bool = repeat_command_groups_printing self._repeat_command_groups_printing: bool = repeat_command_groups_printing
@@ -58,7 +54,6 @@ class BaseApp:
self._autocompleter: AutoCompleter = autocompleter self._autocompleter: AutoCompleter = autocompleter
self._system_router: Router = Router(title=system_router_title) self._system_router: Router = Router(title=system_router_title)
self._stdout_buffer: io.StringIO = io.StringIO()
self.registered_routers: RegisteredRouters = RegisteredRouters() self.registered_routers: RegisteredRouters = RegisteredRouters()
self._messages_on_startup: list[str] = [] self._messages_on_startup: list[str] = []
@@ -67,11 +62,16 @@ class BaseApp:
else: else:
self._renderer: Renderer = RichRenderer() self._renderer: Renderer = RichRenderer()
self._viewer: Viewer = Viewer(self._print_func, self._renderer) self._viewer: Viewer = Viewer(
printer=self._printer,
renderer=self._renderer,
dividing_line=self._dividing_line,
override_system_messages=self._override_system_messages,
)
self._handlers_fabric: BehaviorHandlersFabric = BehaviorHandlersFabric( self._handlers_fabric: BehaviorHandlersFabric = BehaviorHandlersFabric(
self._print_func, printer=self._printer,
self._renderer, renderer=self._renderer,
self._most_similar_command most_similar_command_getter=self._most_similar_command
) )
self._initial_message: str = self._renderer.render_initial_message(initial_message) self._initial_message: str = self._renderer.render_initial_message(initial_message)
@@ -131,34 +131,6 @@ class BaseApp:
""" """
self._exit_command_handler = _ self._exit_command_handler = _
def _print_static_framed_text(self, text: str) -> None:
"""
Private. Outputs text by framing it in a static or dynamic split strip
:param text: framed text
:return: None
"""
match self._dividing_line:
case StaticDividingLine() as dividing_line:
self._print_func(dividing_line.get_full_static_line(is_override=self._override_system_messages))
print(text.strip("\n"))
self._print_func(dividing_line.get_full_static_line(is_override=self._override_system_messages))
case DynamicDividingLine() as dividing_line:
self._print_func(
StaticDividingLine(dividing_line.get_unit_part()).get_full_static_line(
is_override=self._override_system_messages
)
)
print(text.strip("\n"))
self._print_func(
StaticDividingLine(dividing_line.get_unit_part()).get_full_static_line(
is_override=self._override_system_messages
)
)
case None:
print("\n" + text.strip("\n") + "\n")
case _:
raise NotImplementedError(f"Dividing line with type {self._dividing_line} is not implemented")
def _is_exit_command(self, command: InputCommand) -> bool: def _is_exit_command(self, command: InputCommand) -> bool:
""" """
Private. Checks if the given command is an exit command Private. Checks if the given command is an exit command
@@ -178,18 +150,6 @@ class BaseApp:
return True return True
return False return False
def _capture_stdout(self, func: Callable[[], None]) -> str:
"""
Private. Captures stdout from a function call using a reusable buffer
:param func: function to execute with captured stdout
:return: captured stdout as string
"""
self._stdout_buffer.seek(0)
self._stdout_buffer.truncate(0)
with redirect_stdout(self._stdout_buffer):
func()
return self._stdout_buffer.getvalue()
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 Private. Handles parsing errors of the entered command
@@ -276,46 +236,14 @@ class BaseApp:
if not processing_router: if not processing_router:
raise RuntimeError(f"Router for '{input_command.trigger}' not found. Panic!") raise RuntimeError(f"Router for '{input_command.trigger}' not found. Panic!")
match (self._dividing_line, processing_router.disable_redirect_stdout): self._viewer.view_framed_text_from_generator(
case (None, bool()): output_text_generator=lambda: processing_router.finds_appropriate_handler(input_command),
processing_router.finds_appropriate_handler(input_command) is_stdout_redirected_by_router=processing_router.is_redirect_stdout_disabled
case (DynamicDividingLine(), False): )
stdout_result = self._capture_stdout(lambda: processing_router.finds_appropriate_handler(input_command))
clear_text = _ANSI_ESCAPE_RE.sub("", stdout_result)
max_length_line = max([len(line) for line in clear_text.split("\n")])
max_length_line = (
max_length_line if 10 <= max_length_line <= 100 else 100 if max_length_line > 100 else 10
)
self._print_func(
self._dividing_line.get_full_dynamic_line(
length=max_length_line, is_override=self._override_system_messages
)
)
print(clear_text.strip("\n"))
self._print_func(
self._dividing_line.get_full_dynamic_line(
length=max_length_line, is_override=self._override_system_messages
)
)
case (StaticDividingLine() as dividing_line, bool()) | (DynamicDividingLine() as dividing_line, True):
self._print_func(
StaticDividingLine(dividing_line.get_unit_part()).get_full_static_line(
is_override=self._override_system_messages
)
)
processing_router.finds_appropriate_handler(input_command)
self._print_func(
StaticDividingLine(dividing_line.get_unit_part()).get_full_static_line(
is_override=self._override_system_messages
)
)
case _:
raise NotImplementedError(f"Dividing line with type {self._dividing_line} is not implemented")
AVAILABLE_DIVIDING_LINES: TypeAlias = StaticDividingLine | DynamicDividingLine AVAILABLE_DIVIDING_LINES: TypeAlias = StaticDividingLine | DynamicDividingLine | None
DEFAULT_PRINT_FUNC: Printer = Console().print DEFAULT_PRINTER: Printer = Console().print
DEFAULT_EXIT_COMMAND: Command = Command("q", description="Exit command") DEFAULT_EXIT_COMMAND: Command = Command("q", description="Exit command")
@@ -328,11 +256,11 @@ class App(BaseApp):
farewell_message: str = "See you", farewell_message: str = "See you",
exit_command: Command = DEFAULT_EXIT_COMMAND, exit_command: Command = DEFAULT_EXIT_COMMAND,
system_router_title: str = "System points:", system_router_title: str = "System points:",
dividing_line: AVAILABLE_DIVIDING_LINES | None = None, dividing_line: AVAILABLE_DIVIDING_LINES = 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,
print_func: Printer = DEFAULT_PRINT_FUNC, printer: Printer = DEFAULT_PRINTER,
) -> None: ) -> None:
""" """
Public. The essence of the application itself. Public. The essence of the application itself.
@@ -346,7 +274,7 @@ class App(BaseApp):
:param repeat_command_groups_printing: whether to repeat the available commands and their description :param repeat_command_groups_printing: whether to repeat the available commands and their description
:param override_system_messages: whether to redefine the default formatting of system messages :param override_system_messages: whether to redefine the default formatting of system messages
:param autocompleter: the entity of the autocompleter :param autocompleter: the entity of the autocompleter
:param print_func: system messages text output function :param printer: system messages text output function
:return: None :return: None
""" """
super().__init__( super().__init__(
@@ -359,7 +287,7 @@ class App(BaseApp):
repeat_command_groups_printing=repeat_command_groups_printing, repeat_command_groups_printing=repeat_command_groups_printing,
override_system_messages=override_system_messages, override_system_messages=override_system_messages,
autocompleter=autocompleter or AutoCompleter(), autocompleter=autocompleter or AutoCompleter(),
print_func=print_func, printer=printer,
) )
def run_polling(self) -> None: def run_polling(self) -> None:
@@ -367,7 +295,7 @@ class App(BaseApp):
Private. Starts the user input processing cycle Private. Starts the user input processing cycle
:return: None :return: None
""" """
self._print_func(self._initial_message) self._viewer.view_initial_message(self._initial_message)
self._pre_cycle_setup() self._pre_cycle_setup()
while True: while True:
if self._repeat_command_groups_printing: if self._repeat_command_groups_printing:
@@ -380,10 +308,9 @@ class App(BaseApp):
try: try:
input_command: InputCommand = InputCommand.parse(raw_command=raw_command) input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
except InputCommandException as error: # noqa F841 except InputCommandException as error: # noqa F841
stderr_result = self._capture_stdout( self._viewer.view_framed_text_from_generator(
lambda: self._error_handler(error, raw_command) # noqa F821 output_text_generator=lambda: self._error_handler(error, raw_command)
) )
self._print_static_framed_text(stderr_result)
continue continue
if self._is_exit_command(input_command): if self._is_exit_command(input_command):
@@ -391,8 +318,9 @@ class App(BaseApp):
return return
if self._is_unknown_command(input_command): if self._is_unknown_command(input_command):
stdout_res = self._capture_stdout(lambda: self._unknown_command_handler(input_command)) self._viewer.view_framed_text_from_generator(
self._print_static_framed_text(stdout_res) output_text_generator=lambda: self._unknown_command_handler(input_command)
)
continue continue
self._process_exist_and_valid_command(input_command) self._process_exist_and_valid_command(input_command)
+9 -9
View File
@@ -78,7 +78,7 @@ class RichRenderer(Renderer):
@staticmethod @staticmethod
def render_messages_on_startup(messages: Iterable[str]) -> str: def render_messages_on_startup(messages: Iterable[str]) -> str:
return "\n".join(messages) return "\n" + "\n".join(messages)
@staticmethod @staticmethod
def render_command_groups_description( def render_command_groups_description(
@@ -87,10 +87,10 @@ class RichRenderer(Renderer):
) -> str: ) -> str:
command_groups_description = "" command_groups_description = ""
for registered_router in registered_routers: for registered_router in registered_routers:
command_groups_description += "\n" + registered_router.title command_groups_description += "\n\n" + registered_router.title
for command_handler in registered_router.command_handlers: for command_handler in registered_router.command_handlers:
handled_command = command_handler.handled_command handled_command = command_handler.handled_command
command_groups_description += description_message_generator( command_groups_description += '\n' + description_message_generator(
handled_command.trigger, handled_command.trigger,
handled_command.description, handled_command.description,
) )
@@ -114,7 +114,7 @@ class RichRenderer(Renderer):
most_similar_command_trigger: str | None most_similar_command_trigger: str | None
) -> 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 "")
) )
@@ -131,14 +131,14 @@ class PlainRenderer(Renderer):
@staticmethod @staticmethod
def render_farewell_message(text: str) -> str: def render_farewell_message(text: str) -> str:
return f"{text} | [https://github.com/koloideal/Argenta](https://github.com/koloideal/Argenta) | made by kolo" return f"\n{text} | https://github.com/koloideal/Argenta | made by kolo"
@staticmethod @staticmethod
def render_text_for_description_message_generator(command: str, description: str) -> str: def render_text_for_description_message_generator(command: str, description: str) -> str:
return f"{command} *=*=* {description}" return f"{command} *=*=* {description}"
def render_messages_on_startup(self, messages: Iterable[str]) -> str: def render_messages_on_startup(self, messages: Iterable[str]) -> str:
return "\n".join(messages) return "\n" + "\n".join(messages)
@staticmethod @staticmethod
def render_command_groups_description( def render_command_groups_description(
@@ -147,10 +147,10 @@ class PlainRenderer(Renderer):
) -> str: ) -> str:
command_groups_description = "" command_groups_description = ""
for registered_router in registered_routers: for registered_router in registered_routers:
command_groups_description += "\n" + registered_router.title command_groups_description += "\n\n" + registered_router.title
for command_handler in registered_router.command_handlers: for command_handler in registered_router.command_handlers:
handled_command = command_handler.handled_command handled_command = command_handler.handled_command
command_groups_description += description_message_generator( command_groups_description += "\n" + description_message_generator(
handled_command.trigger, handled_command.trigger,
handled_command.description, handled_command.description,
) )
@@ -175,7 +175,7 @@ class PlainRenderer(Renderer):
) -> str: ) -> str:
return ( return (
f"Unknown command: {command_trigger}" f"Unknown command: {command_trigger}"
+ (f", most similar:{most_similar_command_trigger}" + (f", most similar: {most_similar_command_trigger}"
if most_similar_command_trigger else "") if most_similar_command_trigger else "")
) )
+71 -2
View File
@@ -1,14 +1,43 @@
from typing import Iterable __all__ = ["Viewer"]
import re
from contextlib import redirect_stdout
from io import StringIO
from typing import Iterable, Callable, TypeAlias
from rich.text import Text
from argenta.app import StaticDividingLine, DynamicDividingLine
from argenta.app.presentation.renderers import Renderer from argenta.app.presentation.renderers import Renderer
from argenta.app.protocols import Printer, DescriptionMessageGenerator from argenta.app.protocols import Printer, DescriptionMessageGenerator
from argenta.app.registered_routers.entity import RegisteredRouters from argenta.app.registered_routers.entity import RegisteredRouters
AVAILABLE_DIVIDING_LINES: TypeAlias = StaticDividingLine | DynamicDividingLine | None
class Viewer: class Viewer:
def __init__(self, printer: Printer, renderer: Renderer): ANSI_ESCAPE_RE: re.Pattern[str] = re.compile(r"\u001b\[[0-9;]*m")
def __init__(
self,
printer: Printer,
renderer: Renderer,
dividing_line: AVAILABLE_DIVIDING_LINES,
override_system_messages: bool
):
self._printer = printer self._printer = printer
self._renderer = renderer self._renderer = renderer
self._dividing_line = dividing_line
self._override_system_messages = override_system_messages
self._stdout_buffer: StringIO = StringIO()
def _capture_stdout(self, func: Callable[[], None]) -> str:
self._stdout_buffer.seek(0)
self._stdout_buffer.truncate(0)
with redirect_stdout(self._stdout_buffer):
func()
return self._stdout_buffer.getvalue()
def view_messages_on_startup(self, messages: Iterable[str]) -> None: def view_messages_on_startup(self, messages: Iterable[str]) -> None:
self._printer(self._renderer.render_messages_on_startup(messages)) self._printer(self._renderer.render_messages_on_startup(messages))
@@ -25,3 +54,43 @@ class Viewer:
) )
) )
def view_initial_message(self, initial_message: str) -> None:
self._printer(initial_message)
def view_framed_text_from_generator(
self,
output_text_generator: Callable[[], None],
is_stdout_redirected_by_router: bool = False,
) -> None:
match (self._dividing_line, is_stdout_redirected_by_router):
case (None, bool()):
output_text_generator()
case (DynamicDividingLine(), False):
stdout_result = self._capture_stdout(
lambda: output_text_generator()
)
clear_text = self.ANSI_ESCAPE_RE.sub("", stdout_result)
max_length_line = max([len(line) for line in clear_text.split("\n")])
max_length_line = (
max_length_line
if 10 <= max_length_line <= 100
else 100
if max_length_line > 100
else 10
)
dividing_line_as_str: str = self._dividing_line.get_full_dynamic_line(
length=max_length_line, is_override=self._override_system_messages
)
self._printer(dividing_line_as_str + "\n")
self._printer(Text.from_ansi(stdout_result.strip("\n")).markup)
self._printer('\n' + dividing_line_as_str)
case (StaticDividingLine() as dividing_line, bool()) | (DynamicDividingLine() as dividing_line, True):
dividing_line_as_str: str = StaticDividingLine(dividing_line.get_unit_part()).get_full_static_line(
is_override=self._override_system_messages
)
self._printer(dividing_line_as_str + '\n')
output_text_generator()
self._printer('\n' + dividing_line_as_str)
case _:
raise NotImplementedError(f"Dividing line with type {self._dividing_line} is not implemented")
+49 -49
View File
@@ -35,7 +35,7 @@ class Router:
:return: None :return: None
""" """
self.title: str = title self.title: str = title
self.disable_redirect_stdout: bool = disable_redirect_stdout self.is_redirect_stdout_disabled: bool = disable_redirect_stdout
self.command_handlers: CommandHandlers = CommandHandlers() self.command_handlers: CommandHandlers = CommandHandlers()
self.aliases: set[str] = set() self.aliases: set[str] = set()
@@ -56,7 +56,7 @@ class Router:
self._update_routing_keys(redefined_command) self._update_routing_keys(redefined_command)
def decorator(func: HandlerFunc) -> HandlerFunc: def decorator(func: HandlerFunc) -> HandlerFunc:
_validate_func_args(func) self._validate_func_args(func)
self.command_handlers.add_handler(CommandHandler(func, redefined_command)) self.command_handlers.add_handler(CommandHandler(func, redefined_command))
return func return func
@@ -116,7 +116,7 @@ class Router:
handle_command = command_handler.handled_command handle_command = command_handler.handled_command
if handle_command.registered_flags.flags: if handle_command.registered_flags.flags:
if input_command_flags.flags: if input_command_flags.flags:
response: Response = _structuring_input_flags(handle_command, input_command_flags) response: Response = self._structuring_input_flags(handle_command, input_command_flags)
command_handler.handling(response) command_handler.handling(response)
else: else:
response = Response(ResponseStatus.ALL_FLAGS_VALID) response = Response(ResponseStatus.ALL_FLAGS_VALID)
@@ -133,53 +133,53 @@ class Router:
response = Response(ResponseStatus.ALL_FLAGS_VALID) response = Response(ResponseStatus.ALL_FLAGS_VALID)
command_handler.handling(response) command_handler.handling(response)
@staticmethod
def _structuring_input_flags(handled_command: Command, input_flags: InputFlags) -> Response:
"""
Private. Validates flags of input command
:param handled_command: entity of the handled command
:param input_flags:
:return: entity of response as Response
"""
invalid_value_flags, undefined_flags = False, False
def _structuring_input_flags(handled_command: Command, input_flags: InputFlags) -> Response: for flag in input_flags:
""" flag_status: ValidationStatus = handled_command.validate_input_flag(flag)
Private. Validates flags of input command flag.status = flag_status
:param handled_command: entity of the handled command if flag_status == ValidationStatus.INVALID:
:param input_flags: invalid_value_flags = True
:return: entity of response as Response elif flag_status == ValidationStatus.UNDEFINED:
""" undefined_flags = True
invalid_value_flags, undefined_flags = False, False
for flag in input_flags: status = ResponseStatus.from_flags(
flag_status: ValidationStatus = handled_command.validate_input_flag(flag) has_invalid_value_flags=invalid_value_flags,
flag.status = flag_status has_undefined_flags=undefined_flags
if flag_status == ValidationStatus.INVALID:
invalid_value_flags = True
elif flag_status == ValidationStatus.UNDEFINED:
undefined_flags = True
status = ResponseStatus.from_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: HandlerFunc) -> None:
"""
Private. Validates the arguments of the handler
:param func: entity of the handler func
:return: None if func is valid else raise exception
"""
transferred_args = getfullargspec(func).args
if len(transferred_args) == 0:
raise RequiredArgumentNotPassedException()
response_arg: str = transferred_args[0]
func_annotations: dict[str, None] = get_annotations(func)
response_arg_annotation = func_annotations.get(response_arg)
if response_arg_annotation is not None and response_arg_annotation is not Response:
source_line: int = getsourcelines(func)[1]
Console().print(
f'\nFile "{getsourcefile(func)}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
+ f"of argument([green]{response_arg}[/green]) passed to the handler must be [/i][bold blue]{Response}[/bold blue],"
+ f" [i]but[/i] [bold blue]{response_arg_annotation}[/bold blue] [i]is specified[/i]",
highlight=False,
) )
return Response(status=status, input_flags=input_flags)
@staticmethod
def _validate_func_args(func: HandlerFunc) -> None:
"""
Private. Validates the arguments of the handler
:param func: entity of the handler func
:return: None if func is valid else raise exception
"""
transferred_args = getfullargspec(func).args
if len(transferred_args) == 0:
raise RequiredArgumentNotPassedException()
response_arg: str = transferred_args[0]
func_annotations: dict[str, None] = get_annotations(func)
response_arg_annotation = func_annotations.get(response_arg)
if response_arg_annotation is not None and response_arg_annotation is not Response:
source_line: int = getsourcelines(func)[1]
Console().print(
f'\nFile "{getsourcefile(func)}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
+ f"of argument([green]{response_arg}[/green]) passed to the handler must be [/i][bold blue]{Response}[/bold blue],"
+ f" [i]but[/i] [bold blue]{response_arg_annotation}[/bold blue] [i]is specified[/i]",
highlight=False,
)
@@ -35,7 +35,7 @@ def test_empty_input_triggers_empty_command_handler(monkeypatch: pytest.MonkeyPa
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('test command') print('test command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
app.set_empty_command_handler(lambda: print('Empty input command')) app.set_empty_command_handler(lambda: print('Empty input command'))
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -61,7 +61,7 @@ def test_unknown_command_triggers_unknown_command_handler(monkeypatch: pytest.Mo
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('test command') print('test command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}')) app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -82,7 +82,7 @@ def test_mixed_valid_and_unknown_commands_handled_correctly(monkeypatch: pytest.
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('test command') print('test command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}')) app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -107,7 +107,7 @@ def test_multiple_commands_with_unknown_command_in_between(monkeypatch: pytest.M
def test1(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] def test1(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('more command') print('more command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}')) app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -135,7 +135,7 @@ def test_unregistered_flag_without_value_is_accessible(monkeypatch: pytest.Monke
if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED: if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED:
print(f'test command with undefined flag: {undefined_flag.string_entity}') print(f'test command with undefined flag: {undefined_flag.string_entity}')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -159,7 +159,7 @@ def test_unregistered_flag_with_value_is_accessible(monkeypatch: pytest.MonkeyPa
else: else:
raise raise
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -182,7 +182,7 @@ def test_registered_and_unregistered_flags_coexist(monkeypatch: pytest.MonkeyPat
if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED: if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED:
print(f'connecting to host with flag: {undefined_flag.string_entity} {undefined_flag.input_value}') print(f'connecting to host with flag: {undefined_flag.string_entity} {undefined_flag.input_value}')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -207,7 +207,7 @@ def test_flag_without_value_triggers_incorrect_syntax_handler(monkeypatch: pytes
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('test command') print('test command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
app.set_incorrect_input_syntax_handler(lambda command: print(f'Incorrect flag syntax: "{command}"')) app.set_incorrect_input_syntax_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -233,7 +233,7 @@ def test_repeated_flags_trigger_repeated_flags_handler(monkeypatch: pytest.Monke
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('test command') print('test command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
app.set_repeated_input_flags_handler(lambda command: print(f'Repeated input flags: "{command}"')) app.set_repeated_input_flags_handler(lambda command: print(f'Repeated input flags: "{command}"'))
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -36,7 +36,7 @@ def test_simple_command_executes_successfully(monkeypatch: pytest.MonkeyPatch, c
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('test command') print('test command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -60,7 +60,7 @@ def test_two_commands_execute_sequentially(monkeypatch: pytest.MonkeyPatch, caps
def test2(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] def test2(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('some command') print('some command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -88,7 +88,7 @@ def test_three_commands_execute_sequentially(monkeypatch: pytest.MonkeyPatch, ca
def test2(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] def test2(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('more command') print('more command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -116,7 +116,7 @@ def test_custom_flag_without_value_is_recognized(monkeypatch: pytest.MonkeyPatch
if valid_flag and valid_flag.status == ValidationStatus.VALID: if valid_flag and valid_flag.status == ValidationStatus.VALID:
print(f'\nhelp for {valid_flag.name} flag\n') print(f'\nhelp for {valid_flag.name} flag\n')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -139,7 +139,7 @@ def test_custom_flag_with_regex_validation_accepts_valid_value(monkeypatch: pyte
if valid_flag and valid_flag.status == ValidationStatus.VALID: if valid_flag and valid_flag.status == ValidationStatus.VALID:
print(f'flag value for {valid_flag.name} flag : {valid_flag.input_value}') print(f'flag value for {valid_flag.name} flag : {valid_flag.input_value}')
app = App(override_system_messages=True, repeat_command_groups_printing=True, print_func=print) app = App(override_system_messages=True, repeat_command_groups_printing=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -167,7 +167,7 @@ def test_predefined_short_help_flag_is_recognized(monkeypatch: pytest.MonkeyPatc
if valid_flag and valid_flag.status == ValidationStatus.VALID: if valid_flag and valid_flag.status == ValidationStatus.VALID:
print(f'help for {valid_flag.name} flag') print(f'help for {valid_flag.name} flag')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -190,7 +190,7 @@ def test_predefined_info_flag_is_recognized(monkeypatch: pytest.MonkeyPatch, cap
if valid_flag and valid_flag.status == ValidationStatus.VALID: if valid_flag and valid_flag.status == ValidationStatus.VALID:
print('info about test command') print('info about test command')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -213,7 +213,7 @@ def test_predefined_host_flag_with_value_is_recognized(monkeypatch: pytest.Monke
if valid_flag and valid_flag.status == ValidationStatus.VALID: if valid_flag and valid_flag.status == ValidationStatus.VALID:
print(f'connecting to host {valid_flag.input_value}') print(f'connecting to host {valid_flag.input_value}')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)
@@ -242,7 +242,7 @@ def test_two_predefined_flags_are_recognized_together(monkeypatch: pytest.Monkey
if (host_flag and host_flag.status == ValidationStatus.VALID) and (port_flag and port_flag.status == ValidationStatus.VALID): if (host_flag and host_flag.status == ValidationStatus.VALID) and (port_flag and port_flag.status == ValidationStatus.VALID):
print(f'connecting to host {host_flag.input_value} and port {port_flag.input_value}') print(f'connecting to host {host_flag.input_value} and port {port_flag.input_value}')
app = App(override_system_messages=True, print_func=print) app = App(override_system_messages=True, printer=print)
app.include_router(router) app.include_router(router)
orchestrator.start_polling(app) orchestrator.start_polling(app)