mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 10:05:28 +03:00
Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b4b7d5442c | |||
| 684760121c | |||
| dfc3c472ce | |||
| 8a3f742636 |
+105
-30
@@ -5,40 +5,61 @@ from .exceptions import (InvalidRouterInstanceException,
|
||||
OnlyOneMainRouterIsAllowedException,
|
||||
MissingMainRouterException,
|
||||
MissingHandlersForUnknownCommandsOnMainRouterException,
|
||||
HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException)
|
||||
HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException,
|
||||
NoRegisteredRoutersException,
|
||||
NoRegisteredHandlersException,
|
||||
RepeatedCommandInDifferentRoutersException)
|
||||
|
||||
|
||||
class App:
|
||||
def __init__(self,
|
||||
prompt: str = 'Enter a command',
|
||||
exit_command: str = 'q',
|
||||
ignore_exit_command_register: bool = True,
|
||||
initial_greeting: str = '\nHello, I am Argenta\n',
|
||||
farewell_message: str = 'GoodBye',
|
||||
farewell_message: str = '\nGoodBye\n',
|
||||
exit_command: str = 'Q',
|
||||
exit_command_description: str = 'Exit command',
|
||||
exit_command_title: str = 'System points:',
|
||||
ignore_exit_command_register: bool = True,
|
||||
ignore_command_register: bool = False,
|
||||
line_separate: str = '',
|
||||
command_group_description_separate: str = '',
|
||||
repeat_command_groups: bool = True,
|
||||
print_func: Callable[[str], None] = print) -> None:
|
||||
self.prompt = prompt
|
||||
self.print_func = print_func
|
||||
self.exit_command = exit_command
|
||||
self.exit_command_description = exit_command_description
|
||||
self.exit_command_title = exit_command_title
|
||||
self.ignore_exit_command_register = ignore_exit_command_register
|
||||
self.farewell_message = farewell_message
|
||||
self.initial_greeting = initial_greeting
|
||||
self.line_separate = line_separate
|
||||
self.command_group_description_separate = command_group_description_separate
|
||||
self.ignore_command_register = ignore_command_register
|
||||
self.repeat_command_groups = repeat_command_groups
|
||||
|
||||
self._routers: list[Router] = []
|
||||
self._registered_commands: list[dict[str, str | list[dict[str, Callable[[], None] | str]] | Router]] = []
|
||||
self._main_app_router: Router | None = None
|
||||
self._registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | str]] | Router]] = []
|
||||
self._app_main_router: Router | None = None
|
||||
self._description_message_pattern: str = '[{command}] *=*=* {description}'
|
||||
|
||||
|
||||
def start_polling(self) -> None:
|
||||
self.print_func(self.initial_greeting)
|
||||
self._validate_number_of_routers()
|
||||
self._validate_included_routers()
|
||||
self._validate_main_router()
|
||||
self._validate_all_router_commands()
|
||||
|
||||
self.print_func(self.initial_greeting)
|
||||
|
||||
if not self.repeat_command_groups:
|
||||
self._print_command_group_description()
|
||||
if self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
|
||||
while True:
|
||||
self._print_command_group_description()
|
||||
if self.repeat_command_groups:
|
||||
self._print_command_group_description()
|
||||
self.print_func(self.prompt)
|
||||
|
||||
command: str = input()
|
||||
@@ -47,12 +68,12 @@ class App:
|
||||
self.print_func(self.line_separate)
|
||||
|
||||
is_unknown_command: bool = self._check_is_command_unknown(command)
|
||||
|
||||
if is_unknown_command:
|
||||
continue
|
||||
|
||||
for router in self._routers:
|
||||
router.input_command_handler(command)
|
||||
|
||||
self.print_func(self.line_separate)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
|
||||
@@ -74,18 +95,62 @@ class App:
|
||||
self._description_message_pattern: str = pattern
|
||||
|
||||
|
||||
def _validate_main_router(self):
|
||||
if not self._main_app_router:
|
||||
raise MissingMainRouterException()
|
||||
def get_main_router(self) -> Router:
|
||||
return self._app_main_router
|
||||
|
||||
if not self._main_app_router.unknown_command_func:
|
||||
|
||||
def get_all_app_commands(self) -> list[str]:
|
||||
all_commands: list[str] = []
|
||||
for router in self._routers:
|
||||
all_commands.extend(router.get_all_commands())
|
||||
|
||||
return all_commands
|
||||
|
||||
|
||||
def _validate_number_of_routers(self) -> None:
|
||||
if not self._routers:
|
||||
raise NoRegisteredRoutersException()
|
||||
|
||||
|
||||
def _validate_included_routers(self) -> None:
|
||||
for router in self._routers:
|
||||
if not router.get_command_entities():
|
||||
raise NoRegisteredHandlersException(router.get_name())
|
||||
|
||||
|
||||
def _validate_main_router(self):
|
||||
if not self._app_main_router:
|
||||
if len(self._routers) > 1:
|
||||
raise MissingMainRouterException()
|
||||
else:
|
||||
router = self._routers[0]
|
||||
router.set_router_as_main()
|
||||
self._app_main_router = router
|
||||
|
||||
if not self._app_main_router.unknown_command_func:
|
||||
raise MissingHandlersForUnknownCommandsOnMainRouterException()
|
||||
|
||||
for router in self._routers:
|
||||
if router.unknown_command_func and self._main_app_router is not router:
|
||||
if router.unknown_command_func and self._app_main_router is not router:
|
||||
raise HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException()
|
||||
|
||||
|
||||
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 _checking_command_for_exit_command(self, command: str):
|
||||
if command.lower() == self.exit_command.lower():
|
||||
if self.ignore_exit_command_register:
|
||||
@@ -98,25 +163,25 @@ class App:
|
||||
|
||||
|
||||
def _check_is_command_unknown(self, command: str):
|
||||
registered_commands: list[dict[str, str | list[dict[str, Callable[[], None] | str]] | Router]] = self._registered_commands
|
||||
for router in registered_commands:
|
||||
for command_entity in router['commands']:
|
||||
registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | str]] | Router]] = self._registered_router_entities
|
||||
for router_entity in registered_router_entities:
|
||||
for command_entity in router_entity['commands']:
|
||||
if command_entity['command'].lower() == command.lower():
|
||||
if router['router'].ignore_command_register:
|
||||
if self.ignore_command_register:
|
||||
return False
|
||||
else:
|
||||
if command_entity['command'] == command:
|
||||
return False
|
||||
self._main_app_router.unknown_command_handler(command)
|
||||
self._app_main_router.unknown_command_handler(command)
|
||||
self.print_func(self.line_separate)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
return True
|
||||
|
||||
|
||||
def _print_command_group_description(self):
|
||||
for router in self._registered_commands:
|
||||
self.print_func(router['name'])
|
||||
for command_entity in router['commands']:
|
||||
for router_entity in self._registered_router_entities:
|
||||
self.print_func(router_entity['title'])
|
||||
for command_entity in router_entity['commands']:
|
||||
self.print_func(self._description_message_pattern.format(
|
||||
command=command_entity['command'],
|
||||
description=command_entity['description']
|
||||
@@ -124,22 +189,32 @@ class App:
|
||||
)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
|
||||
self.print_func(self.exit_command_title)
|
||||
self.print_func(self._description_message_pattern.format(
|
||||
command=self.exit_command,
|
||||
description=self.exit_command_description
|
||||
)
|
||||
)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
|
||||
def include_router(self, router: Router, is_main: bool = False) -> None:
|
||||
|
||||
def include_router(self, router: Router, is_main: True | False = False) -> None:
|
||||
if not isinstance(router, Router):
|
||||
raise InvalidRouterInstanceException()
|
||||
|
||||
if is_main:
|
||||
if not self._main_app_router:
|
||||
self._main_app_router = router
|
||||
if not self._app_main_router:
|
||||
self._app_main_router = router
|
||||
router.set_router_as_main()
|
||||
else:
|
||||
raise OnlyOneMainRouterIsAllowedException(router)
|
||||
raise OnlyOneMainRouterIsAllowedException(self._app_main_router.get_name())
|
||||
|
||||
router.set_ignore_command_register(self.ignore_command_register)
|
||||
self._routers.append(router)
|
||||
|
||||
registered_commands: list[dict[str, Callable[[], None] | str]] = router.get_registered_commands()
|
||||
self._registered_commands.append({'name': router.get_name(),
|
||||
'router': router,
|
||||
'commands': registered_commands})
|
||||
command_entities: list[dict[str, Callable[[], None] | str]] = router.get_command_entities()
|
||||
self._registered_router_entities.append({'name': router.get_name(),
|
||||
'title': router.get_title(),
|
||||
'entity': router,
|
||||
'commands': command_entities})
|
||||
|
||||
|
||||
@@ -37,3 +37,20 @@ class MissingHandlersForUnknownCommandsOnMainRouterException(Exception):
|
||||
class HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException(Exception):
|
||||
def __str__(self):
|
||||
return '\nThe handler for unknown commands can only be declared for the main router'
|
||||
|
||||
|
||||
class NoRegisteredRoutersException(Exception):
|
||||
def __str__(self):
|
||||
return "No Registered Router Found"
|
||||
|
||||
|
||||
class NoRegisteredHandlersException(Exception):
|
||||
def __init__(self, router_name):
|
||||
self.router_name = router_name
|
||||
def __str__(self):
|
||||
return f"No Registered Handlers Found For '{self.router_name}'"
|
||||
|
||||
|
||||
class RepeatedCommandInDifferentRoutersException(Exception):
|
||||
def __str__(self):
|
||||
return "Commands in different handlers cannot be repeated"
|
||||
|
||||
+76
-24
@@ -1,36 +1,37 @@
|
||||
from typing import Callable, Any
|
||||
from ..router.exceptions import (InvalidCommandInstanceException,
|
||||
UnknownCommandHandlerHasAlreadyBeenCreatedException,
|
||||
InvalidDescriptionInstanceException)
|
||||
InvalidDescriptionInstanceException,
|
||||
RepeatedCommandException)
|
||||
|
||||
|
||||
class Router:
|
||||
def __init__(self,
|
||||
name: str,
|
||||
ignore_command_register: bool = False):
|
||||
title: str = 'Commands group title:',
|
||||
name: str = 'subordinate'):
|
||||
|
||||
self.ignore_command_register = ignore_command_register
|
||||
self.title = title
|
||||
self.name = name
|
||||
|
||||
self._processed_commands: list[dict[str, Callable[[], None] | str]] = []
|
||||
self._command_entities: list[dict[str, Callable[[], None] | str]] = []
|
||||
self.unknown_command_func: Callable[[str], None] | None = None
|
||||
self._is_main_router: bool = False
|
||||
self.ignore_command_register: bool = False
|
||||
|
||||
|
||||
def command(self, command: str, description: str) -> Callable[[Any], Any]:
|
||||
if not isinstance(command, str):
|
||||
raise InvalidCommandInstanceException()
|
||||
if not isinstance(description, str):
|
||||
raise InvalidDescriptionInstanceException()
|
||||
else:
|
||||
def command_decorator(func):
|
||||
self._processed_commands.append({'func': func,
|
||||
'command': command,
|
||||
'description': description})
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
return wrapper
|
||||
return command_decorator
|
||||
def command(self, command: str, description: str = None) -> Callable[[Any], Any]:
|
||||
processed_description = Router._validate_description(command, description)
|
||||
self._validate_command(command)
|
||||
|
||||
def command_decorator(func):
|
||||
self._command_entities.append({'handler_func': func,
|
||||
'command': command,
|
||||
'description': processed_description})
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
return wrapper
|
||||
|
||||
return command_decorator
|
||||
|
||||
|
||||
def unknown_command(self, func):
|
||||
@@ -45,27 +46,78 @@ class Router:
|
||||
|
||||
|
||||
def input_command_handler(self, input_command):
|
||||
for command_entity in self._processed_commands:
|
||||
for command_entity in self._command_entities:
|
||||
if input_command.lower() == command_entity['command'].lower():
|
||||
if self.ignore_command_register:
|
||||
return command_entity['func']()
|
||||
return command_entity['handler_func']()
|
||||
else:
|
||||
if input_command == command_entity['command']:
|
||||
return command_entity['func']()
|
||||
return command_entity['handler_func']()
|
||||
|
||||
|
||||
def unknown_command_handler(self, unknown_command):
|
||||
self.unknown_command_func(unknown_command)
|
||||
|
||||
|
||||
def _validate_command(self, command: str):
|
||||
if not isinstance(command, str):
|
||||
raise InvalidCommandInstanceException()
|
||||
if command in self.get_all_commands():
|
||||
raise RepeatedCommandException()
|
||||
if self.ignore_command_register:
|
||||
if command.lower() in [x.lower() for x in self.get_all_commands()]:
|
||||
raise RepeatedCommandException()
|
||||
|
||||
|
||||
@staticmethod
|
||||
def _validate_description(command: str, description: str):
|
||||
if not isinstance(description, str):
|
||||
if description is None:
|
||||
description = f'description for "{command}" command'
|
||||
else:
|
||||
raise InvalidDescriptionInstanceException()
|
||||
return description
|
||||
|
||||
|
||||
def set_router_as_main(self):
|
||||
if self.name == 'subordinate':
|
||||
self.name = 'main'
|
||||
self._is_main_router = True
|
||||
|
||||
|
||||
def get_registered_commands(self) -> list[dict[str, Callable[[], None] | str]]:
|
||||
return self._processed_commands
|
||||
def set_ignore_command_register(self, ignore_command_register: bool):
|
||||
self.ignore_command_register = ignore_command_register
|
||||
|
||||
|
||||
def get_command_entities(self) -> list[dict[str, Callable[[], None] | str]]:
|
||||
return self._command_entities
|
||||
|
||||
|
||||
def get_name(self) -> str:
|
||||
return self.name
|
||||
|
||||
|
||||
def get_title(self) -> str:
|
||||
return self.title
|
||||
|
||||
|
||||
def get_router_info(self) -> dict:
|
||||
return {
|
||||
'title': self.title,
|
||||
'name': self.name,
|
||||
'ignore_command_register': self.ignore_command_register,
|
||||
'attributes': {
|
||||
'command_entities': self._command_entities,
|
||||
'unknown_command_func': self.unknown_command_func,
|
||||
'is_main_router': self._is_main_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'])
|
||||
|
||||
return all_commands
|
||||
|
||||
@@ -11,3 +11,8 @@ class InvalidDescriptionInstanceException(Exception):
|
||||
class UnknownCommandHandlerHasAlreadyBeenCreatedException(Exception):
|
||||
def __str__(self):
|
||||
return "Only one unknown command handler can be declared"
|
||||
|
||||
|
||||
class RepeatedCommandException(Exception):
|
||||
def __str__(self):
|
||||
return "Commands in handler cannot be repeated"
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "argenta"
|
||||
version = "0.1.3"
|
||||
version = "0.2.1"
|
||||
description = "python library for creating cli apps"
|
||||
authors = [
|
||||
{name = "kolo",email = "kolo.is.main@gmail.com"}
|
||||
|
||||
@@ -5,7 +5,7 @@ with open("README.md", "r", encoding="utf-8") as fh:
|
||||
|
||||
setup(
|
||||
name="argenta",
|
||||
version="0.1.3",
|
||||
version="0.2.1",
|
||||
author="kolo",
|
||||
author_email="kolo.is.main@gmail.com",
|
||||
description="Python library for creating CLI apps",
|
||||
|
||||
@@ -6,10 +6,8 @@ from ..handlers.handlers_implementation.solving_command import start_solving_com
|
||||
from ..handlers.handlers_implementation.upgrade_command import upgrade_command
|
||||
|
||||
|
||||
work_router: Router = Router(name='Work points:',
|
||||
ignore_command_register=False)
|
||||
settings_router: Router = Router(name='Settings points:',
|
||||
ignore_command_register=True)
|
||||
work_router: Router = Router(title='Work points:')
|
||||
settings_router: Router = Router(title='Settings points:')
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
@@ -7,7 +7,8 @@ from rich.console import Console
|
||||
app: App = App(prompt='[italic white bold]What do you want to do(enter number of action)?',
|
||||
line_separate='[bold green]\n---------------------------------------------\n',
|
||||
print_func=Console().print,
|
||||
command_group_description_separate='')
|
||||
command_group_description_separate='',
|
||||
repeat_command_groups=False)
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
@@ -1,28 +1,27 @@
|
||||
from pprint import pprint
|
||||
from rich.console import Console
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
work_router: Router = Router(name='Work points:',
|
||||
ignore_command_register=False)
|
||||
settings_router: Router = Router(name='Settings points:',
|
||||
ignore_command_register=True)
|
||||
work_router: Router = Router(name='work')
|
||||
settings_router: Router = Router(name='settings')
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
@work_router.command(command='0', description='Get Help')
|
||||
@work_router.command(command='a')
|
||||
def command_help():
|
||||
print('help command')
|
||||
console.print('[bold red]command help [/bold red]')
|
||||
|
||||
|
||||
@work_router.command(command='1', description='Start Solving')
|
||||
@work_router.command(command='B', description='tester')
|
||||
def command_start_solving():
|
||||
print('start solving')
|
||||
console.print('[bold red]command start [/bold red]')
|
||||
|
||||
|
||||
@settings_router.command(command='U', description='Update WordMath')
|
||||
def command_update():
|
||||
print('update wordmath')
|
||||
@settings_router.command(command='b')
|
||||
def command_settings():
|
||||
console.print('[bold red]command settings [/bold red]')
|
||||
|
||||
|
||||
@work_router.unknown_command
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
from pprint import pprint
|
||||
from tests.mock_default_app.handlers.routers import work_router, settings_router
|
||||
from argenta.app.entity import App
|
||||
from art import text2art
|
||||
|
||||
|
||||
app: App = App()
|
||||
app: App = App(ignore_command_register=False,
|
||||
line_separate='\n-------------------------------\n')
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
Reference in New Issue
Block a user