mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
Merge pull request #3 from koloideal/feat/di
Implementation of the Dependency Injection principle
This commit is contained in:
+14
-3
@@ -4,15 +4,26 @@ from argenta import App, Orchestrator
|
|||||||
from argenta.app import PredefinedMessages, DynamicDividingLine, AutoCompleter
|
from argenta.app import PredefinedMessages, DynamicDividingLine, AutoCompleter
|
||||||
from argenta.orchestrator import ArgParser
|
from argenta.orchestrator import ArgParser
|
||||||
from argenta.orchestrator.argparser import BooleanArgument, ValueArgument
|
from argenta.orchestrator.argparser import BooleanArgument, ValueArgument
|
||||||
|
from dishka import Provider, provide, Scope # type: ignore
|
||||||
|
|
||||||
|
|
||||||
arg_parser: ArgParser = ArgParser(processed_args=[BooleanArgument(name="repeat", is_deprecated=True),
|
class temProvider(Provider):
|
||||||
ValueArgument(name="required", is_required=True)])
|
@provide(scope=Scope.APP)
|
||||||
|
def get_apace(self) -> int:
|
||||||
|
return 1234
|
||||||
|
|
||||||
|
arg_parser: ArgParser = ArgParser(
|
||||||
|
processed_args=[
|
||||||
|
BooleanArgument(name="repeat", is_deprecated=True),
|
||||||
|
ValueArgument(name="required", is_required=True),
|
||||||
|
]
|
||||||
|
)
|
||||||
app: App = App(
|
app: App = App(
|
||||||
dividing_line=DynamicDividingLine(),
|
dividing_line=DynamicDividingLine(),
|
||||||
autocompleter=AutoCompleter(),
|
autocompleter=AutoCompleter(),
|
||||||
)
|
)
|
||||||
orchestrator: Orchestrator = Orchestrator()
|
orchestrator: Orchestrator = Orchestrator(arg_parser, custom_providers=[temProvider()])
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
app.include_router(work_router)
|
app.include_router(work_router)
|
||||||
|
|||||||
@@ -1,18 +1,20 @@
|
|||||||
from argenta.command import Command, PredefinedFlags, Flags, Flag, PossibleValues
|
from argenta.command import Command, PredefinedFlags, Flags, Flag, PossibleValues
|
||||||
from argenta.response import Response
|
from argenta.response import Response
|
||||||
from argenta import Router
|
from argenta import Router
|
||||||
|
from argenta.di import FromDishka
|
||||||
|
|
||||||
|
|
||||||
work_router: Router = Router(title="Work points:")
|
work_router: Router = Router(title="Work points:", disable_redirect_stdout=True)
|
||||||
|
|
||||||
flag = Flag('csdv', possible_values=PossibleValues.NEITHER)
|
flag = Flag("csdv", possible_values=PossibleValues.NEITHER)
|
||||||
|
|
||||||
|
|
||||||
@work_router.command(
|
@work_router.command(
|
||||||
Command("get",
|
Command(
|
||||||
|
"get",
|
||||||
description="Get Help",
|
description="Get Help",
|
||||||
aliases=["help", "Get_help"],
|
aliases=["help", "Get_help"],
|
||||||
flags=Flags([PredefinedFlags.PORT, PredefinedFlags.HOST])
|
flags=Flags([PredefinedFlags.PORT, PredefinedFlags.HOST]),
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
def command_help(response: Response):
|
def command_help(response: Response):
|
||||||
@@ -21,6 +23,5 @@ def command_help(response: Response):
|
|||||||
|
|
||||||
|
|
||||||
@work_router.command("run")
|
@work_router.command("run")
|
||||||
def command_start_solving(response: Response):
|
def command_start_solving(response: Response, argspace: FromDishka[int]):
|
||||||
print(response.status)
|
print(argspace)
|
||||||
print(response.input_flags.flags)
|
|
||||||
|
|||||||
+5
-2
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "argenta"
|
name = "argenta"
|
||||||
version = "1.1.1"
|
version = "1.1.2"
|
||||||
description = "Python library for building modular CLI applications"
|
description = "Python library for building modular CLI applications"
|
||||||
authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }]
|
authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }]
|
||||||
requires-python = ">=3.11"
|
requires-python = ">=3.11"
|
||||||
@@ -10,6 +10,7 @@ dependencies = [
|
|||||||
"rich (>=14.0.0,<15.0.0)",
|
"rich (>=14.0.0,<15.0.0)",
|
||||||
"art (>=6.4,<7.0)",
|
"art (>=6.4,<7.0)",
|
||||||
"pyreadline3>=3.5.4; sys_platform == 'win32'",
|
"pyreadline3>=3.5.4; sys_platform == 'win32'",
|
||||||
|
"dishka>=1.7.2",
|
||||||
]
|
]
|
||||||
|
|
||||||
[tool.ruff]
|
[tool.ruff]
|
||||||
@@ -25,6 +26,9 @@ exclude = [
|
|||||||
[tool.pyright]
|
[tool.pyright]
|
||||||
typeCheckingMode = "strict"
|
typeCheckingMode = "strict"
|
||||||
|
|
||||||
|
[tool.mypy]
|
||||||
|
disable_error_code = "import-untyped"
|
||||||
|
|
||||||
[build-system]
|
[build-system]
|
||||||
requires = ["hatchling"]
|
requires = ["hatchling"]
|
||||||
build-backend = "hatchling.build"
|
build-backend = "hatchling.build"
|
||||||
@@ -36,4 +40,3 @@ dev = [
|
|||||||
"ruff>=0.12.12",
|
"ruff>=0.12.12",
|
||||||
"wemake-python-styleguide>=0.17.0",
|
"wemake-python-styleguide>=0.17.0",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
__all__ = ["App", "Orchestrator", "Router"]
|
__all__ = ["App", "Orchestrator", "Router"]
|
||||||
|
|
||||||
|
|
||||||
from argenta.orchestrator.entity import Orchestrator, App
|
from argenta.orchestrator.entity import Orchestrator
|
||||||
|
from argenta.app.models import App
|
||||||
from argenta.router.entity import Router
|
from argenta.router.entity import Router
|
||||||
|
|||||||
+128
-57
@@ -3,7 +3,6 @@ import re
|
|||||||
from contextlib import redirect_stdout
|
from contextlib import redirect_stdout
|
||||||
from typing import Never, TypeAlias
|
from typing import Never, TypeAlias
|
||||||
|
|
||||||
from argenta.orchestrator.argparser.entity import ArgSpace
|
|
||||||
from art import text2art # pyright: ignore[reportMissingTypeStubs, reportUnknownVariableType]
|
from art import text2art # pyright: ignore[reportMissingTypeStubs, reportUnknownVariableType]
|
||||||
from rich.console import Console
|
from rich.console import Console
|
||||||
from rich.markup import escape
|
from rich.markup import escape
|
||||||
@@ -32,7 +31,10 @@ Matches: TypeAlias = list[str] | list[Never]
|
|||||||
|
|
||||||
|
|
||||||
class BaseApp:
|
class BaseApp:
|
||||||
def __init__(self, *, prompt: str,
|
def __init__(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
prompt: str,
|
||||||
initial_message: str,
|
initial_message: str,
|
||||||
farewell_message: str,
|
farewell_message: str,
|
||||||
exit_command: Command,
|
exit_command: Command,
|
||||||
@@ -43,7 +45,7 @@ class BaseApp:
|
|||||||
override_system_messages: bool,
|
override_system_messages: bool,
|
||||||
autocompleter: AutoCompleter,
|
autocompleter: AutoCompleter,
|
||||||
print_func: Printer,
|
print_func: Printer,
|
||||||
argspace: ArgSpace | None = None) -> None:
|
) -> None:
|
||||||
self._prompt: str = prompt
|
self._prompt: str = prompt
|
||||||
self._print_func: Printer = print_func
|
self._print_func: Printer = print_func
|
||||||
self._exit_command: Command = exit_command
|
self._exit_command: Command = exit_command
|
||||||
@@ -53,27 +55,44 @@ class BaseApp:
|
|||||||
self._repeat_command_groups_description: bool = repeat_command_groups
|
self._repeat_command_groups_description: bool = repeat_command_groups
|
||||||
self._override_system_messages: bool = override_system_messages
|
self._override_system_messages: bool = override_system_messages
|
||||||
self._autocompleter: AutoCompleter = autocompleter
|
self._autocompleter: AutoCompleter = autocompleter
|
||||||
self._argspace: ArgSpace | None = argspace
|
|
||||||
|
|
||||||
self._farewell_message: str = farewell_message
|
self._farewell_message: str = farewell_message
|
||||||
self._initial_message: str = initial_message
|
self._initial_message: str = initial_message
|
||||||
|
|
||||||
self._description_message_gen: DescriptionMessageGenerator = lambda command, description: f"{command} *=*=* {description}"
|
self._description_message_gen: DescriptionMessageGenerator = (
|
||||||
self._registered_routers: RegisteredRouters = RegisteredRouters()
|
lambda command, description: f"{command} *=*=* {description}"
|
||||||
|
)
|
||||||
|
self.registered_routers: RegisteredRouters = RegisteredRouters()
|
||||||
self._messages_on_startup: list[str] = []
|
self._messages_on_startup: list[str] = []
|
||||||
|
|
||||||
self._matching_lower_triggers_with_routers: dict[str, Router] = {}
|
self._matching_lower_triggers_with_routers: dict[str, Router] = {}
|
||||||
self._matching_default_triggers_with_routers: dict[str, Router] = {}
|
self._matching_default_triggers_with_routers: dict[str, Router] = {}
|
||||||
|
|
||||||
self._current_matching_triggers_with_routers: dict[str, Router] = self._matching_lower_triggers_with_routers if self._ignore_command_register else self._matching_default_triggers_with_routers
|
self._current_matching_triggers_with_routers: dict[str, Router] = (
|
||||||
|
self._matching_lower_triggers_with_routers
|
||||||
|
if self._ignore_command_register
|
||||||
|
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] = (
|
||||||
self._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = lambda _: print_func(f"Repeated input flags: {_}")
|
lambda _: print_func(f"Incorrect flag syntax: {_}")
|
||||||
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._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = (
|
||||||
self._exit_command_handler: NonStandardBehaviorHandler[Response] = lambda _: print_func(self._farewell_message)
|
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._exit_command_handler: NonStandardBehaviorHandler[Response] = (
|
||||||
|
lambda _: print_func(self._farewell_message)
|
||||||
|
)
|
||||||
|
|
||||||
def set_description_message_pattern(self, _: DescriptionMessageGenerator, /) -> None:
|
def set_description_message_pattern(
|
||||||
|
self, _: DescriptionMessageGenerator, /
|
||||||
|
) -> None:
|
||||||
"""
|
"""
|
||||||
Public. Sets the output pattern of the available commands
|
Public. Sets the output pattern of the available commands
|
||||||
:param _: output pattern of the available commands
|
:param _: output pattern of the available commands
|
||||||
@@ -81,7 +100,9 @@ class BaseApp:
|
|||||||
"""
|
"""
|
||||||
self._description_message_gen = _
|
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
|
Public. Sets the handler for incorrect flags when entering a command
|
||||||
:param _: handler for incorrect flags when entering a command
|
:param _: handler for incorrect flags when entering a command
|
||||||
@@ -89,7 +110,9 @@ class BaseApp:
|
|||||||
"""
|
"""
|
||||||
self._incorrect_input_syntax_handler = _
|
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
|
Public. Sets the handler for repeated flags when entering a command
|
||||||
:param _: handler for repeated flags when entering a command
|
:param _: handler for repeated flags when entering a command
|
||||||
@@ -97,7 +120,9 @@ class BaseApp:
|
|||||||
"""
|
"""
|
||||||
self._repeated_input_flags_handler = _
|
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
|
Public. Sets the handler for unknown commands when entering a command
|
||||||
:param _: handler for unknown commands when entering a command
|
:param _: handler for unknown commands when entering a command
|
||||||
@@ -113,7 +138,9 @@ class BaseApp:
|
|||||||
"""
|
"""
|
||||||
self._empty_input_command_handler = _
|
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
|
Public. Sets the handler for exit command when entering a command
|
||||||
:param _: handler for exit command when entering a command
|
:param _: handler for exit command when entering a command
|
||||||
@@ -126,7 +153,7 @@ class BaseApp:
|
|||||||
Private. Prints the description of the available commands
|
Private. Prints the description of the available commands
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
for registered_router in self._registered_routers:
|
for registered_router in self.registered_routers:
|
||||||
if registered_router.title:
|
if registered_router.title:
|
||||||
self._print_func(registered_router.title)
|
self._print_func(registered_router.title)
|
||||||
for command_handler in registered_router.command_handlers:
|
for command_handler in registered_router.command_handlers:
|
||||||
@@ -170,11 +197,15 @@ class BaseApp:
|
|||||||
|
|
||||||
elif isinstance(self._dividing_line, StaticDividingLine): # pyright: ignore[reportUnnecessaryIsInstance]
|
elif isinstance(self._dividing_line, StaticDividingLine): # pyright: ignore[reportUnnecessaryIsInstance]
|
||||||
self._print_func(
|
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"))
|
print(text.strip("\n"))
|
||||||
self._print_func(
|
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:
|
else:
|
||||||
@@ -189,13 +220,9 @@ class BaseApp:
|
|||||||
trigger = command.trigger
|
trigger = command.trigger
|
||||||
exit_trigger = self._exit_command.trigger
|
exit_trigger = self._exit_command.trigger
|
||||||
if self._ignore_command_register:
|
if self._ignore_command_register:
|
||||||
if (
|
if trigger.lower() == exit_trigger.lower():
|
||||||
trigger.lower() == exit_trigger.lower()
|
|
||||||
):
|
|
||||||
return True
|
return True
|
||||||
elif trigger.lower() in [
|
elif trigger.lower() in [x.lower() for x in self._exit_command.aliases]:
|
||||||
x.lower() for x in self._exit_command.aliases
|
|
||||||
]:
|
|
||||||
return True
|
return True
|
||||||
else:
|
else:
|
||||||
if trigger == exit_trigger:
|
if trigger == exit_trigger:
|
||||||
@@ -212,16 +239,18 @@ class BaseApp:
|
|||||||
"""
|
"""
|
||||||
input_command_trigger = command.trigger
|
input_command_trigger = command.trigger
|
||||||
if self._ignore_command_register:
|
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
|
return False
|
||||||
else:
|
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 False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
def _error_handler(
|
def _error_handler(self, error: InputCommandException, raw_command: str) -> None:
|
||||||
self, error: InputCommandException, raw_command: str
|
|
||||||
) -> None:
|
|
||||||
"""
|
"""
|
||||||
Private. Handles parsing errors of the entered command
|
Private. Handles parsing errors of the entered command
|
||||||
:param error: error being handled
|
:param error: error being handled
|
||||||
@@ -246,9 +275,9 @@ class BaseApp:
|
|||||||
def _(response: Response) -> None:
|
def _(response: Response) -> None:
|
||||||
self._exit_command_handler(response)
|
self._exit_command_handler(response)
|
||||||
|
|
||||||
if system_router not in self._registered_routers.registered_routers:
|
if system_router not in self.registered_routers.registered_routers:
|
||||||
system_router.command_register_ignore = self._ignore_command_register
|
system_router.command_register_ignore = self._ignore_command_register
|
||||||
self._registered_routers.add_registered_router(system_router)
|
self.registered_routers.add_registered_router(system_router)
|
||||||
|
|
||||||
def _most_similar_command(self, unknown_command: str) -> str | None:
|
def _most_similar_command(self, unknown_command: str) -> str | None:
|
||||||
all_commands = list(self._current_matching_triggers_with_routers.keys())
|
all_commands = list(self._current_matching_triggers_with_routers.keys())
|
||||||
@@ -275,26 +304,36 @@ class BaseApp:
|
|||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
self._prompt = f"[italic dim bold]{self._prompt}"
|
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 = (
|
self._farewell_message = (
|
||||||
"[bold red]\n\n" +
|
"[bold red]\n\n"
|
||||||
str(text2art(self._farewell_message, font="chanky")) + # pyright: ignore[reportUnknownArgumentType]
|
+ str(text2art(self._farewell_message, font="chanky")) # pyright: ignore[reportUnknownArgumentType]
|
||||||
"\n[/bold red]\n" +
|
+ "\n[/bold red]\n"
|
||||||
"[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\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: (
|
self._description_message_gen = lambda command, description: (
|
||||||
f"[bold red]{escape('[' + command + ']')}[/bold red] "
|
f"[bold red]{escape('[' + command + ']')}[/bold red] "
|
||||||
f"[blue dim]*=*=*[/blue dim] "
|
f"[blue dim]*=*=*[/blue dim] "
|
||||||
f"[bold yellow italic]{escape(description)}"
|
f"[bold yellow italic]{escape(description)}"
|
||||||
)
|
)
|
||||||
self._incorrect_input_syntax_handler = lambda raw_command: self._print_func(f"[red bold]Incorrect flag syntax: {escape(raw_command)}")
|
self._incorrect_input_syntax_handler = lambda raw_command: self._print_func(
|
||||||
self._repeated_input_flags_handler = lambda raw_command: self._print_func(f"[red bold]Repeated input flags: {escape(raw_command)}")
|
f"[red bold]Incorrect flag syntax: {escape(raw_command)}"
|
||||||
self._empty_input_command_handler = lambda: self._print_func("[red bold]Empty input command")
|
)
|
||||||
|
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"
|
||||||
|
)
|
||||||
|
|
||||||
def unknown_command_handler(command: InputCommand) -> None:
|
def unknown_command_handler(command: InputCommand) -> None:
|
||||||
cmd_trg: str = command.trigger
|
cmd_trg: str = command.trigger
|
||||||
mst_sim_cmd: str | None = self._most_similar_command(cmd_trg)
|
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 = (
|
second_part_of_text = (
|
||||||
("[red], most similar:[/red] " + ("[blue]" + mst_sim_cmd + "[/blue]"))
|
("[red], most similar:[/red] " + ("[blue]" + mst_sim_cmd + "[/blue]"))
|
||||||
if mst_sim_cmd
|
if mst_sim_cmd
|
||||||
@@ -311,21 +350,27 @@ class BaseApp:
|
|||||||
"""
|
"""
|
||||||
self._setup_system_router()
|
self._setup_system_router()
|
||||||
|
|
||||||
for router_entity in self._registered_routers:
|
for router_entity in self.registered_routers:
|
||||||
router_triggers = router_entity.triggers
|
router_triggers = router_entity.triggers
|
||||||
router_aliases = router_entity.aliases
|
router_aliases = router_entity.aliases
|
||||||
combined = router_triggers + router_aliases
|
combined = router_triggers + router_aliases
|
||||||
|
|
||||||
for trigger in combined:
|
for trigger in combined:
|
||||||
self._matching_default_triggers_with_routers[trigger] = router_entity
|
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())
|
||||||
|
)
|
||||||
|
|
||||||
seen = {}
|
seen = {}
|
||||||
for item in list(self._current_matching_triggers_with_routers.keys()):
|
for item in list(self._current_matching_triggers_with_routers.keys()):
|
||||||
if item in seen:
|
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:
|
else:
|
||||||
seen[item] = True
|
seen[item] = True
|
||||||
|
|
||||||
@@ -352,7 +397,8 @@ DEFAULT_EXIT_COMMAND: Command = Command("Q", description="Exit command")
|
|||||||
|
|
||||||
class App(BaseApp):
|
class App(BaseApp):
|
||||||
def __init__(
|
def __init__(
|
||||||
self, *,
|
self,
|
||||||
|
*,
|
||||||
prompt: str = "What do you want to do?\n\n",
|
prompt: str = "What do you want to do?\n\n",
|
||||||
initial_message: str = "Argenta\n",
|
initial_message: str = "Argenta\n",
|
||||||
farewell_message: str = "\nSee you\n",
|
farewell_message: str = "\nSee you\n",
|
||||||
@@ -395,12 +441,11 @@ class App(BaseApp):
|
|||||||
print_func=print_func,
|
print_func=print_func,
|
||||||
)
|
)
|
||||||
|
|
||||||
def run_polling(self, argspace: ArgSpace | None) -> None:
|
def run_polling(self) -> None:
|
||||||
"""
|
"""
|
||||||
Private. Starts the user input processing cycle
|
Private. Starts the user input processing cycle
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
self._argspace = argspace
|
|
||||||
self._pre_cycle_setup()
|
self._pre_cycle_setup()
|
||||||
while True:
|
while True:
|
||||||
if self._repeat_command_groups_description:
|
if self._repeat_command_groups_description:
|
||||||
@@ -409,7 +454,9 @@ class App(BaseApp):
|
|||||||
raw_command: str = Console().input(self._prompt)
|
raw_command: str = Console().input(self._prompt)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
|
input_command: InputCommand = InputCommand.parse(
|
||||||
|
raw_command=raw_command
|
||||||
|
)
|
||||||
except InputCommandException as error:
|
except InputCommandException as error:
|
||||||
with redirect_stdout(io.StringIO()) as stderr:
|
with redirect_stdout(io.StringIO()) as stderr:
|
||||||
self._error_handler(error, raw_command)
|
self._error_handler(error, raw_command)
|
||||||
@@ -419,7 +466,9 @@ class App(BaseApp):
|
|||||||
|
|
||||||
if self._is_exit_command(input_command):
|
if self._is_exit_command(input_command):
|
||||||
system_router.finds_appropriate_handler(input_command)
|
system_router.finds_appropriate_handler(input_command)
|
||||||
self._autocompleter.exit_setup(list(self._current_matching_triggers_with_routers.keys()))
|
self._autocompleter.exit_setup(
|
||||||
|
list(self._current_matching_triggers_with_routers.keys())
|
||||||
|
)
|
||||||
return
|
return
|
||||||
|
|
||||||
if self._is_unknown_command(input_command):
|
if self._is_unknown_command(input_command):
|
||||||
@@ -429,18 +478,40 @@ class App(BaseApp):
|
|||||||
self._print_framed_text(stdout_res)
|
self._print_framed_text(stdout_res)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
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:
|
if processing_router.disable_redirect_stdout:
|
||||||
if isinstance(self._dividing_line, StaticDividingLine):
|
if isinstance(self._dividing_line, StaticDividingLine):
|
||||||
self._print_func(self._dividing_line.get_full_static_line(is_override=self._override_system_messages))
|
self._print_func(
|
||||||
|
self._dividing_line.get_full_static_line(
|
||||||
|
is_override=self._override_system_messages
|
||||||
|
)
|
||||||
|
)
|
||||||
processing_router.finds_appropriate_handler(input_command)
|
processing_router.finds_appropriate_handler(input_command)
|
||||||
self._print_func(self._dividing_line.get_full_static_line(is_override=self._override_system_messages))
|
self._print_func(
|
||||||
|
self._dividing_line.get_full_static_line(
|
||||||
|
is_override=self._override_system_messages
|
||||||
|
)
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
dividing_line_unit_part: str = self._dividing_line.get_unit_part()
|
dividing_line_unit_part: str = self._dividing_line.get_unit_part()
|
||||||
self._print_func(StaticDividingLine(dividing_line_unit_part).get_full_static_line(is_override=self._override_system_messages))
|
self._print_func(
|
||||||
|
StaticDividingLine(
|
||||||
|
dividing_line_unit_part
|
||||||
|
).get_full_static_line(
|
||||||
|
is_override=self._override_system_messages
|
||||||
|
)
|
||||||
|
)
|
||||||
processing_router.finds_appropriate_handler(input_command)
|
processing_router.finds_appropriate_handler(input_command)
|
||||||
self._print_func(StaticDividingLine(dividing_line_unit_part).get_full_static_line(is_override=self._override_system_messages))
|
self._print_func(
|
||||||
|
StaticDividingLine(
|
||||||
|
dividing_line_unit_part
|
||||||
|
).get_full_static_line(
|
||||||
|
is_override=self._override_system_messages
|
||||||
|
)
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
with redirect_stdout(io.StringIO()) as stdout:
|
with redirect_stdout(io.StringIO()) as stdout:
|
||||||
processing_router.finds_appropriate_handler(input_command)
|
processing_router.finds_appropriate_handler(input_command)
|
||||||
@@ -455,7 +526,7 @@ class App(BaseApp):
|
|||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
router.command_register_ignore = self._ignore_command_register
|
router.command_register_ignore = self._ignore_command_register
|
||||||
self._registered_routers.add_registered_router(router)
|
self.registered_routers.add_registered_router(router)
|
||||||
|
|
||||||
def include_routers(self, *routers: Router) -> None:
|
def include_routers(self, *routers: Router) -> None:
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -0,0 +1,2 @@
|
|||||||
|
from argenta.di.integration import inject as inject
|
||||||
|
from argenta.di.integration import FromDishka as FromDishka
|
||||||
@@ -0,0 +1,45 @@
|
|||||||
|
__all__ = ["inject", "setup_dishka", "FromDishka"]
|
||||||
|
|
||||||
|
from typing import Any, Callable, TypeVar
|
||||||
|
|
||||||
|
from dishka import Container, FromDishka
|
||||||
|
from dishka.integrations.base import wrap_injection, is_dishka_injected
|
||||||
|
|
||||||
|
from argenta.response import Response
|
||||||
|
from argenta.app import App
|
||||||
|
|
||||||
|
|
||||||
|
T = TypeVar("T")
|
||||||
|
|
||||||
|
|
||||||
|
def inject(func: Callable[..., T]) -> Callable[..., T]:
|
||||||
|
return wrap_injection(
|
||||||
|
func=func,
|
||||||
|
is_async=False,
|
||||||
|
container_getter=_get_container_from_response,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def setup_dishka(app: App, *, auto_inject: bool = False) -> None:
|
||||||
|
if auto_inject:
|
||||||
|
_auto_inject_handlers(app)
|
||||||
|
|
||||||
|
|
||||||
|
def _get_container_from_response(
|
||||||
|
args: tuple[Any, ...], kwargs: dict[str, Any]
|
||||||
|
) -> Container:
|
||||||
|
for arg in args:
|
||||||
|
if isinstance(arg, Response):
|
||||||
|
if hasattr(arg, "_dishka_container"):
|
||||||
|
return arg._dishka_container # pyright: ignore[reportPrivateUsage]
|
||||||
|
break
|
||||||
|
|
||||||
|
raise RuntimeError("dishka container not found in Response")
|
||||||
|
|
||||||
|
|
||||||
|
def _auto_inject_handlers(app: App) -> None:
|
||||||
|
for router in app.registered_routers:
|
||||||
|
for command_handler in router.command_handlers:
|
||||||
|
if not is_dishka_injected(command_handler.handler_as_func):
|
||||||
|
injected_handler = inject(command_handler.handler_as_func)
|
||||||
|
command_handler.handler_as_func = injected_handler
|
||||||
@@ -0,0 +1,14 @@
|
|||||||
|
from argenta.orchestrator.argparser import ArgParser
|
||||||
|
from dishka import Provider, provide, Scope
|
||||||
|
|
||||||
|
from argenta.orchestrator.argparser.entity import ArgSpace
|
||||||
|
|
||||||
|
|
||||||
|
class SystemProvider(Provider):
|
||||||
|
def __init__(self, arg_parser: ArgParser):
|
||||||
|
super().__init__()
|
||||||
|
self._arg_parser: ArgParser = arg_parser
|
||||||
|
|
||||||
|
@provide(scope=Scope.APP)
|
||||||
|
def get_argspace(self) -> ArgSpace:
|
||||||
|
return self._arg_parser.parse_args()
|
||||||
@@ -1,19 +1,28 @@
|
|||||||
from argenta.app.models import App
|
from argenta.app import App
|
||||||
|
from argenta.response import Response
|
||||||
|
|
||||||
from argenta.orchestrator.argparser import ArgParser
|
from argenta.orchestrator.argparser import ArgParser
|
||||||
from argenta.orchestrator.argparser.entity import ArgSpace
|
from argenta.di.integration import setup_dishka
|
||||||
|
from argenta.di.providers import SystemProvider
|
||||||
|
|
||||||
|
from dishka import Provider, make_container
|
||||||
|
|
||||||
|
|
||||||
DEFAULT_ARGPARSER: ArgParser = ArgParser(processed_args=[])
|
DEFAULT_ARGPARSER: ArgParser = ArgParser(processed_args=[])
|
||||||
|
|
||||||
|
|
||||||
class Orchestrator:
|
class Orchestrator:
|
||||||
def __init__(self, arg_parser: ArgParser = DEFAULT_ARGPARSER):
|
def __init__(self, arg_parser: ArgParser = DEFAULT_ARGPARSER,
|
||||||
|
custom_providers: list[Provider] = [],
|
||||||
|
auto_inject_handlers: bool = True):
|
||||||
"""
|
"""
|
||||||
Public. An orchestrator and configurator that defines the behavior of an integrated system, one level higher than the App
|
Public. An orchestrator and configurator that defines the behavior of an integrated system, one level higher than the App
|
||||||
:param arg_parser: Cmd argument parser and configurator at startup
|
:param arg_parser: Cmd argument parser and configurator at startup
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
self._arg_parser: ArgParser = arg_parser
|
self._arg_parser: ArgParser = arg_parser
|
||||||
|
self._custom_providers: list[Provider] = custom_providers
|
||||||
|
self._auto_inject_handlers: bool = auto_inject_handlers
|
||||||
|
|
||||||
def start_polling(self, app: App) -> None:
|
def start_polling(self, app: App) -> None:
|
||||||
"""
|
"""
|
||||||
@@ -21,5 +30,8 @@ class Orchestrator:
|
|||||||
:param app: a running application
|
:param app: a running application
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
parsed_argspace: ArgSpace = self._arg_parser.parse_args()
|
container = make_container(SystemProvider(self._arg_parser), *self._custom_providers)
|
||||||
app.run_polling(argspace=parsed_argspace)
|
Response.patch_by_container(container)
|
||||||
|
setup_dishka(app, auto_inject=self._auto_inject_handlers)
|
||||||
|
|
||||||
|
app.run_polling()
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
from typing import Literal
|
from dishka import Container
|
||||||
|
|
||||||
from argenta.command.flag.flags.models import InputFlags
|
from argenta.command.flag.flags.models import InputFlags
|
||||||
from argenta.response.status import ResponseStatus
|
from argenta.response.status import ResponseStatus
|
||||||
|
|
||||||
@@ -7,7 +8,7 @@ EMPTY_INPUT_FLAGS: InputFlags = InputFlags()
|
|||||||
|
|
||||||
|
|
||||||
class Response:
|
class Response:
|
||||||
__slots__: tuple[Literal['status', 'input_flags'], ...] = ("status", "input_flags")
|
_dishka_container: Container
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@@ -21,3 +22,7 @@ class Response:
|
|||||||
"""
|
"""
|
||||||
self.status: ResponseStatus = status
|
self.status: ResponseStatus = status
|
||||||
self.input_flags: InputFlags = input_flags
|
self.input_flags: InputFlags = input_flags
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def patch_by_container(cls, container: Container) -> None:
|
||||||
|
cls._dishka_container = container
|
||||||
|
|||||||
@@ -6,13 +6,13 @@ from argenta.response import Response
|
|||||||
|
|
||||||
|
|
||||||
class CommandHandler:
|
class CommandHandler:
|
||||||
def __init__(self, handler_as_func: Callable[[Response], None], handled_command: Command):
|
def __init__(self, handler_as_func: Callable[..., None], handled_command: Command):
|
||||||
"""
|
"""
|
||||||
Private. Entity of the model linking the handler and the command being processed
|
Private. Entity of the model linking the handler and the command being processed
|
||||||
:param handler: the handler being called
|
:param handler: the handler being called
|
||||||
:param handled_command: the command being processed
|
:param handled_command: the command being processed
|
||||||
"""
|
"""
|
||||||
self.handler_as_func: Callable[[Response], None] = handler_as_func
|
self.handler_as_func: Callable[..., None] = handler_as_func
|
||||||
self.handled_command: Command = handled_command
|
self.handled_command: Command = handled_command
|
||||||
|
|
||||||
def handling(self, response: Response) -> None:
|
def handling(self, response: Response) -> None:
|
||||||
@@ -30,7 +30,9 @@ class CommandHandlers:
|
|||||||
Private. The model that unites all CommandHandler of the routers
|
Private. The model that unites all CommandHandler of the routers
|
||||||
:param command_handlers: list of CommandHandlers for register
|
:param command_handlers: list of CommandHandlers for register
|
||||||
"""
|
"""
|
||||||
self.command_handlers: list[CommandHandler] = command_handlers if command_handlers else []
|
self.command_handlers: list[CommandHandler] = (
|
||||||
|
command_handlers if command_handlers else []
|
||||||
|
)
|
||||||
|
|
||||||
def add_handler(self, command_handler: CommandHandler) -> None:
|
def add_handler(self, command_handler: CommandHandler) -> None:
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -6,25 +6,23 @@ from argenta.command import Command, InputCommand
|
|||||||
from argenta.command.flag import ValidationStatus
|
from argenta.command.flag import ValidationStatus
|
||||||
from argenta.response import Response, ResponseStatus
|
from argenta.response import Response, ResponseStatus
|
||||||
from argenta.router.command_handler.entity import CommandHandlers, CommandHandler
|
from argenta.router.command_handler.entity import CommandHandlers, CommandHandler
|
||||||
from argenta.command.flag.flags import (
|
from argenta.command.flag.flags import Flags, InputFlags
|
||||||
Flags,
|
|
||||||
InputFlags
|
|
||||||
)
|
|
||||||
from argenta.router.exceptions import (
|
from argenta.router.exceptions import (
|
||||||
RepeatedFlagNameException,
|
RepeatedFlagNameException,
|
||||||
TooManyTransferredArgsException,
|
|
||||||
RequiredArgumentNotPassedException,
|
RequiredArgumentNotPassedException,
|
||||||
TriggerContainSpacesException,
|
TriggerContainSpacesException,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
HandlerFunc: TypeAlias = Callable[[Response], None]
|
HandlerFunc: TypeAlias = Callable[..., None]
|
||||||
|
|
||||||
|
|
||||||
class Router:
|
class Router:
|
||||||
def __init__(
|
def __init__(
|
||||||
self, *, title: str | None = "Default title",
|
self,
|
||||||
disable_redirect_stdout: bool = False
|
*,
|
||||||
|
title: str | None = "Default title",
|
||||||
|
disable_redirect_stdout: bool = False,
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
Public. Directly configures and manages handlers
|
Public. Directly configures and manages handlers
|
||||||
@@ -58,7 +56,6 @@ class Router:
|
|||||||
def decorator(func: HandlerFunc) -> HandlerFunc:
|
def decorator(func: HandlerFunc) -> HandlerFunc:
|
||||||
_validate_func_args(func)
|
_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
|
||||||
|
|
||||||
return decorator
|
return decorator
|
||||||
@@ -91,7 +88,9 @@ 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 = _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)
|
||||||
@@ -102,7 +101,9 @@ class Router:
|
|||||||
for input_flag in input_command_flags:
|
for input_flag in input_command_flags:
|
||||||
input_flag.status = ValidationStatus.UNDEFINED
|
input_flag.status = ValidationStatus.UNDEFINED
|
||||||
undefined_flags.add_flag(input_flag)
|
undefined_flags.add_flag(input_flag)
|
||||||
response = Response(ResponseStatus.UNDEFINED_FLAGS, input_flags=undefined_flags)
|
response = Response(
|
||||||
|
ResponseStatus.UNDEFINED_FLAGS, input_flags=undefined_flags
|
||||||
|
)
|
||||||
command_handler.handling(response)
|
command_handler.handling(response)
|
||||||
else:
|
else:
|
||||||
response = Response(ResponseStatus.ALL_FLAGS_VALID)
|
response = Response(ResponseStatus.ALL_FLAGS_VALID)
|
||||||
@@ -137,14 +138,17 @@ class CommandDecorator:
|
|||||||
self.router: Router = router_instance
|
self.router: Router = router_instance
|
||||||
self.command: Command = command
|
self.command: Command = command
|
||||||
|
|
||||||
def __call__(self, handler_func: Callable[[Response], None]) -> Callable[[Response], None]:
|
def __call__(self, handler_func: Callable[..., None]) -> Callable[..., None]:
|
||||||
_validate_func_args(handler_func)
|
_validate_func_args(handler_func)
|
||||||
self.router.command_handlers.add_handler(CommandHandler(handler_func, self.command))
|
self.router.command_handlers.add_handler(
|
||||||
|
CommandHandler(handler_func, self.command)
|
||||||
|
)
|
||||||
return handler_func
|
return handler_func
|
||||||
|
|
||||||
|
|
||||||
def _structuring_input_flags(handled_command: Command,
|
def _structuring_input_flags(
|
||||||
input_flags: InputFlags) -> Response:
|
handled_command: Command, input_flags: InputFlags
|
||||||
|
) -> Response:
|
||||||
"""
|
"""
|
||||||
Private. Validates flags of input command
|
Private. Validates flags of input command
|
||||||
:param handled_command: entity of the handled command
|
:param handled_command: entity of the handled command
|
||||||
@@ -154,45 +158,42 @@ def _structuring_input_flags(handled_command: Command,
|
|||||||
invalid_value_flags, undefined_flags = False, False
|
invalid_value_flags, undefined_flags = False, False
|
||||||
|
|
||||||
for flag in input_flags:
|
for flag in input_flags:
|
||||||
flag_status: ValidationStatus = (handled_command.validate_input_flag(flag))
|
flag_status: ValidationStatus = handled_command.validate_input_flag(flag)
|
||||||
flag.status = flag_status
|
flag.status = flag_status
|
||||||
if flag_status == ValidationStatus.INVALID:
|
if flag_status == ValidationStatus.INVALID:
|
||||||
invalid_value_flags = True
|
invalid_value_flags = True
|
||||||
elif flag_status == ValidationStatus.UNDEFINED:
|
elif flag_status == ValidationStatus.UNDEFINED:
|
||||||
undefined_flags = True
|
undefined_flags = True
|
||||||
|
|
||||||
status = ResponseStatus.from_flags(has_invalid_value_flags=invalid_value_flags,
|
status = ResponseStatus.from_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[[Response], None]) -> None:
|
return Response(status=status, input_flags=input_flags)
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_func_args(func: Callable[..., None]) -> None:
|
||||||
"""
|
"""
|
||||||
Private. Validates the arguments of the handler
|
Private. Validates the arguments of the handler
|
||||||
:param func: entity of the handler func
|
:param func: entity of the handler func
|
||||||
:return: None if func is valid else raise exception
|
:return: None if func is valid else raise exception
|
||||||
"""
|
"""
|
||||||
transferred_args = getfullargspec(func).args
|
transferred_args = getfullargspec(func).args
|
||||||
if len(transferred_args) > 1:
|
if len(transferred_args) == 0:
|
||||||
raise TooManyTransferredArgsException()
|
|
||||||
elif len(transferred_args) == 0:
|
|
||||||
raise RequiredArgumentNotPassedException()
|
raise RequiredArgumentNotPassedException()
|
||||||
|
|
||||||
transferred_arg: str = transferred_args[0]
|
response_arg: str = transferred_args[0]
|
||||||
func_annotations: dict[str, None] = get_annotations(func)
|
func_annotations: dict[str, None] = get_annotations(func)
|
||||||
|
|
||||||
arg_annotation = func_annotations.get(transferred_arg)
|
response_arg_annotation = func_annotations.get(response_arg)
|
||||||
|
|
||||||
if arg_annotation is not None:
|
if response_arg_annotation is not None:
|
||||||
if arg_annotation is not Response:
|
if response_arg_annotation is not Response:
|
||||||
source_line: int = getsourcelines(func)[1]
|
source_line: int = getsourcelines(func)[1]
|
||||||
Console().print(
|
Console().print(
|
||||||
f'\nFile "{getsourcefile(func)}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint ' +
|
f'\nFile "{getsourcefile(func)}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
|
||||||
f"of argument([green]{transferred_arg}[/green]) passed to the handler must be [/i][bold blue]{Response}[/bold blue]," +
|
+ 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]{arg_annotation}[/bold blue] [i]is specified[/i]",
|
+ f" [i]but[/i] [bold blue]{response_arg_annotation}[/bold blue] [i]is specified[/i]",
|
||||||
highlight=False,
|
highlight=False,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -5,24 +5,17 @@ class RepeatedFlagNameException(Exception):
|
|||||||
"""
|
"""
|
||||||
Private. Raised when a repeated flag name is registered
|
Private. Raised when a repeated flag name is registered
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@override
|
@override
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return "Repeated registered flag names in register command"
|
return "Repeated registered flag names in register command"
|
||||||
|
|
||||||
|
|
||||||
class TooManyTransferredArgsException(Exception):
|
|
||||||
"""
|
|
||||||
Private. Raised when too many arguments are passed
|
|
||||||
"""
|
|
||||||
@override
|
|
||||||
def __str__(self) -> str:
|
|
||||||
return "Too many transferred arguments"
|
|
||||||
|
|
||||||
|
|
||||||
class RequiredArgumentNotPassedException(Exception):
|
class RequiredArgumentNotPassedException(Exception):
|
||||||
"""
|
"""
|
||||||
Private. Raised when a required argument is not passed
|
Private. Raised when a required argument is not passed
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@override
|
@override
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return "Required argument not passed"
|
return "Required argument not passed"
|
||||||
@@ -32,6 +25,7 @@ class TriggerContainSpacesException(Exception):
|
|||||||
"""
|
"""
|
||||||
Private. Raised when there is a space in the trigger being registered
|
Private. Raised when there is a space in the trigger being registered
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@override
|
@override
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return "Command trigger cannot contain spaces"
|
return "Command trigger cannot contain spaces"
|
||||||
|
|||||||
@@ -7,7 +7,6 @@ from argenta.command import Command
|
|||||||
from argenta.router.entity import _structuring_input_flags, _validate_command, _validate_func_args # pyright: ignore[reportPrivateUsage]
|
from argenta.router.entity import _structuring_input_flags, _validate_command, _validate_func_args # pyright: ignore[reportPrivateUsage]
|
||||||
from argenta.router.exceptions import (TriggerContainSpacesException,
|
from argenta.router.exceptions import (TriggerContainSpacesException,
|
||||||
RepeatedFlagNameException,
|
RepeatedFlagNameException,
|
||||||
TooManyTransferredArgsException,
|
|
||||||
RequiredArgumentNotPassedException)
|
RequiredArgumentNotPassedException)
|
||||||
|
|
||||||
import unittest
|
import unittest
|
||||||
@@ -79,12 +78,6 @@ class TestRouter(unittest.TestCase):
|
|||||||
with self.assertRaises(RequiredArgumentNotPassedException):
|
with self.assertRaises(RequiredArgumentNotPassedException):
|
||||||
_validate_func_args(handler) # pyright: ignore[reportArgumentType]
|
_validate_func_args(handler) # pyright: ignore[reportArgumentType]
|
||||||
|
|
||||||
def test_validate_incorrect_func_args2(self):
|
|
||||||
def handler(args, kwargs): # pyright: ignore[reportMissingParameterType, reportUnknownParameterType]
|
|
||||||
pass
|
|
||||||
with self.assertRaises(TooManyTransferredArgsException):
|
|
||||||
_validate_func_args(handler) # pyright: ignore[reportArgumentType]
|
|
||||||
|
|
||||||
def test_get_router_aliases(self):
|
def test_get_router_aliases(self):
|
||||||
router = Router()
|
router = Router()
|
||||||
@router.command(Command('some', aliases=['test', 'case']))
|
@router.command(Command('some', aliases=['test', 'case']))
|
||||||
@@ -108,12 +101,3 @@ class TestRouter(unittest.TestCase):
|
|||||||
def handler(response: Response): # pyright: ignore[reportUnusedFunction]
|
def handler(response: Response): # pyright: ignore[reportUnusedFunction]
|
||||||
pass
|
pass
|
||||||
self.assertListEqual(router.aliases, [])
|
self.assertListEqual(router.aliases, [])
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user