68 Commits

Author SHA1 Message Date
kolo c8e0729be8 fix bugs 2025-05-27 14:19:54 +03:00
kolo c2bbc5f15d first steps sor adding metrics tests 2025-05-24 00:39:39 +03:00
kolo 0acbf54e44 first steps sor adding metrics tests 2025-05-23 22:12:12 +03:00
kolo c3d9541330 working 2025-05-23 14:33:13 +03:00
kolo f6561de9b3 wotk 2025-05-22 20:26:48 +03:00
kolo bebd84969b add Enum PossibleValues for bool values as values of possible_values argument in Flag 2025-05-22 12:10:32 +03:00
kolo 365347ea7f some fix 2025-05-21 17:01:35 +03:00
kolo 33cb528b1d some fix 2025-05-21 13:32:45 +03:00
kolo fd287c5da0 fix type hints with ty help, add ability to configure stdout capture when handling input by the router 2025-05-20 22:44:47 +03:00
kolo 45f410e3e8 make pre_cycle_setup faster on 4 sec, start implemtnation disable redirect stdout in router 2025-05-19 10:31:05 +03:00
kolo 8b06e9cd39 add metrics concept 2025-05-12 16:22:29 +03:00
kolo c38fe10006 translate readme on de 2025-05-10 22:07:52 +03:00
kolo 03cbc64f48 translate readme 2025-05-10 21:56:34 +03:00
kolo cbf7d3c578 new warning when triggers or aliases overlap 2025-05-10 21:31:59 +03:00
kolo ea2d068022 change dataclass to enum 2025-05-10 20:13:42 +03:00
kolo 5991851207 Update README.md 2025-05-10 00:29:03 +03:00
kolo f628c3b5b5 v1.0.2 2025-05-10 00:11:57 +03:00
kolo 05379712f4 some fix 2025-05-09 23:33:54 +03:00
kolo ed1cbf0fcf stable version 2025-05-09 23:27:34 +03:00
kolo 471f05369b Merge branch 'dev' of https://github.com/koloideal/Argenta into dev 2025-05-09 23:25:47 +03:00
kolo 13f7e33db1 ruff format 2025-05-09 23:25:21 +03:00
kolo 9a78aa9263 ruff check 2025-05-09 23:24:12 +03:00
kolo 58ccd6b26d Update README.md 2025-05-09 23:20:45 +03:00
kolo 73144f7ba4 Update README.md 2025-05-09 23:19:07 +03:00
kolo 650f4c9036 some fix 2025-05-09 18:23:08 +03:00
kolo 393f5c7d81 fix 2025-05-08 23:43:08 +03:00
kolo 9eb2bb6c46 new imgs 2025-05-08 23:37:42 +03:00
kolo 79b275eac7 some fix 2025-05-08 01:08:11 +03:00
kolo 07ac2af71e some fix 2025-05-07 19:14:19 +03:00
kolo c4b3aa7db8 working 2025-05-07 02:15:42 +03:00
kolo 61ef6a6466 all tests passed 2025-05-06 21:53:53 +03:00
kolo 477f3a7dec starting refactor tests 2025-05-04 16:40:10 +03:00
kolo adf3431388 more beautiful typehints warning 2025-05-04 03:08:54 +03:00
kolo 83955aa046 first beta - adding hints for similar commands, now - feature freezing 2025-05-04 02:13:05 +03:00
kolo 5a17e916eb work on stable major version 2025-04-30 15:48:38 +03:00
kolo 1159dda16e work on Response model 2025-04-30 00:08:49 +03:00
kolo 315508a36e work on Response 2025-04-29 20:40:47 +03:00
kolo 9d6598c4e0 work on Response model 2025-04-29 00:07:32 +03:00
kolo eb43806da6 new model - Response 2025-04-28 02:21:34 +03:00
kolo e076dbf84f new img in docs 2025-04-27 23:43:14 +03:00
kolo 2f090b6b47 new img in docs 2025-04-27 23:42:56 +03:00
kolo c9dbf2bbae fix print framed text with static dividing line 2025-04-27 23:27:08 +03:00
kolo e768c1bd2c fix 2025-04-27 21:29:14 +03:00
kolo 408450ec12 fix 2025-04-27 21:20:44 +03:00
kolo 106ca058be new tests 2025-04-27 14:11:01 +03:00
kolo b5ddfb3b35 new tests 2025-04-27 13:28:11 +03:00
kolo 61e4502e41 work, fix etc. 2025-04-26 22:23:35 +03:00
kolo 9b2fc87e33 release v1.0.0a1 2025-04-25 02:29:44 +03:00
kolo 89f09c42f8 pre-release v1.0.0 2025-04-24 21:26:41 +03:00
kolo 5bcae8fe68 Update README.md 2025-04-24 17:40:53 +03:00
kolo ca58008431 fix 2025-04-23 22:07:57 +03:00
kolo 30974f48eb fix 2025-04-23 22:06:10 +03:00
kolo df4ba080b0 new docs 2025-04-23 22:02:02 +03:00
kolo f93930d712 work 2025-04-23 21:45:04 +03:00
kolo 036c17ec9a last steps work on new docs, full complete write docstring for all objects 2025-04-23 20:54:03 +03:00
kolo 7281fdeabf new app preview img in docs 2025-04-23 20:50:31 +03:00
kolo 051ec6df28 steps 2025-04-20 23:58:15 +03:00
kolo 00a1e11fc1 new docs 2025-04-19 12:13:29 +03:00
kolo 584df9ba69 new docs 2025-04-18 12:50:30 +03:00
kolo a649022f1d work on docstrings 2025-04-15 23:49:51 +03:00
kolo 26a9d8a6da work on 2025-04-15 22:26:20 +03:00
kolo 9522b0161a work on 2025-04-15 01:09:03 +03:00
kolo e189f8d9aa big step 2025-04-14 16:38:53 +03:00
kolo 3ef8707cfa big step 2025-04-14 14:54:17 +03:00
kolo a5fdcab862 work on 2025-04-14 01:03:24 +03:00
kolo ba035881ee work on support args 2025-04-13 19:24:03 +03:00
kolo 34ebe55531 first steps 2025-04-13 14:39:53 +03:00
kolo 01c9d2dc6d first step 2025-04-13 14:12:08 +03:00
86 changed files with 2455 additions and 1646 deletions
+1 -1
View File
@@ -27,4 +27,4 @@ jobs:
pip install ruff
- name: Run linter
run: ruff check ./argenta
run: ruff check ./src
+4 -2
View File
@@ -1,7 +1,9 @@
.venv
*venv
.idea
dist
poetry.lock
uv.lock
*__pycache__/
*.hist*
build
source
+65
View File
@@ -0,0 +1,65 @@
# Argenta
### Bibliothek zum Erstellen modularer CLI-Anwendungen
![preview](https://github.com/koloideal/Argenta/blob/kolo/imgs/mock_app_preview4.png?raw=True)
---
# 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)
+15 -531
View File
@@ -1,16 +1,14 @@
# Argenta
---
### Library for creating modular CLI applications
## Описание
**Argenta** — Python library for creating TUI
#### 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)
![preview](https://github.com/koloideal/Argenta/blob/kolo/imgs/mock_app_preview_last.png?raw=True)
Пример внешнего вида TUI, написанного с помощью Argenta
![preview](https://github.com/koloideal/Argenta/blob/kolo/imgs/mock_app_preview4.png?raw=True)
---
# Установка
# Installing
```bash
pip install argenta
```
@@ -21,563 +19,49 @@ poetry add argenta
---
# Быстрый старт
# Quick start
Пример простейшей оболочки с командой без зарегистрированных флагов
An example of a simple application
```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():
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)
app.start_polling()
orchestrator.start_polling(app)
if __name__ == '__main__':
main()
```
Пример оболочки с командой, у которой зарегистрированы флаги
```python
# routers.py
import re
from argenta.router import Router
from argenta.command import Command
from argenta.command.flag import Flags, Flag, InputFlags
router = Router()
registered_flags = Flags(
Flag(name='host',
prefix='--',
possible_values=re.compile(r'^192.168.\d{1,3}.\d{1,3}$')),
Flag('port', '--', re.compile(r'^[0-9]{1,4}$')))
@router.command(Command("hello"))
def handler():
print("Hello, world!")
@router.command(Command(trigger="ssh",
description='connect via ssh',
flags=registered_flags))
def handler_with_flags(flags: InputFlags):
for flag in flags:
print(f'Flag name: {flag.get_name()}\n'
f'Flag value: {flag.get_value()}')
```
---
# *classes* :
# Features in development
---
- Full support for autocompleter on Linux
## *class* : : `App`
Класс, определяющий поведение и состояние оболочки
## Full [docs](https://argenta-docs.vercel.app) | MIT 2025 kolo | made by [kolo](https://t.me/kolo_id)
### Конструктор
```python
App(prompt: str = '[italic dim bold]What do you want to do?\n',
initial_message: str = '\nArgenta\n',
farewell_message: str = '\nSee you\n',
exit_command: Command = Command('Q', 'Exit command'),
system_points_title: str | None = 'System points:',
ignore_command_register: bool = True,
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
repeat_command_groups: bool = True,
override_system_messages: bool = False,
autocompleter: AutoCompleter = AutoCompleter(),
print_func: Callable[[str], None] = Console().print)
```
**Аргументы:**
- **name : mean**
- `prompt` (`str`): Сообщение перед вводом команды.
- `initial_message` (`str`): Приветственное сообщение при запуске.
- `farewell_message` (`str`): Сообщение при выходе.
- `exit_command` (`Command`): Сущность команды, которая будет отвечать за завершение работы.
- `system_points_title` (`str`): Заголовок перед списком системных команд.
- `ignore_command_register` (`bool`): Игнорировать регистр всех команд.
- `dividing_line` (`StaticDividingLine | DynamicDividingLine`): Разделительная строка.
- `repeat_command_groups` (`bool`): Повторять описание команд перед вводом.
- `override_system_messages` (`bool`): Переопределить ли дефолтное оформление сообщений ([подробнее см.](#override_defaults))
- `autocompleter` (`AutoCompleter`): Сущность автодополнителя ввода.
- `print_func` (`Callable[[str], None]`): Функция вывода текста в терминал.
---
### ***methods***
---
#### **.start_polling() -> `None`**
*method mean* **::** Запускает цикл обработки ввода
---
#### **.include_router(router: Router) -> `None`**
*param* `router: Router` **::** Регистрируемый роутер
*required* **::** True
*method mean* **::** Регистрирует роутер в оболочке
---
#### **.include_routers(\*routers: Router) -> `None`**
*param* `routers: Router` **::** Неограниченное количество регистрируемых роутеров
*required* **::** True
*method mean* **::** Регистрирует роутер в оболочке
---
#### **.set_description_message_pattern(pattern: str) -> `None`**
*param* `pattern: str` **::** Паттерн описания команды при её выводе в консоль
*required* **::** True
*example* **::** `"[{command}] *=*=* {description}"`
*method mean* **::** Устанавливает паттерн описания команд, который будет использован
при выводе в консоль
---
#### **.add_message_on_startup(message: str) -> `None`**
*param* `message: str` **::** Сообщение, которое будет выведено при запуске приложения
*required* **::** True
*example* **::** `Message on startup`
*method mean* **::** Устанавливает паттерн описания команд, который будет использован
при выводе в консоль
---
<a name="custom_handler"></a>
#### **.repeated_input_flags_handler: `Callable[[str], None])`**
*example* **::** `lambda raw_command: print_func(f'Repeated input flags: "{raw_command}"')`
*attr mean* **::** Устанавливает функцию, которой будет передано управление при
вводе юзером повторяющихся флагов
---
#### **.invalid_input_flags_handler: `Callable[[str], None])`**
*example* **::** `lambda raw_command: print_func(f'Incorrect flag syntax: "{raw_command}"')`
*attr mean* **::** Устанавливает функцию, которой будет передано управление при
вводе юзером команды с некорректным синтаксисом флагов
---
#### **.unknown_command_handler: `Callable[[str], None]`**
*example* **::** `lambda command: print_func(f"Unknown command: {command.get_string_entity()}")`
*attr mean* **::** Устанавливает функцию, которой будет передано управление при
вводе юзером неизвестной команды
---
#### **.empty_command_handler: `Callable[[str], None])`**
*example* **::** `lambda: print_func(f'Empty input command')`
*attr mean* **::** Устанавливает функцию, которой будет передано управление при
вводе юзером пустой команды
---
### Примечания
- В устанавливаемом паттерне сообщения описания команды необходимы быть два ключевых слова:
`command` и `description`, каждое из которых должно быть заключено в фигурные скобки, после обработки
паттерна на места этих ключевых слов будут подставлены соответствующие атрибуты команды, при отсутствии
этих двух ключевых слов будет вызвано исключение `InvalidDescriptionMessagePatternException`
- Команды оболочки не должны повторяться, при значении атрибута `ignore_command_register` равным `True`
допускается создание обработчиков для разных регистров одинаковых символов в команде, для примера `u` и `U`,
при значении атрибута `ignore_command_register` класса `App` равным `False` тот же пример вызывает исключение
`RepeatedCommandInDifferentRoutersException`. Исключение вызывается только при наличии пересекающихся команд
у __<u>разных</u>__ роутеров
- Наиболее частые сообщение при запуске предопределены и доступны для быстрого
использования: `argenta.app.defaults.PredeterminedMessages`
<a name="override_defaults"></a>
- Если `override_system_messages`=`False`, то при переопределении таких атрибутов как `initial_message` и
`farawell_message` будет использовано дефолтное оформление текста, в виде красного ascii арта, при значении
`override_system_messages`=`True` системные сообщения будут отображены в точности какими были переданы
### Исключения
- `InvalidRouterInstanceException` — Переданный объект в метод `App().include_router()` не является экземпляром класса `Router`.
- `InvalidDescriptionMessagePatternException` — Неправильный формат паттерна описания команд.
- `IncorrectNumberOfHandlerArgsException` — У обработчика нестандартного поведения зарегистрировано неверное количество аргументов(в большинстве случаев у него должен быть один аргумент).
- `NoRegisteredHandlersException` — У роутера нет ни одного обработчика команд.
---
## *class* :: `AutoCompleter`
Класс, экземпляр которого представляет собой автодополнитель ввода
### Конструктор
```python
AutoCompleter(history_filename: str = False,
autocomplete_button: str = 'tab')
```
**Аргументы:**
- **name : mean**
- `history_filename` (`str` | `False`): Путь к файлу, который будет являться или является
историй пользовательского ввода, в последующем эти команды будут автодополняться, файл
может не существовать при инициализации, тогда он будет создан, при значении аргумента `False`
история пользовательского ввода будет существовать только в пределах сессии и не сохраняться в файл
- `autocomplete_button` (`str`): Строковое обозначение кнопки на клавиатуре, которая будет
использоваться для автодополнения при вводе, по умолчанию `tab`
---
## *class* :: `StaticDivideLine`
Класс, экземпляр которого представляет собой строковый разделитель фиксированной длины
### Конструктор
```python
StaticDivideLine(unit_part: str = '-',
length: int = 25)
```
**Аргументы:**
- **name : mean**
- `unit_part` (`str`): Единичная часть строкового разделителя
- `length` (`int`): Длина строкового разделителя
---
## *class* :: `DinamicDivideLine`
Строковый разделитель динамической длины, которая определяется длиной обрамляемого вывода команды
### Конструктор
```python
DinamicDivideLine(unit_part: str = '-')
```
**Аргументы:**
- **name : mean**
- `unit_part` (`str`): Единичная часть строкового разделителя
---
## *class* :: `Router`
Класс, который определяет и конфигурирует обработчики команд
### Конструктор
```python
Router(title: str | None = None,
name: str = 'Default')
```
**Аргументы:**
- **name : mean**
- `title` (`str`): Заголовок группы команд.
- `name` (`str`): Персональное название роутера
---
### ***methods***
---
#### **command(command: Command)**
*param* `command: Command` **::** Экземпляр класса `Command`, который определяет строковый триггер команды,
допустимые флаги команды и другое
*required* **::** True
*example* **::** `Command(command='ssh', description='connect via ssh')`
*method mean* **::** Декоратор, который регистрирует функцию как обработчик команды
---
#### **.get_name() -> `str`**
*method mean* **::** Возвращает установленное название роутера
---
#### **.get_title() -> `str`**
*method mean* **::** Возвращает установленный заголовок группы команд данного роутера
---
### Исключения
- `RepeatedFlagNameException` - Повторяющиеся зарегистрированные флаги в команде
- `TooManyTransferredArgsException` - Слишком много зарегистрированных аргументов у обработчика команды
- `RequiredArgumentNotPassedException` - Не зарегистрирован обязательный аргумент у обработчика команды(аргумент, через который будут переданы флаги введённой команды)
- `IncorrectNumberOfHandlerArgsException` - У обработчика нестандартного поведения зарегистрировано неверное количество аргументов(в большинстве случаев у него должен быть один аргумент)
- `TriggerCannotContainSpacesException` - У регистрируемой команды в триггере содержатся пробелы
---
## *class* :: `Command`
Класс, экземпляр которого определяет строковый триггер хэндлера и конфигурирует его атрибуты
### Конструктор
```python
Command(trigger: str,
description: str = None,
flags: Flag | Flags = None)
```
**Аргументы:**
- **name : mean**
- `trigger` (`str`): Строковый триггер
- `description` (`str`): Описание команды, которое будет выведено в консоль при запуске оболочки
- `flags` (`Flag | Flags`): Флаги, которые будут обработаны при их наличии во вводе юзера
---
#### **.get_trigger() -> `str`**
*method mean* **::** Возвращает строковый триггер экземпляра
---
#### **.get_description() -> `str`**
*method mean* **::** Возвращает описание команды
---
#### **.get_registered_flags() -> `Flags | None`**
*method mean* **::** Возвращает зарегистрированные флаги экземпляра
---
### Исключения
- `UnprocessedInputFlagException` - Некорректный синтаксис ввода команды
- `RepeatedInputFlagsException` - Повторяющиеся флаги во введённой команде
- `EmptyInputCommandException` - Введённая команда является пустой(не содержит символов)
**Примечание**
Все вышеуказанные исключения класса `Command` вызываются в рантайме запущенным экземпляром класса
`App`, а также по дефолту обрабатываются, при желании можно задать пользовательские
обработчики для этих исключений ([подробнее см.](#custom_handler))
---
## *class* :: `Flag`
Класс, экземпляры которого в большинстве случаев передаются при создании
экземпляра класса `Command` для регистрации допустимого флага при вводе юзером команды
### Конструктор
```python
Flag(name: str,
prefix: typing.Literal['-', '--', '---'] = '-',
possible_values: list[str] | typing.Pattern[str] | False = True)
```
---
**Аргументы:**
- **name : mean**
- `name` (`str`): Имя флага
- `prefix` (`Literal['-', '--', '---']`): Префикс команды, допустимым значением является от одного до трёх минусов
- `possible_values` (`list[str] | Pattern[str] | bool`): Множество допустимых значений флага, может быть задано
списком с допустимыми значениями или регулярным выражением (рекомендуется `re.compile(r'example exspression')`), при значении
аргумента `False` у введённого флага не может быть значения, иначе будет вызвано исключение и обработано соответствующим
еррор-хэндлером
---
### ***methods***
---
#### **.get_string_entity() -> `str`**
*method mean* **::** Возвращает строковое представление флага(префикс + имя)
---
#### **.get_name() -> `str`**
*method mean* **::** Возвращает имя флага
---
#### **.get_prefix() -> `str`**
*method mean* **::** Возвращает префикс флага
---
## *class* :: `InputFlag`
Класс, экземпляры которого являются введёнными флагами команды, передаётся в хэндлер команды
через `InputFlags`
---
### Примечания
- Наиболее часто используемые флаги предопределены и доступны для быстрого использования:
`argenta.command.flag.defaults.PredeterminedFlags`
---
### Конструктор
```python
InputFlag(name: str,
prefix: typing.Literal['-', '--', '---'] = '-',
value: str = None)
```
---
**Аргументы:**
- **name : mean**
- `name` (`str`): Имя флага
- `prefix` (`Literal['-', '--', '---']`): Префикс команды, допустимым значением является от одного до трёх минусов
- `value` (`str`): Значение введённого флага, если оно есть
---
### ***methods***
---
#### **.get_value() -> `str | None`**
*method mean* **::** Возвращает значение введённого флага
---
## *class* :: `Flags`
Класс, объединяющий список флагов в один объект, используется в качестве
передаваемого аргумента `flags` экземпляру класса `Command`, при регистрации
хэндлера
### Конструктор
```python
Flags(*flags: Flag)
```
---
**Аргументы:**
- **name : mean**
- `*flags` (`Flag`): Неограниченное количество передаваемых флагов
---
### ***methods***
---
#### **.get_flags() -> `list[Flag]`**
*method mean* **::** Возвращает зарегистрированные флаги
---
#### **.add_flag(flag: Flag) -> `None`**
*method mean* **::** Добавляет флаг в группу
---
#### **.add_flags(flags: list[Flag]) -> `None`**
*method mean* **::** Добавляет флаги в группу
---
#### **.get_flag(name: str) -> `Flag | None`**
*param* `name: str` **::** Строковый триггер флага без префикса
*required* **::** True
*example* **::** `'host'`
*method mean* **::** Возвращает флаг по его триггеру или `None`, если флаг не найден
---
## *class* :: `InputFlags`
Класс, объединяющий список введённых флагов в один объект, передаётся соответствующему хэндлеру
в качестве аргумента
### Конструктор
```python
InputFlags(*flags: Flag)
```
---
**Аргументы:**
- **name : mean**
- `*flags` (`InputFlag`): Неограниченное количество передаваемых флагов
---
### ***methods***
---
#### **.get_flags() -> `list[Flag]`**
*method mean* **::** Возвращает введённые флаги
---
#### **.get_flag(name: str) -> `InputFlag | None`**
*param* `name: str` **::** Строковый триггер флага без префикса
*required* **::** True
*example* **::** `'host'`
*method mean* **::** Возвращает введённый флаг по его триггеру или `None`, если флаг не найден
---
# Тесты
Запуск тестов:
```bash
python -m unittest discover
```
or
```bash
python -m unittest discover -v
```
+65
View File
@@ -0,0 +1,65 @@
# Argenta
### Библиотека для создания модульных CLI приложeний
![preview](https://github.com/koloideal/Argenta/blob/kolo/imgs/mock_app_preview4.png?raw=True)
---
# Установка
```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)
-47
View File
@@ -1,47 +0,0 @@
import os
import readline
class AutoCompleter:
def __init__(self, history_filename: str = False, autocomplete_button: str = 'tab'):
self.history_filename = history_filename
self.autocomplete_button = autocomplete_button
self.matches = []
def complete(self, text, state):
matches = sorted(cmd for cmd in self.get_history_items() if cmd.startswith(text))
if len(matches) > 1:
common_prefix = matches[0]
for match in matches[1:]:
i = 0
while i < len(common_prefix) and i < len(match) and common_prefix[i] == match[i]:
i += 1
common_prefix = common_prefix[:i]
if state == 0:
readline.insert_text(common_prefix[len(text):])
readline.redisplay()
return None
elif len(matches) == 1:
return matches[0] if state == 0 else None
else:
return None
def initial_setup(self, all_commands: list[str]):
if self.history_filename:
if os.path.exists(self.history_filename):
readline.read_history_file(self.history_filename)
else:
for line in all_commands:
readline.add_history(line)
readline.set_completer(self.complete)
readline.set_completer_delims(readline.get_completer_delims().replace(' ', ''))
readline.parse_and_bind(f'{self.autocomplete_button}: complete')
def exit_setup(self):
if self.history_filename:
readline.write_history_file(self.history_filename)
@staticmethod
def get_history_items():
return [readline.get_history_item(i) for i in range(1, readline.get_current_history_length() + 1)]
-9
View File
@@ -1,9 +0,0 @@
from dataclasses import dataclass
@dataclass
class PredeterminedMessages:
USAGE = '[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]'
HELP = '[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]'
AUTOCOMPLETE = '[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>'
-24
View File
@@ -1,24 +0,0 @@
class BaseDividingLine:
def __init__(self, unit_part: str = '-'):
self.unit_part = unit_part
def get_unit_part(self):
if len(self.unit_part) == 0:
return ' '
else:
return self.unit_part[0]
class StaticDividingLine(BaseDividingLine):
def __init__(self, unit_part: str = '-', length: int = 25):
super().__init__(unit_part)
self.length = length
def get_full_line(self):
return f'\n[dim]{self.length * self.get_unit_part()}[/dim]\n'
class DynamicDividingLine(BaseDividingLine):
def get_full_line(self, length: int):
return f'\n[dim]{self.get_unit_part() * length}[/dim]\n'
-10
View File
@@ -1,10 +0,0 @@
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}'"
-265
View File
@@ -1,265 +0,0 @@
from typing import Callable
from rich.console import Console
from rich.markup import escape
from art import text2art
from contextlib import redirect_stdout
import io
import re
from argenta.command.models import Command, InputCommand
from argenta.router import Router
from argenta.router.defaults import system_router
from argenta.app.autocompleter import AutoCompleter
from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine
from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException,
BaseInputCommandException)
from argenta.app.exceptions import (NoRegisteredRoutersException,
NoRegisteredHandlersException)
from argenta.app.registered_routers.entity import RegisteredRouters
class AppInit:
def __init__(self,
prompt: str = '[italic dim bold]What do you want to do?\n',
initial_message: str = '\nArgenta\n',
farewell_message: str = '\nSee you\n',
exit_command: Command = Command('Q', 'Exit command'),
system_points_title: str | None = 'System points:',
ignore_command_register: bool = True,
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
repeat_command_groups: bool = True,
override_system_messages: bool = False,
autocompleter: AutoCompleter = AutoCompleter(),
print_func: Callable[[str], None] = Console().print) -> None:
self._prompt = prompt
self._print_func = print_func
self._exit_command = exit_command
self._system_points_title = system_points_title
self._dividing_line = dividing_line
self._ignore_command_register = ignore_command_register
self._repeat_command_groups_description = repeat_command_groups
self._override_system_messages = override_system_messages
self._autocompleter = autocompleter
self._farewell_message = farewell_message
self._initial_message = initial_message
self._description_message_gen: Callable[[str, str], str] = lambda command, description: f'[bold red]{escape('['+command+']')}[/bold red] [blue dim]*=*=*[/blue dim] [bold yellow italic]{escape(description)}'
self._registered_routers: RegisteredRouters = RegisteredRouters()
self._messages_on_startup = []
self._invalid_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'[red bold]Incorrect flag syntax: {escape(raw_command)}')
self._repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'[red bold]Repeated input flags: {escape(raw_command)}')
self._empty_input_command_handler: Callable[[], None] = lambda: print_func('[red bold]Empty input command')
self._unknown_command_handler: Callable[[InputCommand], None] = lambda command: print_func(f"[red bold]Unknown command: {escape(command.get_trigger())}")
self._exit_command_handler: Callable[[], None] = lambda: print_func(self._farewell_message)
class AppSetters(AppInit):
def set_description_message_pattern(self, pattern: Callable[[str, str], str]) -> None:
self._description_message_gen: Callable[[str, str], str] = pattern
def set_invalid_input_flags_handler(self, handler: Callable[[str], None]) -> None:
self._invalid_input_flags_handler = handler
def set_repeated_input_flags_handler(self, handler: Callable[[str], None]) -> None:
self._repeated_input_flags_handler = handler
def set_unknown_command_handler(self, handler: Callable[[str], None]) -> None:
self._unknown_command_handler = handler
def set_empty_command_handler(self, handler: Callable[[], None]) -> None:
self._empty_input_command_handler = handler
def set_exit_command_handler(self, handler: Callable[[], None]) -> None:
self._exit_command_handler = handler
class AppPrinters(AppInit):
def _print_command_group_description(self):
for registered_router in self._registered_routers:
if registered_router.get_title():
self._print_func(registered_router.get_title())
for command_handler in registered_router.get_command_handlers():
self._print_func(self._description_message_gen(
command_handler.get_handled_command().get_trigger(),
command_handler.get_handled_command().get_description()))
self._print_func('')
def _print_framed_text_with_dynamic_line(self, text: str):
clear_text = re.sub(r'\u001b\[[0-9;]*m', '', text)
max_length_line = max([len(line) for line in clear_text.split('\n')])
max_length_line = max_length_line if 10 <= max_length_line <= 80 else 80 if max_length_line > 80 else 10
self._print_func(self._dividing_line.get_full_line(max_length_line))
print(text.strip('\n'))
self._print_func(self._dividing_line.get_full_line(max_length_line))
def _print_framed_text(self, text: str):
if isinstance(self._dividing_line, StaticDividingLine):
self._print_func(self._dividing_line.get_full_line())
self._print_func(text)
self._print_func(self._dividing_line.get_full_line())
elif isinstance(self._dividing_line, DynamicDividingLine):
self._print_framed_text_with_dynamic_line(text)
class AppNonStandardHandlers(AppPrinters):
def _is_exit_command(self, command: InputCommand):
if command.get_trigger().lower() == self._exit_command.get_trigger().lower():
if self._ignore_command_register:
system_router.input_command_handler(command)
return True
elif command.get_trigger() == self._exit_command.get_trigger():
system_router.input_command_handler(command)
return True
return False
def _is_unknown_command(self, command: InputCommand):
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()
handled_command_aliases = command_handler.get_handled_command().get_aliases()
if handled_command_trigger.lower() == command.get_trigger().lower() and self._ignore_command_register:
return False
elif handled_command_trigger == command.get_trigger():
return False
elif handled_command_aliases:
if (command.get_trigger().lower() in [x.lower() for x in handled_command_aliases]) and self._ignore_command_register:
return False
elif command.get_trigger() in handled_command_trigger:
return False
if isinstance(self._dividing_line, StaticDividingLine):
self._print_func(self._dividing_line.get_full_line())
self._unknown_command_handler(command)
self._print_func(self._dividing_line.get_full_line())
elif isinstance(self._dividing_line, DynamicDividingLine):
with redirect_stdout(io.StringIO()) as f:
self._unknown_command_handler(command)
res: str = f.getvalue()
self._print_framed_text_with_dynamic_line(res)
return True
def _error_handler(self, error: BaseInputCommandException, raw_command: str) -> None:
match error:
case UnprocessedInputFlagException():
self._invalid_input_flags_handler(raw_command)
case RepeatedInputFlagsException():
self._repeated_input_flags_handler(raw_command)
case EmptyInputCommandException():
self._empty_input_command_handler()
class AppValidators(AppInit):
def _validate_number_of_routers(self) -> None:
if not self._registered_routers:
raise NoRegisteredRoutersException()
def _validate_included_routers(self) -> None:
for router in self._registered_routers:
if not router.get_command_handlers():
raise NoRegisteredHandlersException(router.get_name())
class AppSetups(AppValidators, AppPrinters):
def _setup_system_router(self):
system_router.set_title(self._system_points_title)
@system_router.command(self._exit_command)
def exit_command():
self._exit_command_handler()
if system_router not in self._registered_routers.get_registered_routers():
system_router.set_ignore_command_register(self._ignore_command_register)
self._registered_routers.add_registered_router(system_router)
def _setup_default_view(self):
if not self._override_system_messages:
self._initial_message = f'\n[bold red]{text2art(self._initial_message, font='tarty1')}\n\n'
self._farewell_message = (
f'[bold red]\n{text2art(f'\n{self._farewell_message}\n', font='chanky')}[/bold red]\n'
f'[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n')
def _pre_cycle_setup(self):
self._setup_default_view()
self._setup_system_router()
self._validate_number_of_routers()
self._validate_included_routers()
all_triggers: list[str] = []
for router_entity in self._registered_routers:
all_triggers.extend(router_entity.get_triggers())
all_triggers.extend(router_entity.get_aliases())
self._autocompleter.initial_setup(all_triggers)
self._print_func(self._initial_message)
for message in self._messages_on_startup:
self._print_func(message)
print('\n\n')
if not self._repeat_command_groups_description:
self._print_command_group_description()
class App(AppSetters, AppNonStandardHandlers, AppSetups):
def start_polling(self) -> None:
self._pre_cycle_setup()
while True:
if self._repeat_command_groups_description:
self._print_command_group_description()
raw_command: str = Console().input(self._prompt)
try:
input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
except BaseInputCommandException as error:
with redirect_stdout(io.StringIO()) as f:
self._error_handler(error, raw_command)
res: str = f.getvalue()
self._print_framed_text(res)
continue
if self._is_exit_command(input_command):
self._autocompleter.exit_setup()
return
if self._is_unknown_command(input_command):
continue
with redirect_stdout(io.StringIO()) as f:
for registered_router in self._registered_routers:
registered_router.input_command_handler(input_command)
res: str = f.getvalue()
self._print_framed_text(res)
if not self._repeat_command_groups_description:
self._print_func(self._prompt)
def include_router(self, router: Router) -> None:
router.set_ignore_command_register(self._ignore_command_register)
self._registered_routers.add_registered_router(router)
def include_routers(self, *routers: Router) -> None:
for router in routers:
self.include_router(router)
def add_message_on_startup(self, message: str) -> None:
self._messages_on_startup.append(message)
-21
View File
@@ -1,21 +0,0 @@
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))
-3
View File
@@ -1,3 +0,0 @@
__all__ = ["Command"]
from argenta.command.models import Command
-23
View File
@@ -1,23 +0,0 @@
from argenta.command.flag.models import InputFlag, Flag
class BaseInputCommandException(Exception):
pass
class UnprocessedInputFlagException(BaseInputCommandException):
def __str__(self):
return "Unprocessed Input Flags"
class RepeatedInputFlagsException(BaseInputCommandException):
def __init__(self, flag: Flag | InputFlag):
self.flag = flag
def __str__(self):
return ("Repeated Input Flags\n"
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'")
class EmptyInputCommandException(BaseInputCommandException):
def __str__(self):
return "Input Command is empty"
-4
View File
@@ -1,4 +0,0 @@
__all__ = ('InputFlags', 'InputFlag', 'Flag', 'Flags')
from argenta.command.flag.models import InputFlags, InputFlag, Flags, Flag
-21
View File
@@ -1,21 +0,0 @@
from dataclasses import dataclass
from argenta.command.flag.models import Flag
import re
@dataclass
class PredeterminedFlags:
HELP = Flag(name='help', possible_values=False)
SHORT_HELP = Flag(name='h', prefix='-', possible_values=False)
INFO = Flag(name='info', possible_values=False)
SHORT_INFO = Flag(name='i', prefix='-', possible_values=False)
ALL = Flag(name='all', possible_values=False)
SHORT_ALL = Flag(name='a', prefix='-', possible_values=False)
HOST = Flag(name='host', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
SHORT_HOST = Flag(name='h', prefix='-', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
PORT = Flag(name='port', possible_values=re.compile(r'^\d{1,5}$'))
SHORT_PORT = Flag(name='p', prefix='-', possible_values=re.compile(r'^\d{1,5}$'))
-140
View File
@@ -1,140 +0,0 @@
from typing import Literal, Pattern
from abc import ABC, abstractmethod
class BaseFlag:
def __init__(self, name: str,
prefix: Literal['-', '--', '---'] = '--'):
self._name = name
self._prefix = prefix
def get_string_entity(self):
string_entity: str = self._prefix + self._name
return string_entity
def get_name(self):
return self._name
def get_prefix(self):
return self._prefix
class InputFlag(BaseFlag):
def __init__(self, name: str,
prefix: Literal['-', '--', '---'] = '--',
value: str = None):
super().__init__(name, prefix)
self._flag_value = value
def get_value(self) -> str | None:
return self._flag_value
def set_value(self, value):
self._flag_value = value
class Flag(BaseFlag):
def __init__(self, name: str,
prefix: Literal['-', '--', '---'] = '--',
possible_values: list[str] | Pattern[str] | False = True):
super().__init__(name, prefix)
self.possible_values = possible_values
def validate_input_flag_value(self, input_flag_value: str | None):
if self.possible_values is False:
if input_flag_value is None:
return True
else:
return False
elif isinstance(self.possible_values, Pattern):
if isinstance(input_flag_value, str):
is_valid = bool(self.possible_values.match(input_flag_value))
if bool(is_valid):
return True
else:
return False
else:
return False
elif isinstance(self.possible_values, list):
if input_flag_value in self.possible_values:
return True
else:
return False
else:
return True
class BaseFlags(ABC):
__slots__ = ('_flags',)
@abstractmethod
def get_flags(self):
pass
@abstractmethod
def add_flag(self, flag: Flag | InputFlag):
pass
@abstractmethod
def add_flags(self, flags: list[Flag] | list[InputFlag]):
pass
@abstractmethod
def get_flag(self, name: str):
pass
def __iter__(self):
return iter(self._flags)
def __next__(self):
return next(iter(self))
def __getitem__(self, item):
return self._flags[item]
class Flags(BaseFlags, ABC):
def __init__(self, *flags: Flag):
self._flags = flags if flags else []
def get_flags(self) -> list[Flag]:
return self._flags
def add_flag(self, flag: Flag):
self._flags.append(flag)
def add_flags(self, flags: list[Flag]):
self._flags.extend(flags)
def get_flag(self, name: str) -> Flag | None:
if name in [flag.get_name() for flag in self._flags]:
return list(filter(lambda flag: flag.get_name() == name, self._flags))[0]
else:
return None
class InputFlags(BaseFlags, ABC):
def __init__(self, *flags: InputFlag):
self._flags = flags if flags else []
def get_flags(self) -> list[InputFlag]:
return self._flags
def add_flag(self, flag: InputFlag):
self._flags.append(flag)
def add_flags(self, flags: list[InputFlag]):
self._flags.extend(flags)
def get_flag(self, name: str) -> InputFlag | None:
if name in [flag.get_name() for flag in self._flags]:
return list(filter(lambda flag: flag.get_name() == name, self._flags))[0]
else:
return None
-112
View File
@@ -1,112 +0,0 @@
from argenta.command.flag.models import Flag, InputFlag, Flags, InputFlags
from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException)
from typing import Generic, TypeVar, cast, Literal
InputCommandType = TypeVar('InputCommandType')
class BaseCommand:
def __init__(self, trigger: str):
self._trigger = trigger
def get_trigger(self) -> str:
return self._trigger
class Command(BaseCommand):
def __init__(self, trigger: str,
description: str = None,
flags: Flag | Flags = None,
aliases: list[str] = None):
super().__init__(trigger)
self._registered_flags: Flags = flags if isinstance(flags, Flags) else Flags(flags) if isinstance(flags, Flag) else Flags()
self._description = f'Description for "{self._trigger}" command' if not description else description
self._aliases = aliases
def get_registered_flags(self) -> Flags:
return self._registered_flags
def get_aliases(self) -> list[str] | None:
return self._aliases
def validate_input_flag(self, flag: InputFlag):
registered_flags: Flags | None = self.get_registered_flags()
if registered_flags:
if isinstance(registered_flags, Flag):
if registered_flags.get_string_entity() == flag.get_string_entity():
is_valid = registered_flags.validate_input_flag_value(flag.get_value())
if is_valid:
return True
else:
for registered_flag in registered_flags:
if registered_flag.get_string_entity() == flag.get_string_entity():
is_valid = registered_flag.validate_input_flag_value(flag.get_value())
if is_valid:
return True
return False
def get_description(self) -> str:
return self._description
class InputCommand(BaseCommand, Generic[InputCommandType]):
def __init__(self, trigger: str,
input_flags: InputFlag | InputFlags = None):
super().__init__(trigger)
self._input_flags: InputFlags = input_flags if isinstance(input_flags, InputFlags) else InputFlags(input_flags) if isinstance(input_flags, InputFlag) else InputFlags()
def _set_input_flags(self, input_flags: InputFlags):
self._input_flags = input_flags
def get_input_flags(self) -> InputFlags:
return self._input_flags
@staticmethod
def parse(raw_command: str) -> InputCommandType:
if not raw_command:
raise EmptyInputCommandException()
list_of_tokens = raw_command.split()
command = list_of_tokens.pop(0)
input_flags: InputFlags = InputFlags()
current_flag_name, current_flag_value = None, None
for k, _ in enumerate(list_of_tokens):
if _.startswith('-'):
if current_flag_name or len(_) < 2 or len(_[:_.rfind('-')]) > 3:
raise UnprocessedInputFlagException()
current_flag_name = _
else:
if not current_flag_name:
raise UnprocessedInputFlagException()
current_flag_value = _
if current_flag_name:
if not len(list_of_tokens) == k+1:
if not list_of_tokens[k+1].startswith('-'):
continue
input_flag = InputFlag(name=current_flag_name[current_flag_name.rfind('-')+1:],
prefix=cast(Literal['-', '--', '---'],
current_flag_name[:current_flag_name.rfind('-')+1]),
value=current_flag_value)
all_flags = [flag.get_string_entity() for flag in input_flags.get_flags()]
if input_flag.get_string_entity() not in all_flags:
input_flags.add_flag(input_flag)
else:
raise RepeatedInputFlagsException(input_flag)
current_flag_name, current_flag_value = None, None
if any([current_flag_name, current_flag_value]):
raise UnprocessedInputFlagException()
else:
return InputCommand(trigger=command, input_flags=input_flags)
-4
View File
@@ -1,4 +0,0 @@
__all__ = ["Router"]
from argenta.router.entity import Router
-21
View File
@@ -1,21 +0,0 @@
from typing import Callable
from argenta.command import Command
from argenta.command.flag.models import InputFlags
class CommandHandler:
def __init__(self, handler: Callable[[], None] | Callable[[InputFlags], None], handled_command: Command):
self._handler = handler
self._handled_command = handled_command
def handling(self, input_flags: InputFlags = 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
-21
View File
@@ -1,21 +0,0 @@
from argenta.router.command_handler.entity import CommandHandler
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))
-5
View File
@@ -1,5 +0,0 @@
from argenta.router import Router
system_router = Router(title='System points:',
name='System')
-149
View File
@@ -1,149 +0,0 @@
from typing import Callable, Any
from inspect import getfullargspec
from argenta.command import Command
from argenta.command.models import InputCommand
from argenta.router.command_handlers.entity import CommandHandlers
from argenta.router.command_handler.entity import CommandHandler
from argenta.command.flag.models import Flag, Flags, InputFlags
from argenta.router.exceptions import (RepeatedFlagNameException,
TooManyTransferredArgsException,
RequiredArgumentNotPassedException,
IncorrectNumberOfHandlerArgsException,
TriggerCannotContainSpacesException)
class Router:
def __init__(self,
title: str = None,
name: str = 'Default'):
self._title = title
self._name = name
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 ''}")
def command(self, command: Command) -> Callable[[Any], Any]:
self._validate_command(command)
def command_decorator(func):
Router._validate_func_args(command, func)
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:
raise IncorrectNumberOfHandlerArgsException()
else:
self._not_valid_flag_handler = func
def input_command_handler(self, input_command: InputCommand):
input_command_name: str = input_command.get_trigger()
input_command_flags: InputFlags = input_command.get_input_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():
self._validate_input_command(input_command_flags, command_handler)
elif handle_command.get_aliases():
if input_command_name.lower() in handle_command.get_aliases():
self._validate_input_command(input_command_flags, command_handler)
def _validate_input_command(self, input_command_flags: InputFlags, command_handler: CommandHandler):
handle_command = command_handler.get_handled_command()
if handle_command.get_registered_flags().get_flags():
if input_command_flags.get_flags():
if self._validate_input_flags(handle_command, input_command_flags):
command_handler.handling(input_command_flags)
return
else:
command_handler.handling(input_command_flags)
return
else:
if input_command_flags.get_flags():
self._not_valid_flag_handler(input_command_flags[0])
return
else:
command_handler.handling()
return
def _validate_input_flags(self, handle_command: Command, input_flags: InputFlags):
for flag in input_flags:
is_valid = handle_command.validate_input_flag(flag)
if not is_valid:
self._not_valid_flag_handler(flag)
return False
return True
@staticmethod
def _validate_command(command: Command):
command_name: str = command.get_trigger()
if command_name.find(' ') != -1:
raise TriggerCannotContainSpacesException()
flags: Flags = command.get_registered_flags()
if flags:
flags_name: list = [x.get_string_entity().lower() for x in flags]
if len(set(flags_name)) < len(flags_name):
raise RepeatedFlagNameException()
@staticmethod
def _validate_func_args(command: Command, func: Callable):
registered_args = command.get_registered_flags()
transferred_args = getfullargspec(func).args
if registered_args.get_flags() and transferred_args:
if len(transferred_args) != 1:
raise TooManyTransferredArgsException()
elif registered_args.get_flags() and not transferred_args:
raise RequiredArgumentNotPassedException()
elif not registered_args.get_flags() and transferred_args:
raise TooManyTransferredArgsException()
def set_ignore_command_register(self, ignore_command_register: bool):
self._ignore_command_register = ignore_command_register
def get_triggers(self):
all_triggers: list[str] = []
for command_handler in self._command_handlers:
all_triggers.append(command_handler.get_handled_command().get_trigger())
return all_triggers
def get_aliases(self):
all_aliases: list[str] = []
for command_handler in self._command_handlers:
if command_handler.get_handled_command().get_aliases():
all_aliases.extend(command_handler.get_handled_command().get_aliases())
return all_aliases
def get_command_handlers(self) -> CommandHandlers:
return self._command_handlers
def get_name(self) -> str:
return self._name
def get_title(self) -> str | None:
return self._title
def set_title(self, title: str):
self._title = title
-23
View File
@@ -1,23 +0,0 @@
class RepeatedFlagNameException(Exception):
def __str__(self):
return "Repeated registered_flag name in register command"
class TooManyTransferredArgsException(Exception):
def __str__(self):
return "Too many transferred arguments"
class RequiredArgumentNotPassedException(Exception):
def __str__(self):
return "Required argument not passed"
class IncorrectNumberOfHandlerArgsException(Exception):
def __str__(self):
return "Handler has incorrect number of arguments"
class TriggerCannotContainSpacesException(Exception):
def __str__(self):
return "Command trigger cannot contain spaces"
Binary file not shown.

After

Width:  |  Height:  |  Size: 1.5 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.9 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 928 KiB

+46
View File
@@ -0,0 +1,46 @@
<?xml version="1.0" standalone="no"?>
<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 20010904//EN"
"http://www.w3.org/TR/2001/REC-SVG-20010904/DTD/svg10.dtd">
<svg version="1.0" xmlns="http://www.w3.org/2000/svg"
width="809.000000pt" height="809.000000pt" viewBox="0 0 809.000000 809.000000"
preserveAspectRatio="xMidYMid meet">
<g transform="translate(0.000000,809.000000) scale(0.100000,-0.100000)"
fill="#000000" stroke="none">
<path d="M3845 7600 c-683 -38 -1333 -259 -1875 -639 -702 -492 -1223 -1251
-1434 -2089 -83 -330 -111 -602 -103 -980 6 -273 19 -393 67 -632 167 -829
642 -1593 1321 -2122 512 -399 1133 -656 1794 -745 165 -22 666 -25 825 -5
779 99 1451 397 2020 898 687 603 1095 1401 1201 2344 17 147 17 571 0 715
-59 518 -204 977 -442 1402 -180 322 -355 552 -629 823 -518 515 -1164 850
-1887 979 -254 45 -594 66 -858 51z m655 -234 c545 -76 1005 -248 1445 -541
195 -130 336 -245 506 -415 457 -456 758 -987 909 -1605 134 -547 132 -1119
-5 -1665 -149 -593 -445 -1112 -885 -1550 -624 -623 -1414 -966 -2315 -1006
-412 -18 -935 73 -1339 232 -656 259 -1228 720 -1610 1299 -298 450 -461 890
-538 1451 -28 210 -31 620 -4 819 103 786 437 1477 975 2016 149 149 266 249
417 357 315 225 692 405 1059 506 211 58 342 81 675 120 87 10 601 -3 710 -18z"/>
<path d="M3691 6759 c-231 -17 -522 -67 -660 -114 -227 -77 -354 -211 -381
-400 -14 -105 -13 -628 2 -643 9 -9 158 -12 611 -12 598 0 628 -2 649 -34 14
-21 8 -66 -12 -86 -20 -20 -33 -20 -908 -20 -857 0 -892 -1 -967 -20 -294 -75
-500 -321 -599 -715 -49 -195 -61 -309 -60 -585 0 -221 3 -272 23 -385 44
-251 116 -418 232 -540 78 -82 143 -123 254 -162 78 -27 85 -27 351 -31 l272
-4 16 23 c14 20 16 58 16 259 0 334 20 446 107 613 80 153 268 310 447 374
128 45 138 46 881 52 683 7 713 8 786 28 201 56 353 177 437 348 80 162 86
240 74 905 -11 608 -10 600 -84 743 -80 153 -240 271 -453 332 -252 73 -653
101 -1034 74z m-383 -466 c126 -78 147 -245 44 -355 -144 -154 -396 -54 -395
157 0 97 62 187 154 222 54 20 143 10 197 -24z"/>
<path d="M5451 5491 c-20 -20 -21 -30 -21 -283 0 -276 -7 -351 -42 -458 -81
-251 -281 -454 -523 -534 -149 -48 -151 -48 -890 -56 -678 -6 -703 -7 -780
-28 -139 -38 -219 -84 -315 -181 -61 -61 -95 -105 -117 -151 -65 -134 -64
-129 -75 -780 -12 -665 -10 -694 47 -817 122 -265 419 -429 895 -494 127 -18
592 -18 730 -1 377 48 646 169 785 355 57 75 67 94 96 182 20 64 23 94 27 325
4 238 3 257 -15 278 l-18 23 -551 -1 c-399 0 -559 3 -577 11 -51 23 -56 102
-8 126 9 4 439 10 956 13 1044 7 981 2 1125 75 178 89 305 255 386 502 84 257
111 618 73 973 -39 364 -167 655 -351 801 -63 50 -121 80 -213 110 -64 20 -93
23 -337 27 -260 4 -267 3 -287 -17z m-2104 -1698 c78 -52 531 -413 561 -447
17 -19 33 -48 37 -65 15 -71 3 -84 -323 -356 -275 -229 -310 -255 -342 -255
-78 1 -112 75 -61 134 9 10 86 75 171 144 247 201 355 294 358 311 1 9 -15 28
-38 44 -127 89 -503 393 -512 413 -23 50 20 103 82 104 15 0 45 -12 67 -27z
m1388 -1244 c59 -15 129 -77 151 -134 36 -96 0 -204 -88 -262 -36 -24 -51 -28
-118 -28 -67 0 -82 4 -118 28 -153 101 -124 338 48 392 57 17 71 18 125 4z"/>
</g>
</svg>

After

Width:  |  Height:  |  Size: 2.9 KiB

Before

Width:  |  Height:  |  Size: 60 KiB

After

Width:  |  Height:  |  Size: 60 KiB

Before

Width:  |  Height:  |  Size: 58 KiB

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 54 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 50 KiB

@@ -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)
-9
View File
@@ -1,9 +0,0 @@
from argenta.app import App
from argenta.app.defaults import PredeterminedMessages
app = App(repeat_command_groups=True)
app.add_message_on_startup(PredeterminedMessages.USAGE + '\n\n')
app.start_polling()
+8 -4
View File
@@ -1,14 +1,18 @@
from argenta.app import App
from argenta.command import Command
from argenta.orchestrator import Orchestrator
from argenta.router import Router
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response):
print('test command')
app = App(ignore_command_register=False)
app = App(ignore_command_register=True,
override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
app.start_polling()
orchestrator.start_polling(app)
View File
@@ -1,10 +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]")
-35
View File
@@ -1,35 +0,0 @@
from rich.console import Console
from argenta.command import Command
from argenta.command.flag import Flags, InputFlags
from argenta.command.flag.defaults import PredeterminedFlags
from argenta.router import Router
from .handlers_implementation.help_command import help_command
work_router: Router = Router(title='Work points:')
settings_router: Router = Router()
console = Console()
@work_router.command(Command('get', 'Get Help', aliases=['help', 'Get_help']))
def command_help():
help_command()
@work_router.command(Command('start', 'Start Solving',
flags=Flags(PredeterminedFlags.HOST, PredeterminedFlags.PORT),
aliases=['starting']))
def command_start_solving(args: InputFlags):
print(args.get_flag('host'))
@settings_router.command(Command('update', 'Update WordMath'))
def command_update():
print('eeeeeee')
+18 -10
View File
@@ -1,24 +1,32 @@
from mock.mock_app.handlers.routers import work_router, settings_router
from mock.mock_app.routers import work_router
from argenta.app import App
from argenta.app.defaults import PredeterminedMessages
from argenta.app.defaults import PredefinedMessages
from argenta.app.dividing_line import DynamicDividingLine
from argenta.app.autocompleter import AutoCompleter
from argenta.orchestrator import Orchestrator
from argenta.orchestrator.argparser import ArgParser
from argenta.orchestrator.argparser.arguments import BooleanArgument
autocompleter = AutoCompleter('./mock/.hist')
app: App = App(dividing_line=DynamicDividingLine(),
autocompleter=autocompleter)
arg_parser = ArgParser(processed_args=[BooleanArgument("repeat")])
app: App = App(
dividing_line=DynamicDividingLine(),
autocompleter=AutoCompleter(),
repeat_command_groups=False,
)
orchestrator: Orchestrator = Orchestrator(arg_parser)
def main():
app.include_routers(work_router, settings_router)
app.include_router(work_router)
app.add_message_on_startup(PredeterminedMessages.USAGE)
app.add_message_on_startup(PredeterminedMessages.AUTOCOMPLETE)
app.add_message_on_startup(PredeterminedMessages.HELP)
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)
app.start_polling()
if __name__ == "__main__":
main()
+39
View File
@@ -0,0 +1,39 @@
from rich.console import Console
from argenta.command import Command
from argenta.command.flag.defaults import PredefinedFlags
from argenta.command.flag import Flags, Flag, PossibleValues
from argenta.response import Response
from argenta.router import Router
work_router: Router = Router(title="Work points:", disable_redirect_stdout=True)
console = Console()
flag = Flag('csdv', possible_values=PossibleValues.DISABLE)
@work_router.command(
Command(
"get",
"Get Help",
aliases=["help", "Get_help"],
flags=Flags(PredefinedFlags.PORT, PredefinedFlags.HOST),
)
)
def command_help(response: Response):
case = input("test > ")
print(case)
print(response.status)
print(response.undefined_flags.get_flags())
print(response.valid_flags.get_flags())
print(response.invalid_value_flags.get_flags())
@work_router.command("run")
def command_start_solving(response: Response):
print(response.status)
print(response.undefined_flags.get_flags())
print(response.valid_flags.get_flags())
print(response.invalid_value_flags.get_flags())
+16 -14
View File
@@ -1,15 +1,16 @@
[project]
name = "argenta"
version = "0.5.0"
description = "Python library for creating TUI"
authors = [
{name = "kolo", email = "kolo.is.main@gmail.com"}
]
license = {text = "MIT"}
version = "1.0.7"
description = "Python library for building modular CLI applications"
authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }]
requires-python = ">=3.8"
readme = "README.md"
requires-python = ">=3.11"
dependencies = ["rich (>=14.0.0,<15.0.0)", "art (>=6.4,<7.0)"]
license = { text = "MIT" }
dependencies = [
"rich (>=14.0.0,<15.0.0)",
"art (>=6.4,<7.0)",
"pyreadline3>=3.5.4",
]
[tool.ruff]
exclude = [
@@ -21,11 +22,12 @@ exclude = [
"tests"
]
[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.poetry.group.dev.dependencies]
pyreadline3 = "^3.5.4"
[dependency-groups]
dev = [
"psutil>=7.0.0",
]
+91
View File
@@ -0,0 +1,91 @@
import os
import readline
from typing import Never
class AutoCompleter:
def __init__(
self, history_filename: str | None = None, autocomplete_button: str = "tab"
) -> None:
"""
Public. Configures and implements auto-completion of input command
:param history_filename: the name of the file for saving the history of the autocompleter
:param autocomplete_button: the button for auto-completion
:return: None
"""
self.history_filename = history_filename
self.autocomplete_button = autocomplete_button
def _complete(self, text, state) -> str | None:
"""
Private. Auto-completion function
:param text: part of the command being entered
:param state: the current cursor position is relative to the beginning of the line
:return: the desired candidate as str or None
"""
matches: list[str] = sorted(
cmd for cmd in self.get_history_items() if cmd.startswith(text)
)
if len(matches) > 1:
common_prefix = matches[0]
for match in matches[1:]:
i = 0
while (
i < len(common_prefix)
and i < len(match)
and common_prefix[i] == match[i]
):
i += 1
common_prefix = common_prefix[:i]
if state == 0:
readline.insert_text(common_prefix[len(text) :])
readline.redisplay()
return None
elif len(matches) == 1:
return matches[0] if state == 0 else None
else:
return None
def initial_setup(self, all_commands: list[str]) -> None:
"""
Private. Initial setup function
:param all_commands: Registered commands for adding them to the autocomplete history
:return: None
"""
if self.history_filename:
if os.path.exists(self.history_filename):
readline.read_history_file(self.history_filename)
else:
for line in all_commands:
readline.add_history(line)
readline.set_completer(self._complete)
readline.set_completer_delims(readline.get_completer_delims().replace(" ", ""))
readline.parse_and_bind(f"{self.autocomplete_button}: complete")
def exit_setup(self, all_commands: list[str]) -> None:
"""
Private. Exit setup function
:return: None
"""
if self.history_filename:
readline.write_history_file(self.history_filename)
with open(self.history_filename, "r") as history_file:
raw_history = history_file.read()
pretty_history: list[str] = []
for line in set(raw_history.strip().split("\n")):
if line.split()[0] in all_commands:
pretty_history.append(line)
with open(self.history_filename, "w") as history_file:
history_file.write("\n".join(pretty_history))
@staticmethod
def get_history_items() -> list[str] | list[Never]:
"""
Private. Returns a list of all commands entered by the user
:return: all commands entered by the user as list[str] | list[Never]
"""
return [
readline.get_history_item(i)
for i in range(1, readline.get_current_history_length() + 1)
]
+11
View File
@@ -0,0 +1,11 @@
from enum import StrEnum
class PredefinedMessages(StrEnum):
"""
Public. A dataclass with predetermined messages for quick use
"""
USAGE = "[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]"
HELP = "[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]"
AUTOCOMPLETE = "[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>"
+66
View File
@@ -0,0 +1,66 @@
from abc import ABC
class BaseDividingLine(ABC):
def __init__(self, unit_part: str = "-") -> None:
"""
Private. The basic dividing line
:param unit_part: the single part of the dividing line
:return: None
"""
self._unit_part = unit_part
def get_unit_part(self) -> str:
"""
Private. Returns the unit part of the dividing line
:return: unit_part of dividing line as str
"""
if len(self._unit_part) == 0:
return " "
else:
return self._unit_part[0]
class StaticDividingLine(BaseDividingLine):
def __init__(self, unit_part: str = "-", length: int = 25) -> None:
"""
Public. The static dividing line
:param unit_part: the single part of the dividing line
:param length: the length of the dividing line
:return: None
"""
super().__init__(unit_part)
self.length = length
def get_full_static_line(self, is_override: bool) -> str:
"""
Private. Returns the full line of the dividing line
:param is_override: has the default text layout been redefined
:return: full line of dividing line as str
"""
if is_override:
return f"\n{self.length * self.get_unit_part()}\n"
else:
return f"\n[dim]{self.length * self.get_unit_part()}[/dim]\n"
class DynamicDividingLine(BaseDividingLine):
def __init__(self, unit_part: str = "-") -> None:
"""
Public. The dynamic dividing line
:param unit_part: the single part of the dividing line
:return: None
"""
super().__init__(unit_part)
def get_full_dynamic_line(self, length: int, is_override: bool) -> str:
"""
Private. Returns the full line of the dividing line
:param length: the length of the dividing line
:param is_override: has the default text layout been redefined
:return: full line of dividing line as str
"""
if is_override:
return f"\n{length * self.get_unit_part()}\n"
else:
return f"\n[dim]{self.get_unit_part() * length}[/dim]\n"
+460
View File
@@ -0,0 +1,460 @@
from typing import Callable
from rich.console import Console
from rich.markup import escape
from art import text2art
from contextlib import redirect_stdout
import io
import re
from argenta.command.models import Command, InputCommand
from argenta.router import Router
from argenta.router.defaults import system_router
from argenta.app.autocompleter import AutoCompleter
from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine
from argenta.command.exceptions import (
UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException,
BaseInputCommandException,
)
from argenta.app.registered_routers.entity import RegisteredRouters
from argenta.response import Response
class BaseApp:
def __init__(self, prompt: str,
initial_message: str,
farewell_message: str,
exit_command: Command,
system_router_title: str | None,
ignore_command_register: bool,
dividing_line: StaticDividingLine | DynamicDividingLine,
repeat_command_groups: bool,
override_system_messages: bool,
autocompleter: AutoCompleter,
print_func: Callable[[str], None]) -> None:
self._prompt = prompt
self._print_func = print_func
self._exit_command = exit_command
self._system_router_title = system_router_title
self._dividing_line = dividing_line
self._ignore_command_register = ignore_command_register
self._repeat_command_groups_description = repeat_command_groups
self._override_system_messages = override_system_messages
self._autocompleter = autocompleter
self._farewell_message = farewell_message
self._initial_message = initial_message
self._description_message_gen: Callable[[str, str], str] = lambda command, description: f"{command} *=*=* {description}"
self._registered_routers: RegisteredRouters = RegisteredRouters()
self._messages_on_startup: list[str] = []
self._matching_lower_triggers_with_routers: dict[str, Router] = {}
self._matching_default_triggers_with_routers: dict[str, Router] = {}
if self._ignore_command_register:
self._current_matching_triggers_with_routers: dict[str, Router] = self._matching_lower_triggers_with_routers
else:
self._current_matching_triggers_with_routers: dict[str, Router] = self._matching_default_triggers_with_routers
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:
"""
Public. Sets the output pattern of the available commands
:param _: output pattern of the available commands
:return: None
"""
self._description_message_gen: Callable[[str, str], str] = _
def set_incorrect_input_syntax_handler(self, _: Callable[[str], None]) -> None:
"""
Public. Sets the handler for incorrect flags when entering a command
:param _: handler for incorrect flags when entering a command
:return: None
"""
self._incorrect_input_syntax_handler = _
def set_repeated_input_flags_handler(self, _: Callable[[str], None]) -> None:
"""
Public. Sets the handler for repeated flags when entering a command
:param _: handler for repeated flags when entering a command
:return: None
"""
self._repeated_input_flags_handler = _
def set_unknown_command_handler(self, _: Callable[[str], None]) -> None:
"""
Public. Sets the handler for unknown commands when entering a command
:param _: handler for unknown commands when entering a command
:return: None
"""
self._unknown_command_handler = _
def set_empty_command_handler(self, _: Callable[[], None]) -> None:
"""
Public. Sets the handler for empty commands when entering a command
:param _: handler for empty commands when entering a command
:return: None
"""
self._empty_input_command_handler = _
def set_exit_command_handler(self, _: Callable[[], None]) -> None:
"""
Public. Sets the handler for exit command when entering a command
:param _: handler for exit command when entering a command
:return: None
"""
self._exit_command_handler = _
def _print_command_group_description(self) -> None:
"""
Private. Prints the description of the available commands
:return: None
"""
for registered_router in self._registered_routers:
if registered_router.title:
self._print_func(registered_router.title)
for command_handler in registered_router.get_command_handlers():
self._print_func(
self._description_message_gen(
command_handler.get_handled_command().get_trigger(),
command_handler.get_handled_command().get_description(),
)
)
self._print_func("")
def _print_framed_text(self, text: str) -> None:
"""
Private. Outputs text by framing it in a static or dynamic split strip
:param text: framed text
:return: None
"""
if isinstance(self._dividing_line, StaticDividingLine):
self._print_func(
self._dividing_line.get_full_static_line(self._override_system_messages)
)
print(text.strip("\n"))
self._print_func(
self._dividing_line.get_full_static_line(self._override_system_messages)
)
elif isinstance(self._dividing_line, DynamicDividingLine):
clear_text = re.sub(r"\u001b\[[0-9;]*m", "", text)
max_length_line = max([len(line) for line in clear_text.split("\n")])
max_length_line = (
max_length_line
if 10 <= max_length_line <= 80
else 80
if max_length_line > 80
else 10
)
self._print_func(
self._dividing_line.get_full_dynamic_line(
max_length_line, self._override_system_messages
)
)
print(text.strip("\n"))
self._print_func(
self._dividing_line.get_full_dynamic_line(
max_length_line, self._override_system_messages
)
)
def _is_exit_command(self, command: InputCommand) -> bool:
"""
Private. Checks if the given command is an exit command
:param command: command to check
:return: is it an exit command or not as bool
"""
if self._ignore_command_register:
if (
command.get_trigger().lower()
== self._exit_command.get_trigger().lower()
):
return True
elif command.get_trigger().lower() in [
x.lower() for x in self._exit_command.get_aliases()
]:
return True
else:
if command.get_trigger() == self._exit_command.get_trigger():
return True
elif command.get_trigger() in self._exit_command.get_aliases():
return True
return False
def _is_unknown_command(self, command: InputCommand) -> bool:
"""
Private. Checks if the given command is an unknown command
:param command: command to check
:return: is it an unknown command or not as bool
"""
input_command_trigger = command.get_trigger()
if self._ignore_command_register:
if input_command_trigger.lower() in list(self._current_matching_triggers_with_routers.keys()):
return False
else:
if input_command_trigger in list(self._current_matching_triggers_with_routers.keys()):
return False
return True
def _error_handler(
self, error: BaseInputCommandException, raw_command: str
) -> None:
"""
Private. Handles parsing errors of the entered command
:param error: error being handled
:param raw_command: the raw input command
:return: None
"""
if isinstance(error, UnprocessedInputFlagException):
self._incorrect_input_syntax_handler(raw_command)
elif isinstance(error, RepeatedInputFlagsException):
self._repeated_input_flags_handler(raw_command)
elif isinstance(error, EmptyInputCommandException):
self._empty_input_command_handler()
def _setup_system_router(self) -> None:
"""
Private. Sets up system router
:return: None
"""
system_router.title = self._system_router_title
@system_router.command(self._exit_command)
def exit_command(response: Response) -> None:
self._exit_command_handler(response)
if system_router not in self._registered_routers.get_registered_routers():
system_router.set_command_register_ignore(self._ignore_command_register)
self._registered_routers.add_registered_router(system_router)
def _most_similar_command(self, unknown_command: str) -> str | None:
all_commands = list(self._current_matching_triggers_with_routers.keys())
matches: list[str] | list = sorted(
cmd for cmd in all_commands if cmd.startswith(unknown_command)
)
if not matches:
matches: list[str] | list = sorted(
cmd for cmd in all_commands if unknown_command.startswith(cmd)
)
if len(matches) == 1:
return matches[0]
elif len(matches) > 1:
return sorted(matches, key=lambda cmd: len(cmd))[0]
else:
return None
def _setup_default_view(self) -> None:
"""
Private. Sets up default app view
:return: None
"""
self._prompt = f"[italic dim bold]{self._prompt}"
self._initial_message = ("\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n")
self._farewell_message = (
"[bold red]\n\n"
+ text2art(self._farewell_message, font="chanky")
+ "\n[/bold red]\n"
"[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: (
f"[bold red]{escape('[' + command + ']')}[/bold red] "
f"[blue dim]*=*=*[/blue dim] "
f"[bold yellow italic]{escape(description)}"
)
self._incorrect_input_syntax_handler = lambda raw_command: self._print_func(f"[red bold]Incorrect flag syntax: {escape(raw_command)}")
self._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:
cmd_trg: str = command.get_trigger()
mst_sim_cmd: str | None = self._most_similar_command(cmd_trg)
first_part_of_text = f"[red]Unknown command:[/red] [blue]{escape(cmd_trg)}[/blue]"
second_part_of_text = (
("[red], most similar:[/red] " + ("[blue]" + mst_sim_cmd + "[/blue]"))
if mst_sim_cmd
else ""
)
self._print_func(first_part_of_text + second_part_of_text)
self._unknown_command_handler = unknown_command_handler
def pre_cycle_setup(self) -> None:
"""
Private. Configures various aspects of the application before the start of the cycle
:return: None
"""
self._setup_system_router()
for router_entity in self._registered_routers:
router_triggers = router_entity.get_triggers()
router_aliases = router_entity.get_aliases()
combined = router_triggers + router_aliases
for trigger in combined:
self._matching_default_triggers_with_routers[trigger] = router_entity
self._matching_lower_triggers_with_routers[trigger.lower()] = router_entity
self._autocompleter.initial_setup(list(self._current_matching_triggers_with_routers.keys()))
seen = {}
for item in list(self._current_matching_triggers_with_routers.keys()):
if item in seen:
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{item}[/b blue]")
else:
seen[item] = True
if not self._override_system_messages:
self._setup_default_view()
self._print_func(self._initial_message)
for message in self._messages_on_startup:
self._print_func(message)
if self._messages_on_startup:
print("\n")
if not self._repeat_command_groups_description:
self._print_command_group_description()
class App(BaseApp):
def __init__(
self,
prompt: str = "What do you want to do?\n",
initial_message: str = "Argenta\n",
farewell_message: str = "\nSee you\n",
exit_command: Command = Command("Q", "Exit command"),
system_router_title: str | None = "System points:",
ignore_command_register: bool = True,
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
repeat_command_groups: bool = True,
override_system_messages: bool = False,
autocompleter: AutoCompleter = AutoCompleter(),
print_func: Callable[[str], None] = Console().print,
) -> None:
"""
Public. The essence of the application itself.
Configures and manages all aspects of the behavior and presentation of the user interacting with the user
:param prompt: displayed before entering the command
:param initial_message: displayed at the start of the app
:param farewell_message: displayed at the end of the app
:param exit_command: the entity of the command that will be terminated when entered
:param system_router_title: system router title
:param ignore_command_register: whether to ignore the case of the entered commands
:param dividing_line: the entity of the dividing line
:param repeat_command_groups: whether to repeat the available commands and their description
:param override_system_messages: whether to redefine the default formatting of system messages
:param autocompleter: the entity of the autocompleter
:param print_func: system messages text output function
:return: None
"""
super().__init__(
prompt=prompt,
initial_message=initial_message,
farewell_message=farewell_message,
exit_command=exit_command,
system_router_title=system_router_title,
ignore_command_register=ignore_command_register,
dividing_line=dividing_line,
repeat_command_groups=repeat_command_groups,
override_system_messages=override_system_messages,
autocompleter=autocompleter,
print_func=print_func,
)
def run_polling(self) -> None:
"""
Private. Starts the user input processing cycle
:return: None
"""
self.pre_cycle_setup()
while True:
if self._repeat_command_groups_description:
self._print_command_group_description()
raw_command: str = Console().input(self._prompt)
try:
input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
except BaseInputCommandException as error:
with redirect_stdout(io.StringIO()) as f:
self._error_handler(error, raw_command)
res: str = f.getvalue()
self._print_framed_text(res)
continue
if self._is_exit_command(input_command):
system_router.finds_appropriate_handler(input_command)
self._autocompleter.exit_setup(list(self._current_matching_triggers_with_routers.keys()))
return
if self._is_unknown_command(input_command):
with redirect_stdout(io.StringIO()) as f:
self._unknown_command_handler(input_command)
res: str = f.getvalue()
self._print_framed_text(res)
continue
processing_router = self._current_matching_triggers_with_routers[input_command.get_trigger().lower()]
if processing_router.disable_redirect_stdout:
if isinstance(self._dividing_line, StaticDividingLine):
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:
"""
Public. Registers the router in the application
:param router: registered router
:return: None
"""
router.set_command_register_ignore(self._ignore_command_register)
self._registered_routers.add_registered_router(router)
def include_routers(self, *routers: Router) -> None:
"""
Public. Registers the routers in the application
:param routers: registered routers
:return: None
"""
for router in routers:
self.include_router(router)
def add_message_on_startup(self, message: str) -> None:
"""
Public. Adds a message that will be displayed when the application is launched
:param message: the message being added
:return: None
"""
self._messages_on_startup.append(message)
@@ -0,0 +1,34 @@
from typing import Iterator
from argenta.router import Router
class RegisteredRouters:
def __init__(self, registered_routers: list[Router] | None = None) -> None:
"""
Private. Combines registered routers
:param registered_routers: list of the registered routers
:return: None
"""
self._registered_routers = registered_routers if registered_routers else []
def get_registered_routers(self) -> list[Router]:
"""
Private. Returns the registered routers
:return: registered routers as list[Router]
"""
return self._registered_routers
def add_registered_router(self, router: Router) -> None:
"""
Private. Adds a new registered router
:param router: registered router
:return: None
"""
self._registered_routers.append(router)
def __iter__(self) -> Iterator[Router]:
return iter(self._registered_routers)
def __next__(self) -> Router:
return next(iter(self._registered_routers))
+3
View File
@@ -0,0 +1,3 @@
__all__ = ["Command"]
from argenta.command.models import Command
+42
View File
@@ -0,0 +1,42 @@
from argenta.command.flag.models import Flag, InputFlag
class BaseInputCommandException(Exception):
"""
Private. Base exception class for all exceptions raised when parse input command
"""
pass
class UnprocessedInputFlagException(BaseInputCommandException):
"""
Private. Raised when an unprocessed input flag is detected
"""
def __str__(self):
return "Unprocessed Input Flags"
class RepeatedInputFlagsException(BaseInputCommandException):
"""
Private. Raised when repeated input flags are detected
"""
def __init__(self, flag: Flag | InputFlag):
self.flag = flag
def __str__(self):
return (
"Repeated Input Flags\n"
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'"
)
class EmptyInputCommandException(BaseInputCommandException):
"""
Private. Raised when an empty input command is detected
"""
def __str__(self):
return "Input Command is empty"
+17
View File
@@ -0,0 +1,17 @@
__all__ = [
"Flag",
"InputFlag",
"UndefinedInputFlags",
"ValidInputFlags",
"InvalidValueInputFlags",
"Flags", "PossibleValues"
]
from argenta.command.flag.models import Flag, InputFlag, PossibleValues
from argenta.command.flag.flags.models import (
UndefinedInputFlags,
ValidInputFlags,
Flags,
InvalidValueInputFlags,
)
+32
View File
@@ -0,0 +1,32 @@
from dataclasses import dataclass
from argenta.command.flag.models import Flag, PossibleValues
import re
@dataclass
class PredefinedFlags:
"""
Public. A dataclass with predefined flags and most frequently used flags for quick use
"""
HELP = Flag(name="help", possible_values=PossibleValues.DISABLE)
SHORT_HELP = Flag(name="H", prefix="-", possible_values=PossibleValues.DISABLE)
INFO = Flag(name="info", possible_values=PossibleValues.DISABLE)
SHORT_INFO = Flag(name="I", prefix="-", possible_values=PossibleValues.DISABLE)
ALL = Flag(name="all", possible_values=PossibleValues.DISABLE)
SHORT_ALL = Flag(name="A", prefix="-", possible_values=PossibleValues.DISABLE)
HOST = Flag(
name="host", possible_values=re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
)
SHORT_HOST = Flag(
name="H",
prefix="-",
possible_values=re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$"),
)
PORT = Flag(name="port", possible_values=re.compile(r"^\d{1,5}$"))
SHORT_PORT = Flag(name="P", prefix="-", possible_values=re.compile(r"^\d{1,5}$"))
@@ -0,0 +1,16 @@
__all__ = [
"Flags",
"InputFlags",
"UndefinedInputFlags",
"InvalidValueInputFlags",
"ValidInputFlags",
]
from argenta.command.flag.flags.models import (
Flags,
InputFlags,
UndefinedInputFlags,
InvalidValueInputFlags,
ValidInputFlags,
)
+90
View File
@@ -0,0 +1,90 @@
from argenta.command.flag.models import InputFlag, Flag
from typing import Generic, TypeVar
FlagType = TypeVar("FlagType")
class BaseFlags(Generic[FlagType]):
def __init__(self, *flags: FlagType):
"""
Public. A model that combines the registered flags
:param flags: the flags that will be registered
:return: None
"""
self._flags = flags if flags else []
def get_flags(self) -> list[FlagType]:
"""
Public. Returns a list of flags
:return: list of flags as list[FlagType]
"""
return self._flags
def add_flag(self, flag: FlagType):
"""
Public. Adds a flag to the list of flags
:param flag: flag to add
:return: None
"""
self._flags.append(flag)
def add_flags(self, flags: list[FlagType]):
"""
Public. Adds a list of flags to the list of flags
:param flags: list of flags to add
:return: None
"""
self._flags.extend(flags)
def get_flag(self, name: str) -> FlagType | None:
"""
Public. Returns the flag entity by its name or None if not found
:param name: the name of the flag to get
:return: entity of the flag or None
"""
if name in [flag.get_name() for flag in self._flags]:
return list(filter(lambda flag: flag.get_name() == name, self._flags))[0]
else:
return None
def __iter__(self):
return iter(self._flags)
def __next__(self):
return next(iter(self))
def __getitem__(self, item):
return self._flags[item]
def __bool__(self):
return bool(self._flags)
def __eq__(self, other):
if len(self.get_flags()) != len(other.get_flags()):
return False
else:
for flag, other_flag in zip(self.get_flags(), other.get_flags()):
if not flag == other_flag:
return False
return True
class Flags(BaseFlags[Flag]):
pass
class InputFlags(BaseFlags[InputFlag]):
pass
class ValidInputFlags(InputFlags):
pass
class UndefinedInputFlags(InputFlags):
pass
class InvalidValueInputFlags(InputFlags):
pass
+134
View File
@@ -0,0 +1,134 @@
from enum import Enum
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:
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--") -> None:
"""
Private. Base class for flags
:param name: the name of the flag
:param prefix: the prefix of the flag
:return: None
"""
self._name = name
self._prefix = prefix
def get_string_entity(self) -> str:
"""
Public. Returns a string representation of the flag
:return: string representation of the flag as str
"""
string_entity: str = self._prefix + self._name
return string_entity
def get_name(self) -> str:
"""
Public. Returns the name of the flag
:return: the name of the flag as str
"""
return self._name
def get_prefix(self) -> str:
"""
Public. Returns the prefix of the flag
:return: the prefix of the flag as str
"""
return self._prefix
def __eq__(self, other) -> bool:
return self.get_string_entity() == other.get_string_entity()
class Flag(BaseFlag):
def __init__(
self,
name: str,
prefix: Literal["-", "--", "---"] = "--",
possible_values: list[str] | Pattern[str] | PossibleValues = PossibleValues.ALL,
) -> None:
"""
Public. The entity of the flag being registered for subsequent processing
:param name: The name of the flag
:param prefix: The prefix of the flag
:param possible_values: The possible values of the flag, if False then the flag cannot have a value
:return: None
"""
super().__init__(name, prefix)
self.possible_values = possible_values
def validate_input_flag_value(self, input_flag_value: str | None):
"""
Private. Validates the input flag value
:param input_flag_value: The input flag value to validate
:return: whether the entered flag is valid as bool
"""
if self.possible_values == PossibleValues.DISABLE:
if input_flag_value is None:
return True
else:
return False
elif isinstance(self.possible_values, Pattern):
if isinstance(input_flag_value, str):
is_valid = bool(self.possible_values.match(input_flag_value))
if bool(is_valid):
return True
else:
return False
else:
return False
elif isinstance(self.possible_values, list):
if input_flag_value in self.possible_values:
return True
else:
return False
else:
return True
class InputFlag(BaseFlag):
def __init__(
self,
name: str,
prefix: Literal["-", "--", "---"] = "--",
value: str | None = None,
):
"""
Public. The entity of the flag of the entered command
:param name: the name of the input flag
:param prefix: the prefix of the input flag
:param value: the value of the input flag
:return: None
"""
super().__init__(name, prefix)
self._flag_value = value
def get_value(self) -> str | None:
"""
Public. Returns the value of the flag
:return: the value of the flag as str
"""
return self._flag_value
def set_value(self, value):
"""
Private. Sets the value of the flag
:param value: the fag value to set
:return: None
"""
self._flag_value = value
def __eq__(self, other) -> bool:
return (
self.get_string_entity() == other.get_string_entity()
and self.get_value() == other.get_value()
)
+195
View File
@@ -0,0 +1,195 @@
from argenta.command.flag.models import Flag, InputFlag
from argenta.command.flag.flags.models import InputFlags, Flags
from argenta.command.exceptions import (
UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException,
)
from typing import cast, Literal
class BaseCommand:
def __init__(self, trigger: str) -> None:
"""
Private. Base class for all commands
:param trigger: A string trigger, which, when entered by the user, indicates that the input corresponds to the command
"""
self._trigger = trigger
def get_trigger(self) -> str:
"""
Public. Returns the trigger of the command
:return: the trigger of the command as str
"""
return self._trigger
class Command(BaseCommand):
def __init__(
self,
trigger: str,
description: str | None = None,
flags: Flag | Flags | None = None,
aliases: list[str] | None = None,
):
"""
Public. The command that can and should be registered in the Router
:param trigger: A string trigger, which, when entered by the user, indicates that the input corresponds to the command
:param description: the description of the command
:param flags: processed commands
:param aliases: string synonyms for the main trigger
"""
super().__init__(trigger)
self._registered_flags: Flags = (
flags
if isinstance(flags, Flags)
else Flags(flags)
if isinstance(flags, Flag)
else Flags()
)
self._description = "Very useful command" if not description else description
self._aliases = aliases if isinstance(aliases, list) else []
def get_registered_flags(self) -> Flags:
"""
Private. Returns the registered flags
:return: the registered flags as Flags
"""
return self._registered_flags
def get_aliases(self) -> list[str] | list:
"""
Public. Returns the aliases of the command
:return: the aliases of the command as list[str] | list
"""
return self._aliases
def validate_input_flag(
self, flag: InputFlag
) -> Literal["Undefined", "Valid", "Invalid"]:
"""
Private. Validates the input flag
:param flag: input flag for validation
:return: is input flag valid as bool
"""
registered_flags: Flags | None = self.get_registered_flags()
if registered_flags:
if isinstance(registered_flags, Flag):
if registered_flags.get_string_entity() == flag.get_string_entity():
is_valid = registered_flags.validate_input_flag_value(
flag.get_value()
)
if is_valid:
return "Valid"
else:
return "Invalid"
else:
return "Undefined"
else:
for registered_flag in registered_flags:
if registered_flag.get_string_entity() == flag.get_string_entity():
is_valid = registered_flag.validate_input_flag_value(
flag.get_value()
)
if is_valid:
return "Valid"
else:
return "Invalid"
return "Undefined"
return "Undefined"
def get_description(self) -> str:
"""
Private. Returns the description of the command
:return: the description of the command as str
"""
return self._description
class InputCommand(BaseCommand):
def __init__(self, trigger: str, input_flags: InputFlag | InputFlags | None = None):
"""
Private. The model of the input command, after parsing
:param trigger:the trigger of the command
:param input_flags: the input flags
:return: None
"""
super().__init__(trigger)
self._input_flags: InputFlags = (
input_flags
if isinstance(input_flags, InputFlags)
else InputFlags(input_flags)
if isinstance(input_flags, InputFlag)
else InputFlags()
)
def _set_input_flags(self, input_flags: InputFlags) -> None:
"""
Private. Sets the input flags
:param input_flags: the input flags to set
:return: None
"""
self._input_flags = input_flags
def get_input_flags(self) -> InputFlags:
"""
Private. Returns the input flags
:return: the input flags as InputFlags
"""
return self._input_flags
@staticmethod
def parse(raw_command: str) -> "InputCommand":
"""
Private. Parse the raw input command
:param raw_command: raw input command
:return: model of the input command, after parsing as InputCommand
"""
if not raw_command:
raise EmptyInputCommandException()
list_of_tokens = raw_command.split()
command = list_of_tokens.pop(0)
input_flags: InputFlags = InputFlags()
current_flag_name, current_flag_value = None, None
for k, _ in enumerate(list_of_tokens):
if _.startswith("-"):
if len(_) < 2 or len(_[: _.rfind("-")]) > 3:
raise UnprocessedInputFlagException()
current_flag_name = _
else:
if not current_flag_name or current_flag_value:
raise UnprocessedInputFlagException()
current_flag_value = _
if current_flag_name:
if not len(list_of_tokens) == k + 1:
if not list_of_tokens[k + 1].startswith("-"):
continue
input_flag = InputFlag(
name=current_flag_name[current_flag_name.rfind("-") + 1 :],
prefix=cast(
Literal["-", "--", "---"],
current_flag_name[: current_flag_name.rfind("-") + 1],
),
value=current_flag_value,
)
all_flags = [
flag.get_string_entity() for flag in input_flags.get_flags()
]
if input_flag.get_string_entity() not in all_flags:
input_flags.add_flag(input_flag)
else:
raise RepeatedInputFlagsException(input_flag)
current_flag_name, current_flag_value = None, None
if any([current_flag_name, current_flag_value]):
raise UnprocessedInputFlagException()
else:
return InputCommand(trigger=command, input_flags=input_flags)
+4
View File
@@ -0,0 +1,4 @@
__all__ = ["get_time_of_pre_cycle_setup"]
from argenta.metrics.main import get_time_of_pre_cycle_setup
+18
View File
@@ -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
+4
View File
@@ -0,0 +1,4 @@
__all__ = ["Orchestrator"]
from argenta.orchestrator.entity import Orchestrator
@@ -0,0 +1,4 @@
__all__ = ["ArgParser"]
from argenta.orchestrator.argparser.entity import ArgParser
@@ -0,0 +1,8 @@
__all__ = ["BooleanArgument", "PositionalArgument", "OptionalArgument"]
from argenta.orchestrator.argparser.arguments.models import (
BooleanArgument,
PositionalArgument,
OptionalArgument,
)
@@ -0,0 +1,56 @@
from abc import ABC, abstractmethod
from typing import Literal
class BaseArgument(ABC):
"""
Private. Base class for all arguments
"""
@abstractmethod
def get_string_entity(self) -> str:
"""
Public. Returns the string representation of the argument
:return: the string representation as a str
"""
pass
class PositionalArgument(BaseArgument):
def __init__(self, name: str):
"""
Public. Required argument at startup
:param name: name of the argument, must not start with minus (-)
"""
self.name = name
def get_string_entity(self):
return self.name
class OptionalArgument(BaseArgument):
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--"):
"""
Public. Optional argument, must have the value
:param name: name of the argument
:param prefix: prefix of the argument
"""
self.name = name
self.prefix = prefix
def get_string_entity(self):
return self.prefix + self.name
class BooleanArgument(BaseArgument):
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--"):
"""
Public. Boolean argument, does not require a value
:param name: name of the argument
:param prefix: prefix of the argument
"""
self.name = name
self.prefix = prefix
def get_string_entity(self):
return self.prefix + self.name
@@ -0,0 +1,59 @@
from argparse import ArgumentParser
from argenta.orchestrator.argparser.arguments.models import (
BooleanArgument,
OptionalArgument,
PositionalArgument,
)
class ArgParser:
def __init__(
self,
processed_args: list[PositionalArgument | OptionalArgument | BooleanArgument],
name: str = "Argenta",
description: str = "Argenta available arguments",
epilog: str = "github.com/koloideal/Argenta | made by kolo",
) -> None:
"""
Public. Cmd argument parser and configurator at startup
:param name: the name of the ArgParse instance
:param description: the description of the ArgParse instance
:param epilog: the epilog of the ArgParse instance
:param processed_args: registered and processed arguments
"""
self.name = name
self.description = description
self.epilog = epilog
self.entity: ArgumentParser = ArgumentParser(
prog=name, description=description, epilog=epilog
)
self.args: (
list[PositionalArgument | OptionalArgument | BooleanArgument] | None
) = processed_args
def set_args(
self, *args: PositionalArgument | OptionalArgument | BooleanArgument
) -> None:
"""
Public. Sets the arguments to be processed
:param args: processed arguments
:return: None
"""
self.args.extend(args)
def register_args(self) -> None:
"""
Private. Registers initialized command line arguments
:return: None
"""
if not self.args:
return
for arg in self.args:
if type(arg) is PositionalArgument:
self.entity.add_argument(arg.get_string_entity())
elif type(arg) is OptionalArgument:
self.entity.add_argument(arg.get_string_entity())
elif type(arg) is BooleanArgument:
self.entity.add_argument(arg.get_string_entity(), action="store_true")
+35
View File
@@ -0,0 +1,35 @@
from argparse import Namespace
from argenta.app import App
from argenta.orchestrator.argparser import ArgParser
class Orchestrator:
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
:param arg_parser: Cmd argument parser and configurator at startup
:return: None
"""
self.arg_parser: ArgParser | None = arg_parser
if arg_parser:
self.arg_parser.register_args()
@staticmethod
def start_polling(app: App) -> None:
"""
Public. Starting the user input processing cycle
:param app: a running application
:return: None
"""
app.run_polling()
def get_input_args(self) -> Namespace | None:
"""
Public. Returns the arguments parsed
:return: None
"""
if self.arg_parser:
return self.arg_parser.entity.parse_args()
else:
return None
+5
View File
@@ -0,0 +1,5 @@
__all__ = ["Response", "Status"]
from argenta.response.entity import Response
from argenta.response.status import Status
+29
View File
@@ -0,0 +1,29 @@
from argenta.response.status import Status
from argenta.command.flag.flags import (
ValidInputFlags,
UndefinedInputFlags,
InvalidValueInputFlags,
)
class Response:
__slots__ = ("status", "valid_flags", "undefined_flags", "invalid_value_flags")
def __init__(
self,
status: Status | None = None,
valid_flags: ValidInputFlags = ValidInputFlags(),
undefined_flags: UndefinedInputFlags = UndefinedInputFlags(),
invalid_value_flags: InvalidValueInputFlags = InvalidValueInputFlags(),
):
"""
Public. The entity of the user input sent to the handler
:param status: the status of the response
:param valid_flags: valid input flags
:param undefined_flags: undefined input flags
:param invalid_value_flags: input flags with invalid values
"""
self.status = status
self.valid_flags = valid_flags
self.undefined_flags = undefined_flags
self.invalid_value_flags = invalid_value_flags
+8
View File
@@ -0,0 +1,8 @@
from enum import Enum
class Status(Enum):
ALL_FLAGS_VALID = "ALL_FLAGS_VALID"
UNDEFINED_FLAGS = "UNDEFINED_FLAGS"
INVALID_VALUE_FLAGS = "INVALID_VALUE_FLAGS"
UNDEFINED_AND_INVALID_FLAGS = "UNDEFINED_AND_INVALID_FLAGS"
+4
View File
@@ -0,0 +1,4 @@
__all__ = ["Router"]
from argenta.router.entity import Router
@@ -0,0 +1,67 @@
from typing import Callable, Iterator
from argenta.command import Command
from argenta.response import Response
class CommandHandler:
def __init__(self, handler: Callable[[Response], None], handled_command: Command):
"""
Private. Entity of the model linking the handler and the command being processed
:param handler: the handler being called
:param handled_command: the command being processed
"""
self._handler = handler
self._handled_command = handled_command
def handling(self, response: Response) -> None:
"""
Private. Direct processing of an input command
:param response: the entity of response: various groups of flags and status of response
:return: None
"""
self._handler(response)
def get_handler(self) -> Callable[[Response], None]:
"""
Private. Returns the handler being called
:return: the handler being called as Callable[[Response], None]
"""
return self._handler
def get_handled_command(self) -> Command:
"""
Private. Returns the command being processed
:return: the command being processed as Command
"""
return self._handled_command
class CommandHandlers:
def __init__(self, command_handlers: list[CommandHandler] | None = None):
"""
Private. The model that unites all CommandHandler of the routers
:param command_handlers: list of CommandHandlers for register
"""
self.command_handlers = command_handlers if command_handlers else []
def get_handlers(self) -> list[CommandHandler]:
"""
Private. Returns the list of CommandHandlers
:return: the list of CommandHandlers as list[CommandHandler]
"""
return self.command_handlers
def add_handler(self, command_handler: CommandHandler) -> None:
"""
Private. Adds a CommandHandler to the list of CommandHandlers
:param command_handler: CommandHandler to be added
:return: None
"""
self.command_handlers.append(command_handler)
def __iter__(self) -> Iterator[CommandHandler]:
return iter(self.command_handlers)
def __next__(self) -> CommandHandler:
return next(iter(self.command_handlers))
+4
View File
@@ -0,0 +1,4 @@
from argenta.router import Router
system_router = Router(title="System points:")
+240
View File
@@ -0,0 +1,240 @@
from typing import Callable, Literal, Type
from inspect import getfullargspec, get_annotations, getsourcefile, getsourcelines
from rich.console import Console
from argenta.command import Command
from argenta.command.models import InputCommand
from argenta.response import Response, Status
from argenta.router.command_handler.entity import CommandHandlers, CommandHandler
from argenta.command.flag.flags import (
Flags,
InputFlags,
UndefinedInputFlags,
ValidInputFlags,
InvalidValueInputFlags,
)
from argenta.router.exceptions import (
RepeatedFlagNameException,
TooManyTransferredArgsException,
RequiredArgumentNotPassedException,
TriggerContainSpacesException,
)
class Router:
def __init__(
self, title: str | None = "Awesome title", disable_redirect_stdout: bool = False
):
"""
Public. Directly configures and manages handlers
:param title: the title of the router, displayed when displaying the available commands
: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
"""
self.title = title
self.disable_redirect_stdout = disable_redirect_stdout
self._command_handlers: CommandHandlers = CommandHandlers()
self._ignore_command_register: bool = False
def command(self, command: Command | str) -> Callable:
"""
Public. Registers handler
:param command: Registered command
:return: decorated handler as Callable
"""
if isinstance(command, str):
redefined_command = Command(command)
else:
redefined_command = command
self._validate_command(redefined_command)
def command_decorator(func):
Router._validate_func_args(func)
self._command_handlers.add_handler(CommandHandler(func, redefined_command))
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return command_decorator
def finds_appropriate_handler(self, input_command: InputCommand) -> None:
"""
Private. Finds the appropriate handler for given input command and passes control to it
:param input_command: input command as InputCommand
:return: None
"""
input_command_name: str = input_command.get_trigger()
input_command_flags: InputFlags = input_command.get_input_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():
self.process_input_command(input_command_flags, command_handler)
if input_command_name.lower() in handle_command.get_aliases():
self.process_input_command(input_command_flags, command_handler)
def process_input_command(
self, input_command_flags: InputFlags, command_handler: CommandHandler
) -> None:
"""
Private. Processes input command with the appropriate handler
:param input_command_flags: input command flags as InputFlags
:param command_handler: command handler for input command as CommandHandler
:return: None
"""
handle_command = command_handler.get_handled_command()
response: Response = Response()
if handle_command.get_registered_flags().get_flags():
if input_command_flags.get_flags():
response: Response = self._structuring_input_flags( handle_command, input_command_flags )
command_handler.handling(response)
else:
response.status = Status.ALL_FLAGS_VALID
command_handler.handling(response)
else:
if input_command_flags.get_flags():
response.status = Status.UNDEFINED_FLAGS
response.undefined_flags = UndefinedInputFlags()
response.undefined_flags.add_flags(input_command_flags.get_flags())
command_handler.handling(response)
else:
response.status = Status.ALL_FLAGS_VALID
command_handler.handling(response)
@staticmethod
def _structuring_input_flags(
handled_command: Command, input_flags: InputFlags
) -> Response:
"""
Private. Validates flags of input command
:param handled_command: entity of the handled command
:param input_flags:
:return: entity of response as Response
"""
valid_input_flags: ValidInputFlags = ValidInputFlags()
invalid_value_input_flags: InvalidValueInputFlags = InvalidValueInputFlags()
undefined_input_flags: UndefinedInputFlags = UndefinedInputFlags()
for flag in input_flags:
flag_status: Literal["Undefined", "Valid", "Invalid"] = (
handled_command.validate_input_flag(flag)
)
if flag_status == "Valid":
valid_input_flags.add_flag(flag)
elif flag_status == "Undefined":
undefined_input_flags.add_flag(flag)
elif flag_status == "Invalid":
invalid_value_input_flags.add_flag(flag)
if (
not invalid_value_input_flags.get_flags()
and not undefined_input_flags.get_flags()
):
status = Status.ALL_FLAGS_VALID
elif (
invalid_value_input_flags.get_flags()
and not undefined_input_flags.get_flags()
):
status = Status.INVALID_VALUE_FLAGS
elif (
not invalid_value_input_flags.get_flags()
and undefined_input_flags.get_flags()
):
status = Status.UNDEFINED_FLAGS
else:
status = Status.UNDEFINED_AND_INVALID_FLAGS
return Response(
invalid_value_flags=invalid_value_input_flags,
valid_flags=valid_input_flags,
status=status,
undefined_flags=undefined_input_flags,
)
@staticmethod
def _validate_command(command: Command) -> None:
"""
Private. Validates the command registered in handler
:param command: validated command
:return: None if command is valid else raise exception
"""
command_name: str = command.get_trigger()
if command_name.find(" ") != -1:
raise TriggerContainSpacesException()
flags: Flags = command.get_registered_flags()
if flags:
flags_name: list = [x.get_string_entity().lower() for x in flags]
if len(set(flags_name)) < len(flags_name):
raise RepeatedFlagNameException()
@staticmethod
def _validate_func_args(func: Callable) -> None:
"""
Private. Validates the arguments of the handler
:param func: entity of the handler func
:return: None if func is valid else raise exception
"""
transferred_args = getfullargspec(func).args
if len(transferred_args) > 1:
raise TooManyTransferredArgsException()
elif len(transferred_args) == 0:
raise RequiredArgumentNotPassedException()
transferred_arg: str = transferred_args[0]
func_annotations: dict[str, Type] = get_annotations(func)
if arg_annotation := func_annotations.get(transferred_arg):
if arg_annotation is Response:
pass
else:
file_path: str | None = getsourcefile(func)
source_line: int = getsourcelines(func)[1]
fprint = Console().print
fprint(
f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
f"of argument([green]{transferred_arg}[/green]) passed to the handler is [/i][bold blue]{Response}[/bold blue],"
f" [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]",
highlight=False,
)
def set_command_register_ignore(self, _: bool) -> None:
"""
Private. Sets the router behavior on the input commands register
:param _: is command register ignore
:return: None
"""
self._ignore_command_register = _
def get_triggers(self) -> list[str]:
"""
Public. Gets registered triggers
:return: registered in router triggers as list[str]
"""
all_triggers: list[str] = []
for command_handler in self._command_handlers:
all_triggers.append(command_handler.get_handled_command().get_trigger())
return all_triggers
def get_aliases(self) -> list[str]:
"""
Public. Gets registered aliases
:return: registered in router aliases as list[str]
"""
all_aliases: list[str] = []
for command_handler in self._command_handlers:
if command_handler.get_handled_command().get_aliases():
all_aliases.extend(command_handler.get_handled_command().get_aliases())
return all_aliases
def get_command_handlers(self) -> CommandHandlers:
"""
Private. Gets registered command handlers
:return: registered command handlers as CommandHandlers
"""
return self._command_handlers
+34
View File
@@ -0,0 +1,34 @@
class RepeatedFlagNameException(Exception):
"""
Private. Raised when a repeated flag name is registered
"""
def __str__(self):
return "Repeated registered flag names in register command"
class TooManyTransferredArgsException(Exception):
"""
Private. Raised when too many arguments are passed
"""
def __str__(self):
return "Too many transferred arguments"
class RequiredArgumentNotPassedException(Exception):
"""
Private. Raised when a required argument is not passed
"""
def __str__(self):
return "Required argument not passed"
class TriggerContainSpacesException(Exception):
"""
Private. Raised when there is a space in the trigger being registered
"""
def __str__(self):
return "Command trigger cannot contain spaces"
@@ -1,32 +1,35 @@
import _io
from unittest.mock import patch, MagicMock
import unittest
from unittest import TestCase
import io
import re
from argenta.app import App
from argenta.command import Command
from argenta.router import Router
from argenta.command.flag.models import Flags, InputFlags
from argenta.command.flag.defaults import PredeterminedFlags
from argenta.command.flag.flags.models import Flags
from argenta.command.flag.defaults import PredefinedFlags
from argenta.orchestrator import Orchestrator
from argenta.response import Response
class TestSystemHandlerNormalWork(unittest.TestCase):
class TestSystemHandlerNormalWork(TestCase):
@patch("builtins.input", side_effect=["help", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response: Response):
print('test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -37,9 +40,10 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_incorrect_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response: Response):
print('test command')
app = App(ignore_command_register=False,
@@ -47,7 +51,7 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
print_func=print)
app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -58,74 +62,80 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_unregistered_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
print(f'test command')
def test(response: Response):
print(f'test command with undefined flag: {response.undefined_flags.get_flag('help').get_string_entity()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
self.assertIn('\nUndefined or incorrect input flag: --help\n', output)
self.assertIn('\ntest command with undefined flag: --help\n', output)
@patch("builtins.input", side_effect=["test --port 22", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_unregistered_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
print('test command')
def test(response: Response):
flag = response.undefined_flags.get_flag("port")
print(f'test command with undefined flag with value: {flag.get_string_entity()} {flag.get_value()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
self.assertIn('\nUndefined or incorrect input flag: --port 22\n', output)
self.assertIn('\ntest command with undefined flag with value: --port 22\n', output)
@patch("builtins.input", side_effect=["test --host 192.168.32.1 --port 132", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_one_correct_flag_an_one_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flags = Flags(PredeterminedFlags.HOST)
orchestrator = Orchestrator()
flags = Flags(PredefinedFlags.HOST)
@router.command(Command('test', flags=flags))
def test(args: InputFlags):
print(f'connecting to host {args.get_flag('host').get_value()}')
def test(response: Response):
flag = response.undefined_flags.get_flag("port")
print(f'connecting to host with flag: {flag.get_string_entity()} {flag.get_value()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
self.assertIn('\nUndefined or incorrect input flag: --port 132\n', output)
self.assertIn('\nconnecting to host with flag: --port 132\n', output)
@patch("builtins.input", side_effect=["test", "some", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_one_correct_command_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response: Response):
print(f'test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -136,20 +146,21 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_two_correct_commands_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response: Response):
print(f'test command')
@router.command(Command('more'))
def test():
def test(response: Response):
print(f'more command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -160,16 +171,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response: Response):
print(f'test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_invalid_input_flags_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
app.start_polling()
app.set_incorrect_input_syntax_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -180,16 +192,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_empty_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response: Response):
print(f'test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_empty_command_handler(lambda: print('Empty input command'))
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -200,16 +213,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_repeated_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test', flags=PredeterminedFlags.PORT))
def test(args: InputFlags):
@router.command(Command('test', flags=PredefinedFlags.PORT))
def test(response: Response):
print('test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_repeated_input_flags_handler(lambda command: print(f'Repeated input flags: "{command}"'))
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -1,31 +1,35 @@
import _io
from unittest.mock import patch, MagicMock
import unittest
from unittest import TestCase
import io
import re
from argenta.app import App
from argenta.command.models import Command
from argenta.command import Command
from argenta.response import Response
from argenta.router import Router
from argenta.command.flag.models import Flag, Flags, InputFlags
from argenta.command.flag.defaults import PredeterminedFlags
from argenta.orchestrator import Orchestrator
from argenta.command.flag import Flag
from argenta.command.flag.flags import Flags
from argenta.command.flag.defaults import PredefinedFlags
class TestSystemHandlerNormalWork(unittest.TestCase):
class TestSystemHandlerNormalWork(TestCase):
@patch("builtins.input", side_effect=["test", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response):
print('test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -36,16 +40,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response):
print('test command')
app = App(ignore_command_register=True,
override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -56,16 +61,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_custom_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
flag = Flag('help', '--', False)
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
print(f'\nhelp for {args.get_flag('help').get_name()} flag\n')
def test(response: Response):
print(f'\nhelp for {response.valid_flags.get_flag('help').get_name()} flag\n')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -75,57 +81,61 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_custom_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
flag = Flag('port', '--', re.compile(r'^\d{1,5}$'))
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
print(f'flag value for {args.get_flag('port').get_name()} flag : {args.get_flag('port').get_value()}')
def test(response: Response):
input_flag = response.valid_flags.get_flag('port')
print(f'flag value for {input_flag.get_name()} flag : {input_flag.get_value()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
self.assertIn('\nflag value for port flag : 22\n', output)
@patch("builtins.input", side_effect=["test -h", "q"])
@patch("builtins.input", side_effect=["test -H", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_default_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flag = PredeterminedFlags.SHORT_HELP
orchestrator = Orchestrator()
flag = PredefinedFlags.SHORT_HELP
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
print(f'help for {args.get_flag('h').get_name()} flag')
def test(response: Response):
print(f'help for {response.valid_flags.get_flag('H').get_name()} flag')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
self.assertIn('\nhelp for h flag\n', output)
self.assertIn('\nhelp for H flag\n', output)
@patch("builtins.input", side_effect=["test --info", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_default_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flag = PredeterminedFlags.INFO
orchestrator = Orchestrator()
flag = PredefinedFlags.INFO
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
if args.get_flag('info'):
def test(response: Response):
if response.valid_flags.get_flag('info'):
print('info about test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -136,16 +146,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_default_flag3(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flag = PredeterminedFlags.HOST
orchestrator = Orchestrator()
flag = PredefinedFlags.HOST
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
print(f'connecting to host {args[0].get_value()}')
def test(response: Response):
print(f'connecting to host {response.valid_flags.get_flag('host').get_value()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -156,16 +167,18 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_two_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flags = Flags(PredeterminedFlags.HOST, PredeterminedFlags.PORT)
orchestrator = Orchestrator()
flags = Flags(PredefinedFlags.HOST, PredefinedFlags.PORT)
@router.command(Command('test', flags=flags))
def test(args: InputFlags):
print(f'connecting to host {args[0].get_value()} and port {args[1].get_value()}')
def test(response: Response):
valid_flags = response.valid_flags
print(f'connecting to host {valid_flags.get_flag('host').get_value()} and port {valid_flags.get_flag('port').get_value()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -176,19 +189,20 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_two_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response):
print(f'test command')
@router.command(Command('some'))
def test2():
def test2(response):
print(f'some command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
@@ -199,23 +213,24 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_three_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test():
def test(response):
print(f'test command')
@router.command(Command('some'))
def test():
def test(response):
print(f'some command')
@router.command(Command('more'))
def test():
def test(response):
print(f'more command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.start_polling()
orchestrator.start_polling(app)
output = mock_stdout.getvalue()
+75
View File
@@ -0,0 +1,75 @@
from argenta.command.models import InputCommand, Command
from argenta.app import App
import unittest
from argenta.router import Router
class MyTestCase(unittest.TestCase):
def test_is_exit_command1(self):
app = App()
self.assertEqual(app._is_exit_command(InputCommand('q')), True)
def test_is_exit_command5(self):
app = App()
self.assertEqual(app._is_exit_command(InputCommand('Q')), True)
def test_is_exit_command2(self):
app = App(ignore_command_register=False)
self.assertEqual(app._is_exit_command(InputCommand('q')), False)
def test_is_exit_command3(self):
app = App(exit_command=Command('quit'))
self.assertEqual(app._is_exit_command(InputCommand('quit')), True)
def test_is_exit_command4(self):
app = App(exit_command=Command('quit'))
self.assertEqual(app._is_exit_command(InputCommand('qUIt')), True)
def test_is_exit_command6(self):
app = App(ignore_command_register=False,
exit_command=Command('quit'))
self.assertEqual(app._is_exit_command(InputCommand('qUIt')), False)
def test_is_unknown_command1(self):
app = App()
app.set_unknown_command_handler(lambda command: None)
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
self.assertEqual(app._is_unknown_command(InputCommand('fr')), False)
def test_is_unknown_command2(self):
app = App()
app.set_unknown_command_handler(lambda command: None)
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
self.assertEqual(app._is_unknown_command(InputCommand('cr')), True)
def test_is_unknown_command3(self):
app = App(ignore_command_register=False)
app.set_unknown_command_handler(lambda command: None)
app._current_matching_triggers_with_routers = {'Pr': Router(), 'tW': Router(), 'deQW': Router()}
self.assertEqual(app._is_unknown_command(InputCommand('pr')), True)
def test_is_unknown_command4(self):
app = App(ignore_command_register=False)
app.set_unknown_command_handler(lambda command: None)
app._current_matching_triggers_with_routers = {'Pr': Router(), 'tW': Router(), 'deQW': Router()}
self.assertEqual(app._is_unknown_command(InputCommand('tW')), False)
+40 -1
View File
@@ -1,9 +1,12 @@
from argenta.command.models import InputCommand
from argenta.command.flag import Flag, InputFlag
from argenta.command.flag.flags import Flags
from argenta.command.models import InputCommand, Command
from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException)
import unittest
import re
class TestInputCommand(unittest.TestCase):
@@ -22,3 +25,39 @@ class TestInputCommand(unittest.TestCase):
with self.assertRaises(EmptyInputCommandException):
InputCommand.parse('')
def test_validate_valid_input_flag1(self):
command = Command('some', flags=Flag('test'))
self.assertEqual(command.validate_input_flag(InputFlag('test')), 'Valid')
def test_validate_valid_input_flag2(self):
command = Command('some', flags=Flags(Flag('test'), Flag('more')))
self.assertEqual(command.validate_input_flag(InputFlag('more')), 'Valid')
def test_validate_undefined_input_flag1(self):
command = Command('some', flags=Flag('test'))
self.assertEqual(command.validate_input_flag(InputFlag('more')), 'Undefined')
def test_validate_undefined_input_flag2(self):
command = Command('some', flags=Flags(Flag('test'), Flag('more')))
self.assertEqual(command.validate_input_flag(InputFlag('case')), 'Undefined')
def test_validate_undefined_input_flag3(self):
command = Command('some')
self.assertEqual(command.validate_input_flag(InputFlag('case')), 'Undefined')
def test_invalid_input_flag1(self):
command = Command('some', flags=Flag('test', possible_values=False))
self.assertEqual(command.validate_input_flag(InputFlag('test', value='example')), 'Invalid')
def test_invalid_input_flag2(self):
command = Command('some', flags=Flag('test', possible_values=['some', 'case']))
self.assertEqual(command.validate_input_flag(InputFlag('test', value='slay')), 'Invalid')
def test_invalid_input_flag3(self):
command = Command('some', flags=Flag('test', possible_values=re.compile(r'^ex\d{, 2}op$')))
self.assertEqual(command.validate_input_flag(InputFlag('test', value='example')), 'Invalid')
def test_isinstance_parse_correct_raw_command(self):
cmd = InputCommand.parse('ssh --host 192.168.0.3')
self.assertIsInstance(cmd, InputCommand)
+2 -2
View File
@@ -6,11 +6,11 @@ import unittest
class TestDividingLine(unittest.TestCase):
def test_get_static_dividing_line_full_line(self):
line = StaticDividingLine('-')
self.assertEqual(line.get_full_line().count('-'), 25)
self.assertEqual(line.get_full_static_line(True).count('-'), 25)
def test_get_dynamic_dividing_line_full_line(self):
line = DynamicDividingLine()
self.assertEqual(line.get_full_line(20).count('-'), 20)
self.assertEqual(line.get_full_dynamic_line(20, True).count('-'), 20)
def test_get_dividing_line_unit_part(self):
line = StaticDividingLine('')
+44 -5
View File
@@ -1,4 +1,5 @@
from argenta.command.flag.models import Flag, InputFlag
from argenta.command.flag import Flag, InputFlag, PossibleValues
from argenta.command.flag.flags import InputFlags, Flags
import unittest
import re
@@ -53,21 +54,59 @@ class TestFlag(unittest.TestCase):
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
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)
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)
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)
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)
def test_get_input_flag1(self):
flag = InputFlag(name='test')
input_flags = InputFlags(flag)
self.assertEqual(input_flags.get_flag('test'), flag)
def test_get_input_flag2(self):
flag = InputFlag(name='test')
flag2 = InputFlag(name='some')
input_flags = InputFlags(flag, flag2)
self.assertEqual(input_flags.get_flag('some'), flag2)
def test_get_undefined_input_flag(self):
flag = InputFlag(name='test')
flag2 = InputFlag(name='some')
input_flags = InputFlags(flag, flag2)
self.assertEqual(input_flags.get_flag('case'), None)
def test_get_flags(self):
flags = Flags()
list_of_flags = [
Flag('test1'),
Flag('test2'),
Flag('test3'),
]
flags.add_flags(list_of_flags)
self.assertEqual(flags.get_flags(),
list_of_flags)
def test_add_flag(self):
flags = Flags()
flags.add_flag(Flag('test'))
self.assertEqual(len(flags.get_flags()), 1)
def test_add_flags(self):
flags = Flags()
flags.add_flags([Flag('test'), Flag('test2')])
self.assertEqual(len(flags.get_flags()), 2)
-26
View File
@@ -1,26 +0,0 @@
from argenta.command.flag.models import Flag, Flags
import unittest
class TestFlags(unittest.TestCase):
def test_get_flags(self):
flags = Flags()
list_of_flags = [
Flag('test1'),
Flag('test2'),
Flag('test3'),
]
flags.add_flags(list_of_flags)
self.assertEqual(flags.get_flags(),
list_of_flags)
def test_add_flag(self):
flags = Flags()
flags.add_flag(Flag('test'))
self.assertEqual(len(flags.get_flags()), 1)
def test_add_flags(self):
flags = Flags()
flags.add_flags([Flag('test'), Flag('test2')])
self.assertEqual(len(flags.get_flags()), 2)
+106 -18
View File
@@ -1,30 +1,118 @@
from argenta.command.flag import InputFlag, Flag
from argenta.command.flag.flags import Flags, InputFlags, UndefinedInputFlags, InvalidValueInputFlags, ValidInputFlags
from argenta.router import Router
from argenta.command import Command
from argenta.router.exceptions import TriggerCannotContainSpacesException
from argenta.router.exceptions import (TriggerContainSpacesException,
RepeatedFlagNameException,
TooManyTransferredArgsException,
RequiredArgumentNotPassedException)
import unittest
import re
class TestRouter(unittest.TestCase):
def test_get_router_name(self):
self.assertEqual(Router(name='test name').get_name(), 'test name')
def test_get_router_title(self):
self.assertEqual(Router(title='test title').get_title(), 'test title')
def test_register_command_with_spaces_in_trigger(self):
router = Router()
with self.assertRaises(TriggerCannotContainSpacesException):
@router.command(Command(trigger='command with spaces'))
def test():
return 'correct result'
with self.assertRaises(TriggerContainSpacesException):
router._validate_command(Command(trigger='command with spaces'))
def test_register_command_with_repeated_flags(self):
router = Router()
with self.assertRaises(RepeatedFlagNameException):
router._validate_command(Command(trigger='command', flags=Flags(Flag('test'), Flag('test'))))
def test_structuring_input_flags1(self):
router = Router()
cmd = Command('cmd')
input_flags = InputFlags(InputFlag('ssh'))
self.assertEqual(router._structuring_input_flags(cmd, input_flags).undefined_flags, UndefinedInputFlags(InputFlag('ssh')))
def test_structuring_input_flags2(self):
router = Router()
cmd = Command('cmd')
input_flags = InputFlags(InputFlag('ssh', value='some'))
self.assertEqual(router._structuring_input_flags(cmd, input_flags).undefined_flags, UndefinedInputFlags(InputFlag('ssh', value='some')))
def test_structuring_input_flags3(self):
router = Router()
cmd = Command('cmd', flags=Flag('port'))
input_flags = InputFlags(InputFlag('ssh', value='some2'))
self.assertEqual(router._structuring_input_flags(cmd, input_flags).undefined_flags, UndefinedInputFlags(InputFlag('ssh', value='some2')))
def test_structuring_input_flags4(self):
router = Router()
command = Command('cmd', flags=Flag('ssh', possible_values=False))
input_flags = InputFlags(InputFlag('ssh', value='some3'))
self.assertEqual(router._structuring_input_flags(command, input_flags).invalid_value_flags, InvalidValueInputFlags(InputFlag('ssh', value='some3')))
def test_structuring_input_flags5(self):
router = Router()
command = Command('cmd', flags=Flag('ssh', possible_values=re.compile(r'some[1-5]$')))
input_flags = InputFlags(InputFlag('ssh', value='some40'))
self.assertEqual(router._structuring_input_flags(command, input_flags).invalid_value_flags, InvalidValueInputFlags(InputFlag('ssh', value='some40')))
def test_structuring_input_flags6(self):
router = Router()
command = Command('cmd', flags=Flag('ssh', possible_values=['example']))
input_flags = InputFlags(InputFlag('ssh', value='example2'))
self.assertEqual(router._structuring_input_flags(command, input_flags).invalid_value_flags, InvalidValueInputFlags(InputFlag('ssh', value='example2')))
def test_structuring_input_flags7(self):
command = Command('cmd', flags=Flag('port'))
input_flags = InputFlags(InputFlag('port', value='some2'))
self.assertEqual(Router()._structuring_input_flags(command, input_flags).valid_flags, ValidInputFlags(InputFlag('port', value='some2')))
def test_structuring_input_flags8(self):
command = Command('cmd', flags=Flag('port', possible_values=['some2', 'some3']))
input_flags = InputFlags(InputFlag('port', value='some2'))
self.assertEqual(Router()._structuring_input_flags(command, input_flags).valid_flags, ValidInputFlags(InputFlag('port', value='some2')))
def test_structuring_input_flags9(self):
command = Command('cmd', flags=Flag('ssh', possible_values=re.compile(r'more[1-5]$')))
input_flags = InputFlags(InputFlag('ssh', value='more5'))
self.assertEqual(Router()._structuring_input_flags(command, input_flags).valid_flags, ValidInputFlags(InputFlag('ssh', value='more5')))
def test_structuring_input_flags10(self):
command = Command('cmd', flags=Flag('ssh', possible_values=False))
input_flags = InputFlags(InputFlag('ssh'))
self.assertEqual(Router()._structuring_input_flags(command, input_flags).valid_flags, ValidInputFlags(InputFlag('ssh')))
def test_validate_incorrect_func_args1(self):
def handler():
pass
with self.assertRaises(RequiredArgumentNotPassedException):
Router()._validate_func_args(handler)
def test_validate_incorrect_func_args2(self):
def handler(args, kwargs):
pass
with self.assertRaises(TooManyTransferredArgsException):
Router()._validate_func_args(handler)
def test_get_router_aliases(self):
router = Router()
@router.command(Command('some', aliases=['test', 'case']))
def handler(response):
pass
self.assertListEqual(router.get_aliases(), ['test', 'case'])
def test_get_router_aliases2(self):
router = Router()
@router.command(Command('some', aliases=['test', 'case']))
def handler(response):
pass
@router.command(Command('ext', aliases=['more', 'foo']))
def handler2(response):
pass
self.assertListEqual(router.get_aliases(), ['test', 'case', 'more', 'foo'])
def test_get_router_aliases3(self):
router = Router()
@router.command(Command('some'))
def handler(response):
pass
self.assertListEqual(router.get_aliases(), [])