This commit is contained in:
2025-05-21 17:01:35 +03:00
parent 33cb528b1d
commit 365347ea7f
9 changed files with 129 additions and 81 deletions
+2 -11
View File
@@ -10,22 +10,13 @@ from argenta.orchestrator import Orchestrator
router = Router()
@router.command(Command('case are'))
@router.command(Command("case are"))
def handler(response: Response):
print(response.status)
app = App(repeat_command_groups=False)
app.include_router(router)
app.run_polling()
+1 -1
View File
@@ -21,7 +21,7 @@ console = Console()
)
)
def command_help(response: Response):
case = input('test > ')
case = input("test > ")
print(case)
print(response.status)
print(response.undefined_flags.get_flags())
+1
View File
@@ -5,6 +5,7 @@ class PredefinedMessages(StrEnum):
"""
Public. A dataclass with predetermined messages for quick use
"""
USAGE = "[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]"
HELP = "[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]"
AUTOCOMPLETE = "[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>"
+70 -23
View File
@@ -22,7 +22,8 @@ from argenta.response import Response
class BaseApp:
def __init__(self,
def __init__(
self,
prompt: str,
initial_message: str,
farewell_message: str,
@@ -33,7 +34,8 @@ class BaseApp:
repeat_command_groups: bool,
override_system_messages: bool,
autocompleter: AutoCompleter,
print_func: Callable[[str], None]) -> None:
print_func: Callable[[str], None],
) -> None:
self._prompt = prompt
self._print_func = print_func
self._exit_command = exit_command
@@ -47,18 +49,30 @@ class BaseApp:
self._farewell_message = farewell_message
self._initial_message = initial_message
self._description_message_gen: Callable[[str, str], str] = (lambda command, description: f"[{command}] *=*=* {description}")
self._description_message_gen: Callable[[str, str], str] = (
lambda command, description: f"[{command}] *=*=* {description}"
)
self._registered_routers: RegisteredRouters = RegisteredRouters()
self._messages_on_startup: list[str] = []
self._all_registered_triggers_in_lower_case: list[str] = []
self._all_registered_triggers_in_default_case: list[str] = []
self._incorrect_input_syntax_handler: Callable[[str], None] = (lambda raw_command: print_func(f"Incorrect flag syntax: {raw_command}"))
self._repeated_input_flags_handler: Callable[[str], None] = (lambda raw_command: print_func(f"Repeated input flags: {raw_command}"))
self._empty_input_command_handler: Callable[[], None] = lambda: print_func("Empty input command")
self._unknown_command_handler: Callable[[InputCommand], None] = (lambda command: print_func(f"Unknown command: {command.get_trigger()}"))
self._exit_command_handler: Callable[[Response], None] = (lambda response: print_func(self._farewell_message))
self._incorrect_input_syntax_handler: Callable[[str], None] = (
lambda raw_command: print_func(f"Incorrect flag syntax: {raw_command}")
)
self._repeated_input_flags_handler: Callable[[str], None] = (
lambda raw_command: print_func(f"Repeated input flags: {raw_command}")
)
self._empty_input_command_handler: Callable[[], None] = lambda: print_func(
"Empty input command"
)
self._unknown_command_handler: Callable[[InputCommand], None] = (
lambda command: print_func(f"Unknown command: {command.get_trigger()}")
)
self._exit_command_handler: Callable[[Response], None] = (
lambda response: print_func(self._farewell_message)
)
def set_description_message_pattern(self, _: Callable[[str, str], str]) -> None:
"""
@@ -194,14 +208,19 @@ class BaseApp:
"""
input_command_trigger = command.get_trigger()
if self._ignore_command_register:
if input_command_trigger.lower() in self._all_registered_triggers_in_lower_case:
if (
input_command_trigger.lower()
in self._all_registered_triggers_in_lower_case
):
return False
else:
if input_command_trigger in self._all_registered_triggers_in_default_case:
return False
return True
def _error_handler(self, error: BaseInputCommandException, raw_command: str) -> None:
def _error_handler(
self, error: BaseInputCommandException, raw_command: str
) -> None:
"""
Private. Handles parsing errors of the entered command
:param error: error being handled
@@ -257,10 +276,12 @@ class BaseApp:
"""
self._prompt = "[italic dim bold]What do you want to do?\n"
self._initial_message = (
"\n"+f"[bold red]{text2art(self._initial_message, font='tarty1')}"+"\n"
"\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n"
)
self._farewell_message = (
"[bold red]\n\n"+text2art(self._farewell_message, font='chanky')+"\n[/bold red]\n"
"[bold red]\n\n"
+ text2art(self._farewell_message, font="chanky")
+ "\n[/bold red]\n"
"[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n"
)
self._description_message_gen = lambda command, description: (
@@ -307,7 +328,9 @@ class BaseApp:
self._all_registered_triggers_in_default_case.extend(combined)
self._all_registered_triggers_in_lower_case.extend(x.lower() for x in combined)
self._all_registered_triggers_in_lower_case.extend(
x.lower() for x in combined
)
self._autocompleter.initial_setup(self._all_registered_triggers_in_lower_case)
@@ -315,14 +338,18 @@ class BaseApp:
seen = {}
for item in self._all_registered_triggers_in_lower_case:
if item in seen:
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{item}[/b blue]")
Console().print(
f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{item}[/b blue]"
)
else:
seen[item] = True
else:
seen = {}
for item in self._all_registered_triggers_in_default_case:
if item in seen:
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{item}[/b blue]")
Console().print(
f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{item}[/b blue]"
)
else:
seen[item] = True
@@ -340,7 +367,8 @@ class BaseApp:
class App(BaseApp):
def __init__(self,
def __init__(
self,
prompt: str = "What do you want to do?\n",
initial_message: str = "Argenta\n",
farewell_message: str = "\nSee you\n",
@@ -351,7 +379,8 @@ class App(BaseApp):
repeat_command_groups: bool = True,
override_system_messages: bool = False,
autocompleter: AutoCompleter = AutoCompleter(),
print_func: Callable[[str], None] = Console().print) -> None:
print_func: Callable[[str], None] = Console().print,
) -> None:
"""
Public. The essence of the application itself.
Configures and manages all aspects of the behavior and presentation of the user interacting with the user
@@ -368,7 +397,8 @@ class App(BaseApp):
:param print_func: system messages text output function
:return: None
"""
super().__init__(prompt=prompt,
super().__init__(
prompt=prompt,
initial_message=initial_message,
farewell_message=farewell_message,
exit_command=exit_command,
@@ -378,7 +408,8 @@ class App(BaseApp):
repeat_command_groups=repeat_command_groups,
override_system_messages=override_system_messages,
autocompleter=autocompleter,
print_func=print_func)
print_func=print_func,
)
def run_polling(self) -> None:
"""
@@ -425,13 +456,29 @@ class App(BaseApp):
for registered_router in self._registered_routers:
if registered_router.disable_redirect_stdout:
if isinstance(self._dividing_line, StaticDividingLine):
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
self._print_func(
self._dividing_line.get_full_static_line(
self._override_system_messages
)
)
registered_router.finds_appropriate_handler(input_command)
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
self._print_func(
self._dividing_line.get_full_static_line(
self._override_system_messages
)
)
else:
self._print_func(StaticDividingLine(self._dividing_line.get_unit_part()).get_full_static_line(self._override_system_messages))
self._print_func(
StaticDividingLine(
self._dividing_line.get_unit_part()
).get_full_static_line(self._override_system_messages)
)
registered_router.finds_appropriate_handler(input_command)
self._print_func(StaticDividingLine(self._dividing_line.get_unit_part()).get_full_static_line(self._override_system_messages))
self._print_func(
StaticDividingLine(
self._dividing_line.get_unit_part()
).get_full_static_line(self._override_system_messages)
)
else:
with redirect_stdout(io.StringIO()) as f:
registered_router.finds_appropriate_handler(input_command)
+14 -4
View File
@@ -1,7 +1,17 @@
__all__ = ["Flag", "InputFlag", "UndefinedInputFlags", "ValidInputFlags", "InvalidValueInputFlags", "Flags"]
__all__ = [
"Flag",
"InputFlag",
"UndefinedInputFlags",
"ValidInputFlags",
"InvalidValueInputFlags",
"Flags",
]
from argenta.command.flag.models import Flag, InputFlag
from argenta.command.flag.flags.models import (UndefinedInputFlags,
ValidInputFlags, Flags,
InvalidValueInputFlags)
from argenta.command.flag.flags.models import (
UndefinedInputFlags,
ValidInputFlags,
Flags,
InvalidValueInputFlags,
)
+4 -1
View File
@@ -87,7 +87,10 @@ class Flag(BaseFlag):
class InputFlag(BaseFlag):
def __init__(
self, name: str, prefix: Literal["-", "--", "---"] = "--", value: str | None = None
self,
name: str,
prefix: Literal["-", "--", "---"] = "--",
value: str | None = None,
):
"""
Public. The entity of the flag of the entered command
+1 -1
View File
@@ -139,7 +139,7 @@ class InputCommand(BaseCommand):
return self._input_flags
@staticmethod
def parse(raw_command: str) -> 'InputCommand':
def parse(raw_command: str) -> "InputCommand":
"""
Private. Parse the raw input command
:param raw_command: raw input command
-8
View File
@@ -16,11 +16,3 @@ def get_time_of_pre_cycle_setup(app: App) -> float:
app.pre_cycle_setup()
end = time()
return end - start
+7 -3
View File
@@ -22,7 +22,9 @@ from argenta.router.exceptions import (
class Router:
def __init__(self, title: str | None = "Awesome title", disable_redirect_stdout: bool = False):
def __init__(
self, title: str | None = "Awesome title", disable_redirect_stdout: bool = False
):
"""
Public. Directly configures and manages handlers
:param title: the title of the router, displayed when displaying the available commands
@@ -196,10 +198,12 @@ class Router:
file_path: str | None = getsourcefile(func)
source_line: int = getsourcelines(func)[1]
fprint = Console().print
fprint(f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
fprint(
f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
f"of argument([green]{transferred_arg}[/green]) passed to the handler is [/i][bold blue]{Response}[/bold blue],"
f" [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]",
highlight=False)
highlight=False,
)
def set_command_register_ignore(self, _: bool) -> None:
"""