new models, logic refactor, renaminf constants and other

This commit is contained in:
2025-03-28 00:54:12 +03:00
parent 4256d67789
commit 1ec8ea53b4
8 changed files with 126 additions and 81 deletions
+23 -43
View File
@@ -8,12 +8,12 @@ from argenta.router.defaults import system_router
from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException)
from .exceptions import (InvalidRouterInstanceException,
from argenta.app.exceptions import (InvalidRouterInstanceException,
InvalidDescriptionMessagePatternException,
NoRegisteredRoutersException,
NoRegisteredHandlersException,
RepeatedCommandInDifferentRoutersException,
IncorrectNumberOfHandlerArgsException)
from argenta.app.models import RegisteredRouters
class App:
@@ -47,9 +47,8 @@ class App:
self.repeat_command_groups = repeat_command_groups
self.messages_on_startup = messages_on_startup if messages_on_startup else []
self._routers: list[Router] = []
self._description_message_pattern: str = '[{command}] *=*=* {description}'
self._registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | Command]] | Router]] = []
self._registered_routers: RegisteredRouters = RegisteredRouters()
self._invalid_input_flags_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(f'Empty input command')
@@ -61,7 +60,6 @@ class App:
self._setup_system_router()
self._validate_number_of_routers()
self._validate_included_routers()
self._validate_all_router_commands()
self.print_func(self.initial_message)
@@ -122,8 +120,8 @@ class App:
self.print_func(self.prompt)
continue
for router in self._routers:
router.input_command_handler(input_command)
for registered_router in self._registered_routers:
registered_router.input_command_handler(input_command)
self.print_func(self.line_separate)
self.print_func(self.command_group_description_separate)
@@ -198,49 +196,31 @@ class App:
raise InvalidRouterInstanceException()
router.set_ignore_command_register(self.ignore_command_register)
self._routers.append(router)
self._registered_routers.add_registered_router(router)
command_entities: list[dict[str, Callable[[], None] | Command]] = router.get_command_entities()
self._registered_router_entities.append({'name': router.get_name(),
'title': router.get_title(),
'entity': router,
'commands': command_entities})
def include_routers(self, *routers: Router) -> None:
for router in routers:
self.include_router(router)
def _validate_number_of_routers(self) -> None:
if not self._routers:
if not self._registered_routers:
raise NoRegisteredRoutersException()
def _validate_included_routers(self) -> None:
for router in self._routers:
if not router.get_command_entities():
for router in self._registered_routers:
if not router.get_command_handlers():
raise NoRegisteredHandlersException(router.get_name())
def _validate_all_router_commands(self) -> None:
for idx in range(len(self._registered_router_entities)):
current_router: Router = self._registered_router_entities[idx]['entity']
routers_without_current_router = self._registered_router_entities.copy()
routers_without_current_router.pop(idx)
current_router_all_commands: list[str] = current_router.get_all_commands()
for router_entity in routers_without_current_router:
if len(set(current_router_all_commands).intersection(set(router_entity['entity'].get_all_commands()))) > 0:
raise RepeatedCommandInDifferentRoutersException()
if self.ignore_command_register:
if len(set([x.lower() for x in current_router_all_commands]).intersection(set([x.lower() for x in router_entity['entity'].get_all_commands()]))) > 0:
raise RepeatedCommandInDifferentRoutersException()
def _setup_system_router(self):
system_router.set_title(self.system_points_title)
@system_router.command(Command(self.exit_command, self.exit_command_description))
def exit_command():
self._exit_command_handler()
if system_router not in [router['entity'] for router in self._registered_router_entities]:
if system_router not in self._registered_routers.get_registered_routers():
self.include_router(system_router)
@@ -257,26 +237,26 @@ class App:
def _check_is_command_unknown(self, command: Command):
registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | Command]] | Router]] = self._registered_router_entities
for router_entity in registered_router_entities:
for command_entity in router_entity['commands']:
if command_entity['command'].get_trigger().lower() == command.get_trigger().lower():
for router_entity in self._registered_routers:
for command_handler in router_entity.get_command_handlers():
handled_command_trigger = command_handler.get_handled_command().get_trigger()
if handled_command_trigger.lower() == command.get_trigger().lower():
if self.ignore_command_register:
return False
else:
if command_entity['command'].get_trigger() == command.get_trigger():
if handled_command_trigger == command.get_trigger():
return False
self._unknown_command_handler(command)
return True
def _print_command_group_description(self):
for router_entity in self._registered_router_entities:
self.print_func(router_entity['title'])
for command_entity in router_entity['commands']:
for registered_router in self._registered_routers:
self.print_func(registered_router.get_title())
for command_handler in registered_router.get_command_handlers():
self.print_func(self._description_message_pattern.format(
command=command_entity['command'].get_trigger(),
description=command_entity['command'].get_description()
command=command_handler.get_handled_command().get_trigger(),
description=command_handler.get_handled_command().get_description()
)
)
self.print_func(self.command_group_description_separate)
-5
View File
@@ -25,11 +25,6 @@ class NoRegisteredHandlersException(Exception):
return f"No Registered Handlers Found For '{self.router_name}'"
class RepeatedCommandInDifferentRoutersException(Exception):
def __str__(self):
return "Commands in different handlers cannot be repeated"
class IncorrectNumberOfHandlerArgsException(Exception):
def __str__(self):
return "Incorrect Input Flags Handler has incorrect number of arguments"
+21
View File
@@ -0,0 +1,21 @@
from argenta.router import Router
class RegisteredRouters:
def __init__(self, registered_routers: list[Router] = None) -> None:
self._registered_routers = registered_routers if registered_routers else []
def get_registered_routers(self) -> list[Router]:
return self._registered_routers
def add_registered_router(self, router: Router):
self._registered_routers.append(router)
def add_registered_routers(self, *routers: Router):
self._registered_routers.extend(routers)
def __iter__(self):
return iter(self._registered_routers)
def __next__(self):
return next(iter(self._registered_routers))
+10 -10
View File
@@ -5,17 +5,17 @@ import re
@dataclass
class DefaultFlags:
help_flag = Flag(flag_name='help', possible_flag_values=False)
short_help_flag = Flag(flag_name='h', flag_prefix='-', possible_flag_values=False)
HELP = Flag(flag_name='help', possible_flag_values=False)
SHORT_HELP = Flag(flag_name='h', flag_prefix='-', possible_flag_values=False)
info_flag = Flag(flag_name='info', possible_flag_values=False)
short_info_flag = Flag(flag_name='i', flag_prefix='-', possible_flag_values=False)
INFO = Flag(flag_name='info', possible_flag_values=False)
SHORT_INFO = Flag(flag_name='i', flag_prefix='-', possible_flag_values=False)
all_flag = Flag(flag_name='all', possible_flag_values=False)
short_all_flag = Flag(flag_name='a', flag_prefix='-', possible_flag_values=False)
ALL = Flag(flag_name='all', possible_flag_values=False)
SHORT_ALL = Flag(flag_name='a', flag_prefix='-', possible_flag_values=False)
host_flag = Flag(flag_name='host', possible_flag_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
short_host_flag = Flag(flag_name='h', flag_prefix='-', possible_flag_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
HOST = Flag(flag_name='host', possible_flag_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
SHORT_HOST = Flag(flag_name='h', flag_prefix='-', possible_flag_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
port_flag = Flag(flag_name='port', possible_flag_values=re.compile(r'^\d{1,5}$'))
short_port_flag = Flag(flag_name='p', flag_prefix='-', possible_flag_values=re.compile(r'^\d{1,5}$'))
PORT = Flag(flag_name='port', possible_flag_values=re.compile(r'^\d{1,5}$'))
SHORT_PORT = Flag(flag_name='p', flag_prefix='-', possible_flag_values=re.compile(r'^\d{1,5}$'))
+20 -18
View File
@@ -1,10 +1,10 @@
from typing import Callable, Any
from inspect import getfullargspec
from ..command.entity import Command
from argenta.command.flag.entity import Flag
from argenta.command.flag.flags_group import FlagsGroup
from ..router.exceptions import (RepeatedCommandException,
from argenta.command import Command
from argenta.router.models import CommandHandler, CommandHandlers
from argenta.command.flag import Flag, FlagsGroup
from argenta.router.exceptions import (RepeatedCommandException,
RepeatedFlagNameException,
TooManyTransferredArgsException,
RequiredArgumentNotPassedException,
@@ -20,7 +20,7 @@ class Router:
self._title = title
self._name = name
self._command_entities: list[dict[str, Callable[[], None] | Command]] = []
self._command_handlers: CommandHandlers = CommandHandlers()
self._ignore_command_register: bool = False
self._not_valid_flag_handler: Callable[[Flag], None] = lambda flag: print(f"Undefined or incorrect input flag: {flag.get_string_entity()}{(' '+flag.get_value()) if flag.get_value() else ''}")
@@ -31,14 +31,15 @@ class Router:
def command_decorator(func):
Router._validate_func_args(command, func)
self._command_entities.append({'handler_func': func,
'command': command})
self._command_handlers.add_command_handler(CommandHandler(func, command))
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return command_decorator
def set_invalid_input_flag_handler(self, func):
processed_args = getfullargspec(func).args
if len(processed_args) != 1:
@@ -50,24 +51,25 @@ class Router:
def input_command_handler(self, input_command: Command):
input_command_name: str = input_command.get_trigger()
input_command_flags: FlagsGroup = input_command.get_input_flags()
for command_entity in self._command_entities:
if input_command_name.lower() == command_entity['command'].get_trigger().lower():
if command_entity['command'].get_registered_flags():
for command_handler in self._command_handlers:
handle_command = command_handler.get_handled_command()
if input_command_name.lower() == handle_command.get_trigger().lower():
if handle_command.get_registered_flags():
if input_command_flags:
for flag in input_command_flags:
is_valid = command_entity['command'].validate_input_flag(flag)
is_valid = handle_command.validate_input_flag(flag)
if not is_valid:
self._not_valid_flag_handler(flag)
return
return command_entity['handler_func'](input_command_flags.unparse_to_dict())
return command_handler.handling(input_command_flags.unparse_to_dict())
else:
return command_entity['handler_func']({})
return command_handler.handling({})
else:
if input_command_flags:
self._not_valid_flag_handler(input_command_flags[0])
return
else:
return command_entity['handler_func']()
return command_handler.handling()
def _validate_command(self, command: Command):
@@ -104,8 +106,8 @@ class Router:
self._ignore_command_register = ignore_command_register
def get_command_entities(self) -> list[dict[str, Callable[[], None] | Command]]:
return self._command_entities
def get_command_handlers(self) -> CommandHandlers:
return self._command_handlers
def get_name(self) -> str:
@@ -122,7 +124,7 @@ class Router:
def get_all_commands(self) -> list[str]:
all_commands: list[str] = []
for command_entity in self._command_entities:
all_commands.append(command_entity['command'].get_trigger())
for command_handler in self._command_handlers:
all_commands.append(command_handler.get_handled_command().get_trigger())
return all_commands
+49
View File
@@ -0,0 +1,49 @@
from typing import Callable
from argenta.command import Command
class CommandHandler:
def __init__(self, handler: Callable[[], None] | Callable[[dict], None], handled_command: Command):
self.handler = handler
self.handled_command = handled_command
def handling(self, input_flags: dict = None):
if input_flags is not None:
self.handler(input_flags)
else:
self.handler()
def get_handler(self):
return self.handler
def get_handled_command(self):
return self.handled_command
class CommandHandlers:
def __init__(self, command_handlers: list[CommandHandler] = None):
self.command_handlers = command_handlers if command_handlers else []
def get_command_handlers(self) -> list[CommandHandler]:
return self.command_handlers
def add_command_handler(self, command_handler: CommandHandler):
self.command_handlers.append(command_handler)
def add_command_handlers(self, *command_handlers: CommandHandler):
self.command_handlers.extend(command_handlers)
def __iter__(self):
return iter(self.command_handlers)
def __next__(self):
return next(iter(self.command_handlers))