mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
Compare commits
13 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c8e0729be8 | |||
| c2bbc5f15d | |||
| 0acbf54e44 | |||
| c3d9541330 | |||
| f6561de9b3 | |||
| bebd84969b | |||
| 365347ea7f | |||
| 33cb528b1d | |||
| fd287c5da0 | |||
| 45f410e3e8 | |||
| 8b06e9cd39 | |||
| c38fe10006 | |||
| 03cbc64f48 |
@@ -0,0 +1,65 @@
|
|||||||
|
# Argenta
|
||||||
|
|
||||||
|
### Bibliothek zum Erstellen modularer CLI-Anwendungen
|
||||||
|
|
||||||
|

|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
# Installation
|
||||||
|
```bash
|
||||||
|
pip install argenta
|
||||||
|
```
|
||||||
|
or
|
||||||
|
```bash
|
||||||
|
poetry add argenta
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
# Schnellstart
|
||||||
|
|
||||||
|
Ein Beispiel für eine einfache Anwendung
|
||||||
|
```python
|
||||||
|
# routers.py
|
||||||
|
from argenta.router import Router
|
||||||
|
from argenta.command import Command
|
||||||
|
from argenta.response import Response
|
||||||
|
|
||||||
|
|
||||||
|
router = Router()
|
||||||
|
|
||||||
|
@router.command(Command("hello"))
|
||||||
|
def handler(response: Response):
|
||||||
|
print("Hello, world!")
|
||||||
|
```
|
||||||
|
|
||||||
|
```python
|
||||||
|
# main.py
|
||||||
|
from argenta.app import App
|
||||||
|
from argenta.orchestrator import Orchestrator
|
||||||
|
from routers import router
|
||||||
|
|
||||||
|
app: App = App()
|
||||||
|
orchestrator: Orchestrator = Orchestrator()
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
main()
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
# Funktionen in der Entwicklung
|
||||||
|
|
||||||
|
- Vollständige Unterstützung für Autocompleter unter Linux
|
||||||
|
|
||||||
|
## Vollständige [Dokumentation](https://argenta-docs.vercel.app) | MIT 2025 kolo | made by [kolo](https://t.me/kolo_id)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@@ -1,12 +1,14 @@
|
|||||||
# Argenta
|
# Argenta
|
||||||
|
|
||||||
### Библиотека для создания модульных CLI приложeний
|
### Library for creating modular CLI applications
|
||||||
|
|
||||||
|
#### RU - [README.ru.md](https://github.com/koloideal/Argenta/blob/kolo/README.ru.md) • DE - [README.de.md](https://github.com/koloideal/Argenta/blob/kolo/README.de.md)
|
||||||
|
|
||||||

|

|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
# Установка
|
# Installing
|
||||||
```bash
|
```bash
|
||||||
pip install argenta
|
pip install argenta
|
||||||
```
|
```
|
||||||
@@ -17,9 +19,9 @@ poetry add argenta
|
|||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
# Быстрый старт
|
# Quick start
|
||||||
|
|
||||||
Пример простейшего приложения
|
An example of a simple application
|
||||||
```python
|
```python
|
||||||
# routers.py
|
# routers.py
|
||||||
from argenta.router import Router
|
from argenta.router import Router
|
||||||
@@ -55,12 +57,11 @@ if __name__ == '__main__':
|
|||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
# Фичи в разработке
|
# Features in development
|
||||||
|
|
||||||
- Полноценная поддержка автокомплитера на Linux
|
- Full support for autocompleter on Linux
|
||||||
- Возможность настройки захвата stdout при обработке хэндлером ввода
|
|
||||||
|
|
||||||
## Полная [документация](https://argenta-docs.vercel.app) | MIT 2025 kolo | made by [kolo](https://t.me/kolo_id)
|
## Full [docs](https://argenta-docs.vercel.app) | MIT 2025 kolo | made by [kolo](https://t.me/kolo_id)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,65 @@
|
|||||||
|
# Argenta
|
||||||
|
|
||||||
|
### Библиотека для создания модульных CLI приложeний
|
||||||
|
|
||||||
|

|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
# Установка
|
||||||
|
```bash
|
||||||
|
pip install argenta
|
||||||
|
```
|
||||||
|
or
|
||||||
|
```bash
|
||||||
|
poetry add argenta
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
# Быстрый старт
|
||||||
|
|
||||||
|
Пример простейшего приложения
|
||||||
|
```python
|
||||||
|
# routers.py
|
||||||
|
from argenta.router import Router
|
||||||
|
from argenta.command import Command
|
||||||
|
from argenta.response import Response
|
||||||
|
|
||||||
|
|
||||||
|
router = Router()
|
||||||
|
|
||||||
|
@router.command(Command("hello"))
|
||||||
|
def handler(response: Response):
|
||||||
|
print("Hello, world!")
|
||||||
|
```
|
||||||
|
|
||||||
|
```python
|
||||||
|
# main.py
|
||||||
|
from argenta.app import App
|
||||||
|
from argenta.orchestrator import Orchestrator
|
||||||
|
from routers import router
|
||||||
|
|
||||||
|
app: App = App()
|
||||||
|
orchestrator: Orchestrator = Orchestrator()
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
main()
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
# Фичи в разработке
|
||||||
|
|
||||||
|
- Полноценная поддержка автокомплитера на Linux
|
||||||
|
|
||||||
|
## Полная [документация](https://argenta-docs.vercel.app) | MIT 2025 kolo | made by [kolo](https://t.me/kolo_id)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@@ -0,0 +1,36 @@
|
|||||||
|
from argenta.command import Command
|
||||||
|
from argenta.metrics import get_time_of_pre_cycle_setup
|
||||||
|
from argenta.response import Response
|
||||||
|
from argenta.router import Router
|
||||||
|
from argenta.app import App
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
class TimeOfPreCycleSetup:
|
||||||
|
@staticmethod
|
||||||
|
def commands_with_two_aliases(num_of_commands: int):
|
||||||
|
router = Router()
|
||||||
|
|
||||||
|
for i in range(num_of_commands):
|
||||||
|
@router.command(Command(f'cmd{i}', aliases=[f'cdr{i}', f'prt{i}']))
|
||||||
|
def handler(response: Response):
|
||||||
|
pass
|
||||||
|
|
||||||
|
app = App()
|
||||||
|
app.include_router(router)
|
||||||
|
|
||||||
|
return get_time_of_pre_cycle_setup(app)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def commands_with_one_aliases(num_of_commands: int):
|
||||||
|
router = Router()
|
||||||
|
|
||||||
|
for i in range(num_of_commands):
|
||||||
|
@router.command(Command(f'cmd{i}', aliases=[f'cdr{i}']))
|
||||||
|
def handler(response: Response):
|
||||||
|
pass
|
||||||
|
|
||||||
|
app = App()
|
||||||
|
app.include_router(router)
|
||||||
|
|
||||||
|
return get_time_of_pre_cycle_setup(app)
|
||||||
@@ -1,27 +0,0 @@
|
|||||||
from mock.mock_app.handlers.routers import work_router
|
|
||||||
|
|
||||||
from argenta.app import App
|
|
||||||
from argenta.app.defaults import PredefinedMessages
|
|
||||||
from argenta.app.autocompleter import AutoCompleter
|
|
||||||
from argenta.orchestrator import Orchestrator
|
|
||||||
from argenta.orchestrator.argparser import ArgParser
|
|
||||||
from argenta.orchestrator.argparser.arguments import BooleanArgument
|
|
||||||
|
|
||||||
|
|
||||||
arg_parser = ArgParser(processed_args=[BooleanArgument("repeat")])
|
|
||||||
app: App = App(autocompleter=AutoCompleter(".hist"))
|
|
||||||
orchestrator: Orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
|
|
||||||
def main():
|
|
||||||
app.include_router(work_router)
|
|
||||||
|
|
||||||
app.add_message_on_startup(PredefinedMessages.USAGE)
|
|
||||||
app.add_message_on_startup(PredefinedMessages.AUTOCOMPLETE)
|
|
||||||
app.add_message_on_startup(PredefinedMessages.HELP)
|
|
||||||
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
main()
|
|
||||||
+11
-82
@@ -1,89 +1,18 @@
|
|||||||
from argenta.router import Router
|
|
||||||
from argenta.command import Command
|
|
||||||
from argenta.response import Response
|
|
||||||
from argenta.response.status import Status
|
|
||||||
from argenta.command.flag import Flag
|
|
||||||
from argenta.command.flags import Flags
|
|
||||||
from argenta.app import App
|
from argenta.app import App
|
||||||
|
from argenta.command import Command
|
||||||
from argenta.orchestrator import Orchestrator
|
from argenta.orchestrator import Orchestrator
|
||||||
|
from argenta.router import Router
|
||||||
# Создание маршрутизатора
|
|
||||||
file_router = Router("Операции с файлами")
|
|
||||||
|
|
||||||
# Определение флагов для команды копирования
|
|
||||||
copy_flags = Flags(
|
|
||||||
Flag('source', '--'),
|
|
||||||
Flag('destination', '--'),
|
|
||||||
Flag('recursive', '--', False), # Булевый флаг без значения
|
|
||||||
Flag('force', '-', False) # Короткий булевый флаг
|
|
||||||
)
|
|
||||||
@file_router.command(Command('case', aliases=['cp', 'ch']))
|
|
||||||
def handler(response: Response):
|
|
||||||
print('test')
|
|
||||||
|
|
||||||
# Регистрация команды копирования
|
|
||||||
@file_router.command(Command(
|
|
||||||
trigger="ch",
|
|
||||||
description="Копирование файлов",
|
|
||||||
flags=copy_flags,
|
|
||||||
aliases=["cp"]
|
|
||||||
))
|
|
||||||
def copy_files(response: Response):
|
|
||||||
# Получаем значения корректных флагов
|
|
||||||
source = None
|
|
||||||
destination = None
|
|
||||||
recursive = False
|
|
||||||
force = False
|
|
||||||
|
|
||||||
for flag in response.valid_flags:
|
|
||||||
if flag.get_name() == "source":
|
|
||||||
source = flag.get_value()
|
|
||||||
elif flag.get_name() == "destination":
|
|
||||||
destination = flag.get_value()
|
|
||||||
elif flag.get_name() == "recursive":
|
|
||||||
recursive = True
|
|
||||||
elif flag.get_name() == "force":
|
|
||||||
force = True
|
|
||||||
|
|
||||||
# Проверка обязательных параметров
|
|
||||||
if not source or not destination:
|
|
||||||
print("Ошибка: необходимо указать источник и назначение")
|
|
||||||
return
|
|
||||||
|
|
||||||
print(f"Копирование из {source} в {destination}")
|
|
||||||
if recursive:
|
|
||||||
print("Рекурсивное копирование включено")
|
|
||||||
if force:
|
|
||||||
print("Принудительное копирование включено")
|
|
||||||
|
|
||||||
# Обработка неопределенных флагов
|
|
||||||
if response.undefined_flags:
|
|
||||||
print("\nПредупреждение: обнаружены незарегистрированные флаги:")
|
|
||||||
for flag in response.undefined_flags:
|
|
||||||
print(f" - {flag.get_name()}" +
|
|
||||||
(f" = {flag.get_value()}" if flag.get_value() else ""))
|
|
||||||
|
|
||||||
# Обработка флагов с некорректными значениями
|
|
||||||
if response.invalid_value_flags:
|
|
||||||
print("\nПредупреждение: обнаружены флаги с некорректными значениями:")
|
|
||||||
for flag in response.invalid_value_flags:
|
|
||||||
print(f" - {flag.get_name()} = {flag.get_value()}")
|
|
||||||
|
|
||||||
# Принятие решения на основе статуса
|
|
||||||
if response.status != Status.ALL_FLAGS_VALID:
|
|
||||||
print("\nВыполнение с предупреждениями из-за проблем с флагами.")
|
|
||||||
|
|
||||||
|
|
||||||
|
router = Router()
|
||||||
app = App()
|
|
||||||
app.include_router(file_router)
|
|
||||||
orchestrator = Orchestrator()
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
|
@router.command(Command('test'))
|
||||||
|
def test(response):
|
||||||
|
print('test command')
|
||||||
|
|
||||||
|
app = App(ignore_command_register=True,
|
||||||
|
override_system_messages=True,
|
||||||
|
print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
orchestrator.start_polling(app)
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,12 +0,0 @@
|
|||||||
from rich.console import Console
|
|
||||||
|
|
||||||
|
|
||||||
console = Console()
|
|
||||||
|
|
||||||
|
|
||||||
def help_command():
|
|
||||||
console.print(
|
|
||||||
"[italic bold]The main functionality of the script is to convert an expression from a string "
|
|
||||||
"to a mathematical one and then calculate this expression. "
|
|
||||||
"Project GitHub: https://github.com/koloideal/WordMath[/italic bold]"
|
|
||||||
)
|
|
||||||
@@ -1,4 +1,4 @@
|
|||||||
from mock.mock_app.handlers.routers import work_router
|
from mock.mock_app.routers import work_router
|
||||||
|
|
||||||
from argenta.app import App
|
from argenta.app import App
|
||||||
from argenta.app.defaults import PredefinedMessages
|
from argenta.app.defaults import PredefinedMessages
|
||||||
|
|||||||
@@ -2,15 +2,17 @@ from rich.console import Console
|
|||||||
|
|
||||||
from argenta.command import Command
|
from argenta.command import Command
|
||||||
from argenta.command.flag.defaults import PredefinedFlags
|
from argenta.command.flag.defaults import PredefinedFlags
|
||||||
from argenta.command.flags import Flags
|
from argenta.command.flag import Flags, Flag, PossibleValues
|
||||||
from argenta.response import Response
|
from argenta.response import Response
|
||||||
from argenta.router import Router
|
from argenta.router import Router
|
||||||
|
|
||||||
|
|
||||||
work_router: Router = Router(title="Work points:")
|
work_router: Router = Router(title="Work points:", disable_redirect_stdout=True)
|
||||||
|
|
||||||
console = Console()
|
console = Console()
|
||||||
|
|
||||||
|
flag = Flag('csdv', possible_values=PossibleValues.DISABLE)
|
||||||
|
|
||||||
|
|
||||||
@work_router.command(
|
@work_router.command(
|
||||||
Command(
|
Command(
|
||||||
@@ -21,6 +23,8 @@ console = Console()
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
def command_help(response: Response):
|
def command_help(response: Response):
|
||||||
|
case = input("test > ")
|
||||||
|
print(case)
|
||||||
print(response.status)
|
print(response.status)
|
||||||
print(response.undefined_flags.get_flags())
|
print(response.undefined_flags.get_flags())
|
||||||
print(response.valid_flags.get_flags())
|
print(response.valid_flags.get_flags())
|
||||||
+6
-1
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "argenta"
|
name = "argenta"
|
||||||
version = "1.0.3"
|
version = "1.0.7"
|
||||||
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.8"
|
requires-python = ">=3.8"
|
||||||
@@ -26,3 +26,8 @@ exclude = [
|
|||||||
requires = ["hatchling"]
|
requires = ["hatchling"]
|
||||||
build-backend = "hatchling.build"
|
build-backend = "hatchling.build"
|
||||||
|
|
||||||
|
[dependency-groups]
|
||||||
|
dev = [
|
||||||
|
"psutil>=7.0.0",
|
||||||
|
]
|
||||||
|
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ from typing import Never
|
|||||||
|
|
||||||
class AutoCompleter:
|
class AutoCompleter:
|
||||||
def __init__(
|
def __init__(
|
||||||
self, history_filename: str = False, autocomplete_button: str = "tab"
|
self, history_filename: str | None = None, autocomplete_button: str = "tab"
|
||||||
) -> None:
|
) -> None:
|
||||||
"""
|
"""
|
||||||
Public. Configures and implements auto-completion of input command
|
Public. Configures and implements auto-completion of input command
|
||||||
|
|||||||
@@ -1,10 +1,11 @@
|
|||||||
from enum import Enum
|
from enum import StrEnum
|
||||||
|
|
||||||
|
|
||||||
class PredefinedMessages(Enum):
|
class PredefinedMessages(StrEnum):
|
||||||
"""
|
"""
|
||||||
Public. A dataclass with predetermined messages for quick use
|
Public. A dataclass with predetermined messages for quick use
|
||||||
"""
|
"""
|
||||||
|
|
||||||
USAGE = "[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]"
|
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]"
|
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>"
|
AUTOCOMPLETE = "[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>"
|
||||||
|
|||||||
+106
-97
@@ -21,9 +21,9 @@ from argenta.app.registered_routers.entity import RegisteredRouters
|
|||||||
from argenta.response import Response
|
from argenta.response import Response
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
class BaseApp:
|
class BaseApp:
|
||||||
def __init__(self,
|
def __init__(self, prompt: str,
|
||||||
prompt: str,
|
|
||||||
initial_message: str,
|
initial_message: str,
|
||||||
farewell_message: str,
|
farewell_message: str,
|
||||||
exit_command: Command,
|
exit_command: Command,
|
||||||
@@ -47,18 +47,33 @@ class BaseApp:
|
|||||||
self._farewell_message = farewell_message
|
self._farewell_message = farewell_message
|
||||||
self._initial_message = initial_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._registered_routers: RegisteredRouters = RegisteredRouters()
|
||||||
self._messages_on_startup: list[str] = []
|
self._messages_on_startup: list[str] = []
|
||||||
|
|
||||||
self._all_registered_triggers_in_lower_case: list[str] = []
|
self._matching_lower_triggers_with_routers: dict[str, Router] = {}
|
||||||
self._all_registered_triggers_in_default_case: list[str] = []
|
self._matching_default_triggers_with_routers: dict[str, Router] = {}
|
||||||
|
|
||||||
self._incorrect_input_syntax_handler: Callable[[str], None] = (lambda raw_command: print_func(f"Incorrect flag syntax: {raw_command}"))
|
if self._ignore_command_register:
|
||||||
self._repeated_input_flags_handler: Callable[[str], None] = (lambda raw_command: print_func(f"Repeated input flags: {raw_command}"))
|
self._current_matching_triggers_with_routers: dict[str, Router] = self._matching_lower_triggers_with_routers
|
||||||
self._empty_input_command_handler: Callable[[], None] = lambda: print_func("Empty input command")
|
else:
|
||||||
self._unknown_command_handler: Callable[[InputCommand], None] = (lambda command: print_func(f"Unknown command: {command.get_trigger()}"))
|
self._current_matching_triggers_with_routers: dict[str, Router] = self._matching_default_triggers_with_routers
|
||||||
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:
|
def set_description_message_pattern(self, _: Callable[[str, str], str]) -> None:
|
||||||
"""
|
"""
|
||||||
@@ -194,10 +209,10 @@ class BaseApp:
|
|||||||
"""
|
"""
|
||||||
input_command_trigger = command.get_trigger()
|
input_command_trigger = command.get_trigger()
|
||||||
if self._ignore_command_register:
|
if self._ignore_command_register:
|
||||||
if input_command_trigger.lower() in self._all_registered_triggers_in_lower_case:
|
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 self._all_registered_triggers_in_default_case:
|
if input_command_trigger in list(self._current_matching_triggers_with_routers.keys()):
|
||||||
return False
|
return False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
@@ -210,13 +225,12 @@ class BaseApp:
|
|||||||
:param raw_command: the raw input command
|
:param raw_command: the raw input command
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
match error:
|
if isinstance(error, UnprocessedInputFlagException):
|
||||||
case UnprocessedInputFlagException():
|
self._incorrect_input_syntax_handler(raw_command)
|
||||||
self._incorrect_input_syntax_handler(raw_command)
|
elif isinstance(error, RepeatedInputFlagsException):
|
||||||
case RepeatedInputFlagsException():
|
self._repeated_input_flags_handler(raw_command)
|
||||||
self._repeated_input_flags_handler(raw_command)
|
elif isinstance(error, EmptyInputCommandException):
|
||||||
case EmptyInputCommandException():
|
self._empty_input_command_handler()
|
||||||
self._empty_input_command_handler()
|
|
||||||
|
|
||||||
def _setup_system_router(self) -> None:
|
def _setup_system_router(self) -> None:
|
||||||
"""
|
"""
|
||||||
@@ -234,11 +248,8 @@ class BaseApp:
|
|||||||
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 = (
|
all_commands = list(self._current_matching_triggers_with_routers.keys())
|
||||||
self._all_registered_triggers_in_lower_case
|
|
||||||
if self._ignore_command_register
|
|
||||||
else self._all_registered_triggers_in_default_case
|
|
||||||
)
|
|
||||||
matches: list[str] | list = sorted(
|
matches: list[str] | list = sorted(
|
||||||
cmd for cmd in all_commands if cmd.startswith(unknown_command)
|
cmd for cmd in all_commands if cmd.startswith(unknown_command)
|
||||||
)
|
)
|
||||||
@@ -258,35 +269,27 @@ class BaseApp:
|
|||||||
Private. Sets up default app view
|
Private. Sets up default app view
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
self._prompt = "[italic dim bold]What do you want to do?\n"
|
self._prompt = f"[italic dim bold]{self._prompt}"
|
||||||
self._initial_message = (
|
self._initial_message = ("\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n")
|
||||||
f"\n[bold red]{text2art(self._initial_message, font='tarty1')}\n"
|
|
||||||
)
|
|
||||||
self._farewell_message = (
|
self._farewell_message = (
|
||||||
f"[bold red]\n{text2art(f'\n{self._farewell_message}\n', font='chanky')}[/bold red]\n"
|
"[bold red]\n\n"
|
||||||
f"[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\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: (
|
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(
|
self._incorrect_input_syntax_handler = lambda raw_command: self._print_func(f"[red bold]Incorrect flag syntax: {escape(raw_command)}")
|
||||||
f"[red bold]Incorrect flag syntax: {escape(raw_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")
|
||||||
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.get_trigger()
|
cmd_trg: str = command.get_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 = (
|
first_part_of_text = f"[red]Unknown command:[/red] [blue]{escape(cmd_trg)}[/blue]"
|
||||||
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
|
||||||
@@ -296,7 +299,7 @@ class BaseApp:
|
|||||||
|
|
||||||
self._unknown_command_handler = unknown_command_handler
|
self._unknown_command_handler = unknown_command_handler
|
||||||
|
|
||||||
def _pre_cycle_setup(self) -> None:
|
def pre_cycle_setup(self) -> None:
|
||||||
"""
|
"""
|
||||||
Private. Configures various aspects of the application before the start of the cycle
|
Private. Configures various aspects of the application before the start of the cycle
|
||||||
:return: None
|
:return: None
|
||||||
@@ -304,22 +307,22 @@ class BaseApp:
|
|||||||
self._setup_system_router()
|
self._setup_system_router()
|
||||||
|
|
||||||
for router_entity in self._registered_routers:
|
for router_entity in self._registered_routers:
|
||||||
self._all_registered_triggers_in_default_case.extend(router_entity.get_triggers())
|
router_triggers = router_entity.get_triggers()
|
||||||
self._all_registered_triggers_in_default_case.extend(router_entity.get_aliases())
|
router_aliases = router_entity.get_aliases()
|
||||||
|
combined = router_triggers + router_aliases
|
||||||
|
|
||||||
self._all_registered_triggers_in_lower_case.extend([x.lower() for x in router_entity.get_triggers()])
|
for trigger in combined:
|
||||||
self._all_registered_triggers_in_lower_case.extend([x.lower() for x in router_entity.get_aliases()])
|
self._matching_default_triggers_with_routers[trigger] = router_entity
|
||||||
|
self._matching_lower_triggers_with_routers[trigger.lower()] = router_entity
|
||||||
|
|
||||||
self._autocompleter.initial_setup(self._all_registered_triggers_in_lower_case)
|
self._autocompleter.initial_setup(list(self._current_matching_triggers_with_routers.keys()))
|
||||||
|
|
||||||
if self._ignore_command_register:
|
seen = {}
|
||||||
for cmd in set(self._all_registered_triggers_in_lower_case):
|
for item in list(self._current_matching_triggers_with_routers.keys()):
|
||||||
if self._all_registered_triggers_in_lower_case.count(cmd) != 1:
|
if item in seen:
|
||||||
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{cmd}[/b blue]")
|
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{item}[/b blue]")
|
||||||
else:
|
else:
|
||||||
for cmd in set(self._all_registered_triggers_in_default_case):
|
seen[item] = True
|
||||||
if self._all_registered_triggers_in_default_case.count(cmd) != 1:
|
|
||||||
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{cmd}[/b blue]")
|
|
||||||
|
|
||||||
if not self._override_system_messages:
|
if not self._override_system_messages:
|
||||||
self._setup_default_view()
|
self._setup_default_view()
|
||||||
@@ -330,24 +333,25 @@ class BaseApp:
|
|||||||
self._print_func(message)
|
self._print_func(message)
|
||||||
if self._messages_on_startup:
|
if self._messages_on_startup:
|
||||||
print("\n")
|
print("\n")
|
||||||
|
|
||||||
if not self._repeat_command_groups_description:
|
if not self._repeat_command_groups_description:
|
||||||
self._print_command_group_description()
|
self._print_command_group_description()
|
||||||
|
|
||||||
|
|
||||||
class App(BaseApp):
|
class App(BaseApp):
|
||||||
def __init__(self,
|
def __init__(
|
||||||
prompt: str = "What do you want to do?\n",
|
self,
|
||||||
initial_message: str = "Argenta\n",
|
prompt: str = "What do you want to do?\n",
|
||||||
farewell_message: str = "\nSee you\n",
|
initial_message: str = "Argenta\n",
|
||||||
exit_command: Command = Command("Q", "Exit command"),
|
farewell_message: str = "\nSee you\n",
|
||||||
system_router_title: str | None = "System points:",
|
exit_command: Command = Command("Q", "Exit command"),
|
||||||
ignore_command_register: bool = True,
|
system_router_title: str | None = "System points:",
|
||||||
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
|
ignore_command_register: bool = True,
|
||||||
repeat_command_groups: bool = True,
|
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
|
||||||
override_system_messages: bool = False,
|
repeat_command_groups: bool = True,
|
||||||
autocompleter: AutoCompleter = AutoCompleter(),
|
override_system_messages: bool = False,
|
||||||
print_func: Callable[[str], None] = Console().print) -> None:
|
autocompleter: AutoCompleter = AutoCompleter(),
|
||||||
|
print_func: Callable[[str], None] = Console().print,
|
||||||
|
) -> None:
|
||||||
"""
|
"""
|
||||||
Public. The essence of the application itself.
|
Public. The essence of the application itself.
|
||||||
Configures and manages all aspects of the behavior and presentation of the user interacting with the user
|
Configures and manages all aspects of the behavior and presentation of the user interacting with the user
|
||||||
@@ -364,24 +368,26 @@ class App(BaseApp):
|
|||||||
:param print_func: system messages text output function
|
:param print_func: system messages text output function
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
super().__init__(prompt=prompt,
|
super().__init__(
|
||||||
initial_message=initial_message,
|
prompt=prompt,
|
||||||
farewell_message=farewell_message,
|
initial_message=initial_message,
|
||||||
exit_command=exit_command,
|
farewell_message=farewell_message,
|
||||||
system_router_title=system_router_title,
|
exit_command=exit_command,
|
||||||
ignore_command_register=ignore_command_register,
|
system_router_title=system_router_title,
|
||||||
dividing_line=dividing_line,
|
ignore_command_register=ignore_command_register,
|
||||||
repeat_command_groups=repeat_command_groups,
|
dividing_line=dividing_line,
|
||||||
override_system_messages=override_system_messages,
|
repeat_command_groups=repeat_command_groups,
|
||||||
autocompleter=autocompleter,
|
override_system_messages=override_system_messages,
|
||||||
print_func=print_func)
|
autocompleter=autocompleter,
|
||||||
|
print_func=print_func,
|
||||||
|
)
|
||||||
|
|
||||||
def run_polling(self) -> 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._pre_cycle_setup()
|
self.pre_cycle_setup()
|
||||||
while True:
|
while True:
|
||||||
if self._repeat_command_groups_description:
|
if self._repeat_command_groups_description:
|
||||||
self._print_command_group_description()
|
self._print_command_group_description()
|
||||||
@@ -389,9 +395,7 @@ class App(BaseApp):
|
|||||||
raw_command: str = Console().input(self._prompt)
|
raw_command: str = Console().input(self._prompt)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
input_command: InputCommand = InputCommand.parse(
|
input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
|
||||||
raw_command=raw_command
|
|
||||||
)
|
|
||||||
except BaseInputCommandException as error:
|
except BaseInputCommandException as error:
|
||||||
with redirect_stdout(io.StringIO()) as f:
|
with redirect_stdout(io.StringIO()) as f:
|
||||||
self._error_handler(error, raw_command)
|
self._error_handler(error, raw_command)
|
||||||
@@ -401,14 +405,7 @@ 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)
|
||||||
if self._ignore_command_register:
|
self._autocompleter.exit_setup(list(self._current_matching_triggers_with_routers.keys()))
|
||||||
self._autocompleter.exit_setup(
|
|
||||||
self._all_registered_triggers_in_lower_case
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
self._autocompleter.exit_setup(
|
|
||||||
self._all_registered_triggers_in_default_case
|
|
||||||
)
|
|
||||||
return
|
return
|
||||||
|
|
||||||
if self._is_unknown_command(input_command):
|
if self._is_unknown_command(input_command):
|
||||||
@@ -418,11 +415,23 @@ class App(BaseApp):
|
|||||||
self._print_framed_text(res)
|
self._print_framed_text(res)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
with redirect_stdout(io.StringIO()) as f:
|
processing_router = self._current_matching_triggers_with_routers[input_command.get_trigger().lower()]
|
||||||
for registered_router in self._registered_routers:
|
|
||||||
registered_router.finds_appropriate_handler(input_command)
|
if processing_router.disable_redirect_stdout:
|
||||||
res: str = f.getvalue()
|
if isinstance(self._dividing_line, StaticDividingLine):
|
||||||
self._print_framed_text(res)
|
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
|
||||||
|
processing_router.finds_appropriate_handler(input_command)
|
||||||
|
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))
|
||||||
|
processing_router.finds_appropriate_handler(input_command)
|
||||||
|
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:
|
||||||
|
processing_router.finds_appropriate_handler(input_command)
|
||||||
|
res: str = f.getvalue()
|
||||||
|
if res:
|
||||||
|
self._print_framed_text(res)
|
||||||
|
|
||||||
def include_router(self, router: Router) -> None:
|
def include_router(self, router: Router) -> None:
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ from argenta.router import Router
|
|||||||
|
|
||||||
|
|
||||||
class RegisteredRouters:
|
class RegisteredRouters:
|
||||||
def __init__(self, registered_routers: list[Router] = None) -> None:
|
def __init__(self, registered_routers: list[Router] | None = None) -> None:
|
||||||
"""
|
"""
|
||||||
Private. Combines registered routers
|
Private. Combines registered routers
|
||||||
:param registered_routers: list of the registered routers
|
:param registered_routers: list of the registered routers
|
||||||
|
|||||||
@@ -1,4 +1,17 @@
|
|||||||
__all__ = ["Flag", "InputFlag"]
|
__all__ = [
|
||||||
|
"Flag",
|
||||||
|
"InputFlag",
|
||||||
|
"UndefinedInputFlags",
|
||||||
|
"ValidInputFlags",
|
||||||
|
"InvalidValueInputFlags",
|
||||||
|
"Flags", "PossibleValues"
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
from argenta.command.flag.models import Flag, InputFlag
|
from argenta.command.flag.models import Flag, InputFlag, PossibleValues
|
||||||
|
from argenta.command.flag.flags.models import (
|
||||||
|
UndefinedInputFlags,
|
||||||
|
ValidInputFlags,
|
||||||
|
Flags,
|
||||||
|
InvalidValueInputFlags,
|
||||||
|
)
|
||||||
|
|||||||
@@ -1,22 +1,23 @@
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from argenta.command.flag.models import Flag
|
from argenta.command.flag.models import Flag, PossibleValues
|
||||||
import re
|
import re
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class PredefinedFlags:
|
class PredefinedFlags:
|
||||||
"""
|
"""
|
||||||
Public. A dataclass with predefined flags and most frequently used flags for quick use
|
Public. A dataclass with predefined flags and most frequently used flags for quick use
|
||||||
"""
|
"""
|
||||||
|
|
||||||
HELP = Flag(name="help", possible_values=False)
|
HELP = Flag(name="help", possible_values=PossibleValues.DISABLE)
|
||||||
SHORT_HELP = Flag(name="H", prefix="-", possible_values=False)
|
SHORT_HELP = Flag(name="H", prefix="-", possible_values=PossibleValues.DISABLE)
|
||||||
|
|
||||||
INFO = Flag(name="info", possible_values=False)
|
INFO = Flag(name="info", possible_values=PossibleValues.DISABLE)
|
||||||
SHORT_INFO = Flag(name="I", prefix="-", possible_values=False)
|
SHORT_INFO = Flag(name="I", prefix="-", possible_values=PossibleValues.DISABLE)
|
||||||
|
|
||||||
ALL = Flag(name="all", possible_values=False)
|
ALL = Flag(name="all", possible_values=PossibleValues.DISABLE)
|
||||||
SHORT_ALL = Flag(name="A", prefix="-", possible_values=False)
|
SHORT_ALL = Flag(name="A", prefix="-", possible_values=PossibleValues.DISABLE)
|
||||||
|
|
||||||
HOST = Flag(
|
HOST = Flag(
|
||||||
name="host", possible_values=re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
|
name="host", possible_values=re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ __all__ = [
|
|||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
from argenta.command.flags.models import (
|
from argenta.command.flag.flags.models import (
|
||||||
Flags,
|
Flags,
|
||||||
InputFlags,
|
InputFlags,
|
||||||
UndefinedInputFlags,
|
UndefinedInputFlags,
|
||||||
@@ -1,6 +1,16 @@
|
|||||||
|
from enum import Enum
|
||||||
from typing import Literal, Pattern
|
from typing import Literal, Pattern
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
class PossibleValues(Enum):
|
||||||
|
DISABLE: Literal[False] = False
|
||||||
|
ALL: Literal[True] = True
|
||||||
|
|
||||||
|
def __eq__(self, other: bool) -> bool:
|
||||||
|
return self.value == other
|
||||||
|
|
||||||
|
|
||||||
class BaseFlag:
|
class BaseFlag:
|
||||||
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--") -> None:
|
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--") -> None:
|
||||||
"""
|
"""
|
||||||
@@ -43,7 +53,7 @@ class Flag(BaseFlag):
|
|||||||
self,
|
self,
|
||||||
name: str,
|
name: str,
|
||||||
prefix: Literal["-", "--", "---"] = "--",
|
prefix: Literal["-", "--", "---"] = "--",
|
||||||
possible_values: list[str] | Pattern[str] | False = True,
|
possible_values: list[str] | Pattern[str] | PossibleValues = PossibleValues.ALL,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""
|
"""
|
||||||
Public. The entity of the flag being registered for subsequent processing
|
Public. The entity of the flag being registered for subsequent processing
|
||||||
@@ -61,7 +71,7 @@ class Flag(BaseFlag):
|
|||||||
:param input_flag_value: The input flag value to validate
|
:param input_flag_value: The input flag value to validate
|
||||||
:return: whether the entered flag is valid as bool
|
:return: whether the entered flag is valid as bool
|
||||||
"""
|
"""
|
||||||
if self.possible_values is False:
|
if self.possible_values == PossibleValues.DISABLE:
|
||||||
if input_flag_value is None:
|
if input_flag_value is None:
|
||||||
return True
|
return True
|
||||||
else:
|
else:
|
||||||
@@ -87,7 +97,10 @@ class Flag(BaseFlag):
|
|||||||
|
|
||||||
class InputFlag(BaseFlag):
|
class InputFlag(BaseFlag):
|
||||||
def __init__(
|
def __init__(
|
||||||
self, name: str, prefix: Literal["-", "--", "---"] = "--", value: str = None
|
self,
|
||||||
|
name: str,
|
||||||
|
prefix: Literal["-", "--", "---"] = "--",
|
||||||
|
value: str | None = None,
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
Public. The entity of the flag of the entered command
|
Public. The entity of the flag of the entered command
|
||||||
|
|||||||
@@ -1,14 +1,11 @@
|
|||||||
from argenta.command.flag.models import Flag, InputFlag
|
from argenta.command.flag.models import Flag, InputFlag
|
||||||
from argenta.command.flags.models import InputFlags, Flags
|
from argenta.command.flag.flags.models import InputFlags, Flags
|
||||||
from argenta.command.exceptions import (
|
from argenta.command.exceptions import (
|
||||||
UnprocessedInputFlagException,
|
UnprocessedInputFlagException,
|
||||||
RepeatedInputFlagsException,
|
RepeatedInputFlagsException,
|
||||||
EmptyInputCommandException,
|
EmptyInputCommandException,
|
||||||
)
|
)
|
||||||
from typing import Generic, TypeVar, cast, Literal
|
from typing import cast, Literal
|
||||||
|
|
||||||
|
|
||||||
InputCommandType = TypeVar("InputCommandType")
|
|
||||||
|
|
||||||
|
|
||||||
class BaseCommand:
|
class BaseCommand:
|
||||||
@@ -31,9 +28,9 @@ class Command(BaseCommand):
|
|||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
trigger: str,
|
trigger: str,
|
||||||
description: str = None,
|
description: str | None = None,
|
||||||
flags: Flag | Flags = None,
|
flags: Flag | Flags | None = None,
|
||||||
aliases: list[str] = None,
|
aliases: list[str] | None = None,
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
Public. The command that can and should be registered in the Router
|
Public. The command that can and should be registered in the Router
|
||||||
@@ -94,6 +91,7 @@ class Command(BaseCommand):
|
|||||||
is_valid = registered_flag.validate_input_flag_value(
|
is_valid = registered_flag.validate_input_flag_value(
|
||||||
flag.get_value()
|
flag.get_value()
|
||||||
)
|
)
|
||||||
|
|
||||||
if is_valid:
|
if is_valid:
|
||||||
return "Valid"
|
return "Valid"
|
||||||
else:
|
else:
|
||||||
@@ -109,8 +107,8 @@ class Command(BaseCommand):
|
|||||||
return self._description
|
return self._description
|
||||||
|
|
||||||
|
|
||||||
class InputCommand(BaseCommand, Generic[InputCommandType]):
|
class InputCommand(BaseCommand):
|
||||||
def __init__(self, trigger: str, input_flags: InputFlag | InputFlags = None):
|
def __init__(self, trigger: str, input_flags: InputFlag | InputFlags | None = None):
|
||||||
"""
|
"""
|
||||||
Private. The model of the input command, after parsing
|
Private. The model of the input command, after parsing
|
||||||
:param trigger:the trigger of the command
|
:param trigger:the trigger of the command
|
||||||
@@ -142,7 +140,7 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
|||||||
return self._input_flags
|
return self._input_flags
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def parse(raw_command: str) -> InputCommandType:
|
def parse(raw_command: str) -> "InputCommand":
|
||||||
"""
|
"""
|
||||||
Private. Parse the raw input command
|
Private. Parse the raw input command
|
||||||
:param raw_command: raw input command
|
:param raw_command: raw input command
|
||||||
|
|||||||
@@ -0,0 +1,4 @@
|
|||||||
|
__all__ = ["get_time_of_pre_cycle_setup"]
|
||||||
|
|
||||||
|
|
||||||
|
from argenta.metrics.main import get_time_of_pre_cycle_setup
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
import io
|
||||||
|
from contextlib import redirect_stdout
|
||||||
|
from time import time
|
||||||
|
|
||||||
|
from argenta.app import App
|
||||||
|
|
||||||
|
|
||||||
|
def get_time_of_pre_cycle_setup(app: App) -> float:
|
||||||
|
"""
|
||||||
|
Public. Return time of pre cycle setup
|
||||||
|
:param app: app instance for testing time of pre cycle setup
|
||||||
|
:return: time of pre cycle setup as float
|
||||||
|
"""
|
||||||
|
start = time()
|
||||||
|
with redirect_stdout(io.StringIO()):
|
||||||
|
app.pre_cycle_setup()
|
||||||
|
end = time()
|
||||||
|
return end - start
|
||||||
@@ -5,13 +5,13 @@ from argenta.orchestrator.argparser import ArgParser
|
|||||||
|
|
||||||
|
|
||||||
class Orchestrator:
|
class Orchestrator:
|
||||||
def __init__(self, arg_parser: ArgParser = False):
|
def __init__(self, arg_parser: ArgParser | None = None):
|
||||||
"""
|
"""
|
||||||
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 | False = arg_parser
|
self.arg_parser: ArgParser | None = arg_parser
|
||||||
if arg_parser:
|
if arg_parser:
|
||||||
self.arg_parser.register_args()
|
self.arg_parser.register_args()
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
from argenta.response.status import Status
|
from argenta.response.status import Status
|
||||||
from argenta.command.flags import (
|
from argenta.command.flag.flags import (
|
||||||
ValidInputFlags,
|
ValidInputFlags,
|
||||||
UndefinedInputFlags,
|
UndefinedInputFlags,
|
||||||
InvalidValueInputFlags,
|
InvalidValueInputFlags,
|
||||||
@@ -11,7 +11,7 @@ class Response:
|
|||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
status: Status = None,
|
status: Status | None = None,
|
||||||
valid_flags: ValidInputFlags = ValidInputFlags(),
|
valid_flags: ValidInputFlags = ValidInputFlags(),
|
||||||
undefined_flags: UndefinedInputFlags = UndefinedInputFlags(),
|
undefined_flags: UndefinedInputFlags = UndefinedInputFlags(),
|
||||||
invalid_value_flags: InvalidValueInputFlags = InvalidValueInputFlags(),
|
invalid_value_flags: InvalidValueInputFlags = InvalidValueInputFlags(),
|
||||||
|
|||||||
@@ -38,7 +38,7 @@ class CommandHandler:
|
|||||||
|
|
||||||
|
|
||||||
class CommandHandlers:
|
class CommandHandlers:
|
||||||
def __init__(self, command_handlers: list[CommandHandler] = None):
|
def __init__(self, command_handlers: list[CommandHandler] | None = None):
|
||||||
"""
|
"""
|
||||||
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
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ from argenta.command import Command
|
|||||||
from argenta.command.models import InputCommand
|
from argenta.command.models import InputCommand
|
||||||
from argenta.response import Response, Status
|
from argenta.response import Response, Status
|
||||||
from argenta.router.command_handler.entity import CommandHandlers, CommandHandler
|
from argenta.router.command_handler.entity import CommandHandlers, CommandHandler
|
||||||
from argenta.command.flags.models import (
|
from argenta.command.flag.flags import (
|
||||||
Flags,
|
Flags,
|
||||||
InputFlags,
|
InputFlags,
|
||||||
UndefinedInputFlags,
|
UndefinedInputFlags,
|
||||||
@@ -22,13 +22,21 @@ from argenta.router.exceptions import (
|
|||||||
|
|
||||||
|
|
||||||
class Router:
|
class Router:
|
||||||
def __init__(self, title: str | None = "Awesome title"):
|
def __init__(
|
||||||
|
self, title: str | None = "Awesome title", disable_redirect_stdout: bool = False
|
||||||
|
):
|
||||||
"""
|
"""
|
||||||
Public. Directly configures and manages handlers
|
Public. Directly configures and manages handlers
|
||||||
:param title: the title of the router, displayed when displaying the available commands
|
:param title: the title of the router, displayed when displaying the available commands
|
||||||
|
:param disable_redirect_stdout: Disables stdout forwarding, if the argument value is True,
|
||||||
|
the StaticDividingLine will be forced to be used as a line separator for this router,
|
||||||
|
disabled forwarding is needed when there is text output in conjunction with a text input request (for example, input()),
|
||||||
|
if the argument value is True, the output of the input() prompt is intercepted and not displayed,
|
||||||
|
which is ambiguous behavior and can lead to unexpected work
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
self.title = title
|
self.title = title
|
||||||
|
self.disable_redirect_stdout = disable_redirect_stdout
|
||||||
|
|
||||||
self._command_handlers: CommandHandlers = CommandHandlers()
|
self._command_handlers: CommandHandlers = CommandHandlers()
|
||||||
self._ignore_command_register: bool = False
|
self._ignore_command_register: bool = False
|
||||||
@@ -39,13 +47,15 @@ class Router:
|
|||||||
:param command: Registered command
|
:param command: Registered command
|
||||||
:return: decorated handler as Callable
|
:return: decorated handler as Callable
|
||||||
"""
|
"""
|
||||||
self._validate_command(command)
|
|
||||||
if isinstance(command, str):
|
if isinstance(command, str):
|
||||||
command = Command(command)
|
redefined_command = Command(command)
|
||||||
|
else:
|
||||||
|
redefined_command = command
|
||||||
|
self._validate_command(redefined_command)
|
||||||
|
|
||||||
def command_decorator(func):
|
def command_decorator(func):
|
||||||
Router._validate_func_args(func)
|
Router._validate_func_args(func)
|
||||||
self._command_handlers.add_handler(CommandHandler(func, command))
|
self._command_handlers.add_handler(CommandHandler(func, redefined_command))
|
||||||
|
|
||||||
def wrapper(*args, **kwargs):
|
def wrapper(*args, **kwargs):
|
||||||
return func(*args, **kwargs)
|
return func(*args, **kwargs)
|
||||||
@@ -83,9 +93,7 @@ class Router:
|
|||||||
response: Response = Response()
|
response: Response = Response()
|
||||||
if handle_command.get_registered_flags().get_flags():
|
if handle_command.get_registered_flags().get_flags():
|
||||||
if input_command_flags.get_flags():
|
if input_command_flags.get_flags():
|
||||||
response: Response = self._structuring_input_flags(
|
response: Response = self._structuring_input_flags( handle_command, input_command_flags )
|
||||||
handle_command, input_command_flags
|
|
||||||
)
|
|
||||||
command_handler.handling(response)
|
command_handler.handling(response)
|
||||||
else:
|
else:
|
||||||
response.status = Status.ALL_FLAGS_VALID
|
response.status = Status.ALL_FLAGS_VALID
|
||||||
@@ -117,13 +125,12 @@ class Router:
|
|||||||
flag_status: Literal["Undefined", "Valid", "Invalid"] = (
|
flag_status: Literal["Undefined", "Valid", "Invalid"] = (
|
||||||
handled_command.validate_input_flag(flag)
|
handled_command.validate_input_flag(flag)
|
||||||
)
|
)
|
||||||
match flag_status:
|
if flag_status == "Valid":
|
||||||
case "Valid":
|
valid_input_flags.add_flag(flag)
|
||||||
valid_input_flags.add_flag(flag)
|
elif flag_status == "Undefined":
|
||||||
case "Undefined":
|
undefined_input_flags.add_flag(flag)
|
||||||
undefined_input_flags.add_flag(flag)
|
elif flag_status == "Invalid":
|
||||||
case "Invalid":
|
invalid_value_input_flags.add_flag(flag)
|
||||||
invalid_value_input_flags.add_flag(flag)
|
|
||||||
|
|
||||||
if (
|
if (
|
||||||
not invalid_value_input_flags.get_flags()
|
not invalid_value_input_flags.get_flags()
|
||||||
@@ -151,25 +158,20 @@ class Router:
|
|||||||
)
|
)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _validate_command(command: Command | str) -> None:
|
def _validate_command(command: Command) -> None:
|
||||||
"""
|
"""
|
||||||
Private. Validates the command registered in handler
|
Private. Validates the command registered in handler
|
||||||
:param command: validated command
|
:param command: validated command
|
||||||
:return: None if command is valid else raise exception
|
:return: None if command is valid else raise exception
|
||||||
"""
|
"""
|
||||||
match type(command).__name__:
|
command_name: str = command.get_trigger()
|
||||||
case "Command":
|
if command_name.find(" ") != -1:
|
||||||
command_name: str = command.get_trigger()
|
raise TriggerContainSpacesException()
|
||||||
if command_name.find(" ") != -1:
|
flags: Flags = command.get_registered_flags()
|
||||||
raise TriggerContainSpacesException()
|
if flags:
|
||||||
flags: Flags = command.get_registered_flags()
|
flags_name: list = [x.get_string_entity().lower() for x in flags]
|
||||||
if flags:
|
if len(set(flags_name)) < len(flags_name):
|
||||||
flags_name: list = [x.get_string_entity().lower() for x in flags]
|
raise RepeatedFlagNameException()
|
||||||
if len(set(flags_name)) < len(flags_name):
|
|
||||||
raise RepeatedFlagNameException()
|
|
||||||
case "str":
|
|
||||||
if command.find(" ") != -1:
|
|
||||||
raise TriggerContainSpacesException()
|
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _validate_func_args(func: Callable) -> None:
|
def _validate_func_args(func: Callable) -> None:
|
||||||
@@ -191,13 +193,15 @@ class Router:
|
|||||||
if arg_annotation is Response:
|
if arg_annotation is Response:
|
||||||
pass
|
pass
|
||||||
else:
|
else:
|
||||||
file_path: str = getsourcefile(func)
|
file_path: str | None = getsourcefile(func)
|
||||||
source_line: int = getsourcelines(func)[1]
|
source_line: int = getsourcelines(func)[1]
|
||||||
fprint = Console().print
|
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"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]",
|
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:
|
def set_command_register_ignore(self, _: bool) -> None:
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import re
|
|||||||
from argenta.app import App
|
from argenta.app import App
|
||||||
from argenta.command import Command
|
from argenta.command import Command
|
||||||
from argenta.router import Router
|
from argenta.router import Router
|
||||||
from argenta.command.flags.models import Flags
|
from argenta.command.flag.flags.models import Flags
|
||||||
from argenta.command.flag.defaults import PredefinedFlags
|
from argenta.command.flag.defaults import PredefinedFlags
|
||||||
from argenta.orchestrator import Orchestrator
|
from argenta.orchestrator import Orchestrator
|
||||||
from argenta.response import Response
|
from argenta.response import Response
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ from argenta.response import Response
|
|||||||
from argenta.router import Router
|
from argenta.router import Router
|
||||||
from argenta.orchestrator import Orchestrator
|
from argenta.orchestrator import Orchestrator
|
||||||
from argenta.command.flag import Flag
|
from argenta.command.flag import Flag
|
||||||
from argenta.command.flags import Flags
|
from argenta.command.flag.flags import Flags
|
||||||
from argenta.command.flag.defaults import PredefinedFlags
|
from argenta.command.flag.defaults import PredefinedFlags
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -3,6 +3,8 @@ from argenta.app import App
|
|||||||
|
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
|
from argenta.router import Router
|
||||||
|
|
||||||
|
|
||||||
class MyTestCase(unittest.TestCase):
|
class MyTestCase(unittest.TestCase):
|
||||||
def test_is_exit_command1(self):
|
def test_is_exit_command1(self):
|
||||||
@@ -33,25 +35,25 @@ class MyTestCase(unittest.TestCase):
|
|||||||
def test_is_unknown_command1(self):
|
def test_is_unknown_command1(self):
|
||||||
app = App()
|
app = App()
|
||||||
app.set_unknown_command_handler(lambda command: None)
|
app.set_unknown_command_handler(lambda command: None)
|
||||||
app._all_registered_triggers_in_lower_case = ['fr', 'tr', 'de']
|
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
|
||||||
self.assertEqual(app._is_unknown_command(InputCommand('fr')), False)
|
self.assertEqual(app._is_unknown_command(InputCommand('fr')), False)
|
||||||
|
|
||||||
def test_is_unknown_command2(self):
|
def test_is_unknown_command2(self):
|
||||||
app = App()
|
app = App()
|
||||||
app.set_unknown_command_handler(lambda command: None)
|
app.set_unknown_command_handler(lambda command: None)
|
||||||
app._all_registered_triggers_in_lower_case = ['fr', 'tr', 'de']
|
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
|
||||||
self.assertEqual(app._is_unknown_command(InputCommand('cr')), True)
|
self.assertEqual(app._is_unknown_command(InputCommand('cr')), True)
|
||||||
|
|
||||||
def test_is_unknown_command3(self):
|
def test_is_unknown_command3(self):
|
||||||
app = App(ignore_command_register=False)
|
app = App(ignore_command_register=False)
|
||||||
app.set_unknown_command_handler(lambda command: None)
|
app.set_unknown_command_handler(lambda command: None)
|
||||||
app._all_registered_triggers_in_default_case = ['Pr', 'tW', 'deQW']
|
app._current_matching_triggers_with_routers = {'Pr': Router(), 'tW': Router(), 'deQW': Router()}
|
||||||
self.assertEqual(app._is_unknown_command(InputCommand('pr')), True)
|
self.assertEqual(app._is_unknown_command(InputCommand('pr')), True)
|
||||||
|
|
||||||
def test_is_unknown_command4(self):
|
def test_is_unknown_command4(self):
|
||||||
app = App(ignore_command_register=False)
|
app = App(ignore_command_register=False)
|
||||||
app.set_unknown_command_handler(lambda command: None)
|
app.set_unknown_command_handler(lambda command: None)
|
||||||
app._all_registered_triggers_in_default_case = ['Pr', 'tW', 'deQW']
|
app._current_matching_triggers_with_routers = {'Pr': Router(), 'tW': Router(), 'deQW': Router()}
|
||||||
self.assertEqual(app._is_unknown_command(InputCommand('tW')), False)
|
self.assertEqual(app._is_unknown_command(InputCommand('tW')), False)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
from argenta.command.flag import Flag, InputFlag
|
from argenta.command.flag import Flag, InputFlag
|
||||||
from argenta.command.flags import Flags
|
from argenta.command.flag.flags import Flags
|
||||||
from argenta.command.models import InputCommand, Command
|
from argenta.command.models import InputCommand, Command
|
||||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||||
RepeatedInputFlagsException,
|
RepeatedInputFlagsException,
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
from argenta.command.flag import Flag, InputFlag
|
from argenta.command.flag import Flag, InputFlag, PossibleValues
|
||||||
from argenta.command.flags import InputFlags, Flags
|
from argenta.command.flag.flags import InputFlags, Flags
|
||||||
|
|
||||||
import unittest
|
import unittest
|
||||||
import re
|
import re
|
||||||
@@ -54,19 +54,19 @@ class TestFlag(unittest.TestCase):
|
|||||||
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
|
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
|
||||||
|
|
||||||
def test_validate_correct_empty_flag_value_without_possible_flag_values(self):
|
def test_validate_correct_empty_flag_value_without_possible_flag_values(self):
|
||||||
flag = Flag(name='test', possible_values=False)
|
flag = Flag(name='test', possible_values=PossibleValues.DISABLE)
|
||||||
self.assertEqual(flag.validate_input_flag_value(None), True)
|
self.assertEqual(flag.validate_input_flag_value(None), True)
|
||||||
|
|
||||||
def test_validate_correct_empty_flag_value_with_possible_flag_values(self):
|
def test_validate_correct_empty_flag_value_with_possible_flag_values(self):
|
||||||
flag = Flag(name='test', possible_values=True)
|
flag = Flag(name='test', possible_values=PossibleValues.DISABLE)
|
||||||
self.assertEqual(flag.validate_input_flag_value(None), True)
|
self.assertEqual(flag.validate_input_flag_value(None), True)
|
||||||
|
|
||||||
def test_validate_incorrect_random_flag_value_without_possible_flag_values(self):
|
def test_validate_incorrect_random_flag_value_without_possible_flag_values(self):
|
||||||
flag = Flag(name='test', possible_values=False)
|
flag = Flag(name='test', possible_values=PossibleValues.DISABLE)
|
||||||
self.assertEqual(flag.validate_input_flag_value('random value'), False)
|
self.assertEqual(flag.validate_input_flag_value('random value'), False)
|
||||||
|
|
||||||
def test_validate_correct_random_flag_value_with_possible_flag_values(self):
|
def test_validate_correct_random_flag_value_with_possible_flag_values(self):
|
||||||
flag = Flag(name='test', possible_values=True)
|
flag = Flag(name='test', possible_values=PossibleValues.ALL)
|
||||||
self.assertEqual(flag.validate_input_flag_value('random value'), True)
|
self.assertEqual(flag.validate_input_flag_value('random value'), True)
|
||||||
|
|
||||||
def test_get_input_flag1(self):
|
def test_get_input_flag1(self):
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
from argenta.command.flag import InputFlag, Flag
|
from argenta.command.flag import InputFlag, Flag
|
||||||
from argenta.command.flags import Flags, InputFlags, UndefinedInputFlags, InvalidValueInputFlags, ValidInputFlags
|
from argenta.command.flag.flags import Flags, InputFlags, UndefinedInputFlags, InvalidValueInputFlags, ValidInputFlags
|
||||||
from argenta.router import Router
|
from argenta.router import Router
|
||||||
from argenta.command import Command
|
from argenta.command import Command
|
||||||
from argenta.router.exceptions import (TriggerContainSpacesException,
|
from argenta.router.exceptions import (TriggerContainSpacesException,
|
||||||
|
|||||||
Reference in New Issue
Block a user