Merge pull request #7 from koloideal/improve-perf

Improve perf
This commit is contained in:
kolo
2026-01-15 17:00:38 +03:00
committed by GitHub
43 changed files with 1204 additions and 543 deletions
+1 -1
View File
@@ -9,7 +9,7 @@ Argenta is the **"Simplest"**, **"Most Modular"**, and **"Most Elegant"** way to
---
![preview](https://i.ibb.co/fzWcfgFq/2025-12-04-173045.png)
![preview](https://vhs.charm.sh/vhs-2hvLCEgclmwZPJZt1vLGKi.gif)
**Argenta** allows you to build interactive CLI applications incredibly easily. There's no need to manually parse complex command structures or manage state transitions — just use routers and commands!
+1 -1
View File
@@ -9,7 +9,7 @@ Argenta — это **"Самый простой"**, **"Самый модульн
---
![preview](https://i.ibb.co/fzWcfgFq/2025-12-04-173045.png)
![preview](https://vhs.charm.sh/vhs-2hvLCEgclmwZPJZt1vLGKi.gif)
**Argenta** позволяет создавать интерактивные CLI-приложения невероятно легко. Не нужно вручную парсить сложные структуры команд или управлять переходами состояний — просто используйте роутеры и команды!
@@ -7,7 +7,7 @@ msgid ""
msgstr ""
"Project-Id-Version: Argenta \n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2025-12-04 20:39+0300\n"
"POT-Creation-Date: 2026-01-13 21:50+0300\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language: en\n"
@@ -38,23 +38,23 @@ msgstr ""
msgid "Инициализация"
msgstr "Initialization"
#: ../../root/api/app/index.rst:38
#: ../../root/api/app/index.rst:37
msgid "Создаёт и настраивает экземпляр приложения."
msgstr "Creates and configures an application instance."
#: ../../root/api/app/index.rst:40
#: ../../root/api/app/index.rst:39
msgid "``prompt``: Приглашение к вводу, отображаемое перед каждой командой."
msgstr "``prompt``: Input prompt displayed before each command."
#: ../../root/api/app/index.rst:41
#: ../../root/api/app/index.rst:40
msgid "``initial_message``: Сообщение, выводимое при запуске приложения."
msgstr "``initial_message``: Message displayed when the application starts."
#: ../../root/api/app/index.rst:42
#: ../../root/api/app/index.rst:41
msgid "``farewell_message``: Сообщение, выводимое при выходе из приложения."
msgstr "``farewell_message``: Message displayed when exiting the application."
#: ../../root/api/app/index.rst:43
#: ../../root/api/app/index.rst:42
msgid ""
"``exit_command``: Команда, которая маркируется как триггер для выхода из "
"приложения."
@@ -62,7 +62,7 @@ msgstr ""
"``exit_command``: Command that is marked as a trigger for exiting the "
"application."
#: ../../root/api/app/index.rst:44
#: ../../root/api/app/index.rst:43
msgid ""
"``system_router_title``: Заголовок для системного роутера (содержит "
"команду выхода)."
@@ -70,15 +70,7 @@ msgstr ""
"``system_router_title``: Title for the system router (contains the exit "
"command)."
#: ../../root/api/app/index.rst:45
msgid ""
"``ignore_command_register``: Если ``True``, регистр вводимых команд "
"игнорируется при поиске обработчика."
msgstr ""
"``ignore_command_register``: If ``True``, command case is ignored when "
"searching for a handler."
#: ../../root/api/app/index.rst:46
#: ../../root/api/app/index.rst:44
msgid ""
"``dividing_line``: Тип разделительной линии (``StaticDividingLine`` или "
"``DynamicDividingLine``)."
@@ -86,7 +78,7 @@ msgstr ""
"``dividing_line``: Type of dividing line (``StaticDividingLine`` or "
"``DynamicDividingLine``)."
#: ../../root/api/app/index.rst:47
#: ../../root/api/app/index.rst:45
msgid ""
"``repeat_command_groups_printing``: Если ``True``, список доступных "
"команд выводится перед каждым вводом."
@@ -94,7 +86,7 @@ msgstr ""
"``repeat_command_groups_printing``: If ``True``, the list of available "
"commands is displayed before each input."
#: ../../root/api/app/index.rst:48
#: ../../root/api/app/index.rst:46
msgid ""
"``override_system_messages``: Если ``True``, стандартное форматирование "
"(цвета, ASCII-арт) отключается."
@@ -102,7 +94,7 @@ msgstr ""
"``override_system_messages``: If ``True``, standard formatting (colors, "
"ASCII art) is disabled."
#: ../../root/api/app/index.rst:49
#: ../../root/api/app/index.rst:47
msgid ""
"``autocompleter``: Экземпляр класса :ref:`AutoCompleter "
"<root_api_app_autocompleter>`, отвечающий за автодополнение команд."
@@ -111,7 +103,7 @@ msgstr ""
"<root_api_app_autocompleter>` class responsible for command "
"autocompletion."
#: ../../root/api/app/index.rst:50
#: ../../root/api/app/index.rst:48
msgid ""
"``print_func``: Функция для вывода всех системных сообщений (по умолчанию"
" ``rich.Console().print``)."
@@ -119,11 +111,21 @@ msgstr ""
"``print_func``: Function for outputting all system messages (defaults to "
"``rich.Console().print``)."
#: ../../root/api/app/index.rst:55
#: ../../root/api/app/index.rst:53
msgid ""
"В приложениях на Argenta регистр вводимых команд не важен, проверка на "
"существование и роутинг команд производится на основании триггеров, "
"приведённых к нижнему регистру."
msgstr ""
"In applications on Argenta, the case of the entered commands is not important, checking for the "
" existence and routing of commands is performed based on triggers "
"reduced to lowercase."
#: ../../root/api/app/index.rst:56
msgid "Основные методы"
msgstr "Main Methods"
#: ../../root/api/app/index.rst:59
#: ../../root/api/app/index.rst:60
msgid ""
"Регистрирует роутер в приложении. Все команды из этого роутера становятся"
" доступными для вызова."
@@ -135,19 +137,19 @@ msgstr ""
msgid "Parameters"
msgstr "Parameters"
#: ../../root/api/app/index.rst:61
#: ../../root/api/app/index.rst:62
msgid "Экземпляр ``Router`` для регистрации."
msgstr "``Router`` instance to register."
#: ../../root/api/app/index.rst:65
#: ../../root/api/app/index.rst:66
msgid "Регистрирует несколько роутеров одновременно."
msgstr "Registers multiple routers simultaneously."
#: ../../root/api/app/index.rst:67
#: ../../root/api/app/index.rst:68
msgid "Последовательность экземпляров ``Router`` для регистрации."
msgstr "Sequence of ``Router`` instances to register."
#: ../../root/api/app/index.rst:71
#: ../../root/api/app/index.rst:72
msgid ""
"Добавляет текстовое сообщение, которое выводится при запуске приложения "
"после ``initial_message``."
@@ -155,11 +157,11 @@ msgstr ""
"Adds a text message that is displayed when the application starts after "
"``initial_message``."
#: ../../root/api/app/index.rst:73
#: ../../root/api/app/index.rst:74
msgid "Строка с сообщением."
msgstr "String with the message."
#: ../../root/api/app/index.rst:76
#: ../../root/api/app/index.rst:77
msgid ""
"Для вывода стандартных сообщений можно использовать готовые шаблоны из "
":ref:`PredefinedMessages <root_api_predefined_messages>`."
@@ -167,11 +169,11 @@ msgstr ""
"For outputting standard messages, you can use ready-made templates from "
":ref:`PredefinedMessages <root_api_predefined_messages>`."
#: ../../root/api/app/index.rst:81
#: ../../root/api/app/index.rst:82
msgid "Методы установки обработчиков"
msgstr "Handler Setup Methods"
#: ../../root/api/app/index.rst:83
#: ../../root/api/app/index.rst:84
msgid ""
"``App`` позволяет настраивать реакцию на различные события, такие как "
"ошибки ввода или неизвестные команды."
@@ -179,7 +181,7 @@ msgstr ""
"``App`` allows you to configure responses to various events, such as "
"input errors or unknown commands."
#: ../../root/api/app/index.rst:86
#: ../../root/api/app/index.rst:87
msgid ""
"Подробнее об исключениях и их обработке в соответствующем :ref:`разделе "
"документации <root_error_handling>`."
@@ -187,59 +189,59 @@ msgstr ""
"For more details on exceptions and their handling, see the corresponding "
":ref:`documentation section <root_error_handling>`."
#: ../../root/api/app/index.rst:92
#: ../../root/api/app/index.rst:93
msgid "Устанавливает шаблон для форматирования описания команды."
msgstr "Sets the template for formatting command descriptions."
#: ../../root/api/app/index.rst:94
#: ../../root/api/app/index.rst:95
msgid "Обработчик принимает триггер команды (``str``) и её описание (``str``)."
msgstr ""
"The handler accepts the command trigger (``str``) and its description "
"(``str``)."
#: ../../root/api/app/index.rst:100
#: ../../root/api/app/index.rst:101
msgid "Устанавливает обработчик при некорректном введённом синтаксисе флагов."
msgstr "Sets the handler for incorrect flag syntax input."
#: ../../root/api/app/index.rst:102 ../../root/api/app/index.rst:110
#: ../../root/api/app/index.rst:103 ../../root/api/app/index.rst:111
msgid "Обработчик принимает строку, введённую пользователем."
msgstr "The handler accepts the string entered by the user."
#: ../../root/api/app/index.rst:108
#: ../../root/api/app/index.rst:109
msgid "Устанавливает обработчик при повторяющихся флагах в введённой команде."
msgstr "Sets the handler for duplicate flags in the entered command."
#: ../../root/api/app/index.rst:116
#: ../../root/api/app/index.rst:117
msgid "Устанавливает обработчик при вводе неизвестной команды."
msgstr "Sets the handler for entering an unknown command."
#: ../../root/api/app/index.rst:118
#: ../../root/api/app/index.rst:119
msgid "Обработчик принимает объект ``InputCommand`` - объект введённой команды."
msgstr ""
"The handler accepts an ``InputCommand`` object - the entered command "
"object."
#: ../../root/api/app/index.rst:124
#: ../../root/api/app/index.rst:125
msgid "Устанавливает обработчик при вводе пустой строки."
msgstr "Sets the handler for entering an empty string."
#: ../../root/api/app/index.rst:126
#: ../../root/api/app/index.rst:127
msgid "Обработчик не принимает аргументов."
msgstr "The handler accepts no arguments."
#: ../../root/api/app/index.rst:132
#: ../../root/api/app/index.rst:133
msgid "Переопределяет стандартное поведение при вызове команды выхода."
msgstr "Overrides the default behavior when the exit command is invoked."
#: ../../root/api/app/index.rst:134
#: ../../root/api/app/index.rst:135
msgid "Обработчик принимает объект ``Response``."
msgstr "The handler accepts a ``Response`` object."
#: ../../root/api/app/index.rst:147
#: ../../root/api/app/index.rst:148
msgid "PredefinedMessages"
msgstr "PredefinedMessages"
#: ../../root/api/app/index.rst:149
#: ../../root/api/app/index.rst:150
msgid ""
"``PredefinedMessages`` — это контейнер, содержащий набор готовых к "
"использованию сообщений. Они отформатированы с использованием синтаксиса "
@@ -250,31 +252,40 @@ msgstr ""
"messages. They are formatted using ``rich`` syntax and are intended for "
"displaying standard information, such as usage hints."
#: ../../root/api/app/index.rst:151
#: ../../root/api/app/index.rst:152
msgid "Рекомендуется использовать их при старте приложения."
msgstr "It is recommended to use them when starting the application."
#: ../../root/api/app/index.rst:178
#: ../../root/api/app/index.rst:179
msgid "Строка: ``[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]``"
msgstr "String: ``[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]``"
#: ../../root/api/app/index.rst:180
#: ../../root/api/app/index.rst:181
msgid "Отображается как: ``Usage: <command> <flags>``"
msgstr "Displayed as: ``Usage: <command> <flags>``"
#: ../../root/api/app/index.rst:184
#: ../../root/api/app/index.rst:185
msgid "Строка: ``[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]``"
msgstr "String: ``[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]``"
#: ../../root/api/app/index.rst:186
#: ../../root/api/app/index.rst:187
msgid "Отображается как: ``Help: <command> --help``"
msgstr "Displayed as: ``Help: <command> --help``"
#: ../../root/api/app/index.rst:190
#: ../../root/api/app/index.rst:191
msgid "Строка: ``[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>``"
msgstr "String: ``[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>``"
#: ../../root/api/app/index.rst:192
#: ../../root/api/app/index.rst:193
msgid "Отображается как: ``Autocomplete: <part> <tab>``"
msgstr "Displayed as: ``Autocomplete: <part> <tab>``"
#~ msgid ""
#~ "``ignore_command_register``: Если ``True``, регистр"
#~ " вводимых команд игнорируется при поиске"
#~ " обработчика."
#~ msgstr ""
#~ "``ignore_command_register``: If ``True``, command"
#~ " case is ignored when searching for"
#~ " a handler."
@@ -7,7 +7,7 @@ msgid ""
msgstr ""
"Project-Id-Version: Argenta \n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2025-12-04 20:39+0300\n"
"POT-Creation-Date: 2025-12-08 19:48+0300\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language: en\n"
@@ -35,10 +35,10 @@ msgstr ""
#: ../../root/api/command/index.rst:8
msgid ""
"``Command`` инкапсулирует всю информацию о команде: её триггер (ключевое "
"слово для вызова), описание, набор флагов и список псевдонимов."
"слово для вызова), описание, набор флагов и множество псевдонимов."
msgstr ""
"``Command`` encapsulates all information about a command: its trigger "
"(keyword for invocation), description, set of flags, and list of aliases."
"(keyword for invocation), description, set of flags, and set of aliases."
#: ../../root/api/command/index.rst:13
msgid "Инициализация"
@@ -73,8 +73,8 @@ msgstr ""
"``Flag`` object or a ``Flags`` collection."
#: ../../root/api/command/index.rst:28
msgid "``aliases``: Список строковых псевдонимов для основного триггера."
msgstr "``aliases``: List of string aliases for the main trigger."
msgid "``aliases``: Множество строковых псевдонимов для основного триггера."
msgstr "``aliases``: Set of string aliases for the main trigger."
#: ../../root/api/command/index.rst:30 ../../root/api/command/index.rst:108
msgid "**Атрибуты:**"
@@ -107,8 +107,8 @@ msgstr ""
"during initialization."
#: ../../root/api/command/index.rst:46
msgid "Список строковых псевдонимов. Пуст, если псевдонимы не заданы."
msgstr "List of string aliases. Empty if no aliases are defined."
msgid "Множество строковых псевдонимов. Пуст, если псевдонимы не заданы."
msgstr "Set of string aliases. Empty if no aliases are defined."
#: ../../root/api/command/index.rst:48
msgid "**Пример использования:**"
@@ -119,8 +119,8 @@ msgid ""
"Подробнее про флаги: :ref:`Flags <root_api_command_flags>` и :ref:`Флаги "
"команд <root_flags>`."
msgstr ""
"More about flags: :ref:`Flags <root_api_command_flags>` and :ref:`Command "
"flags <root_flags>`."
"More about flags: :ref:`Flags <root_api_command_flags>` and :ref:`Command"
" flags <root_flags>`."
#: ../../root/api/command/index.rst:59
msgid "Регистрация команд"
+47 -24
View File
@@ -7,7 +7,7 @@ msgid ""
msgstr ""
"Project-Id-Version: Argenta \n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2025-12-02 22:27+0300\n"
"POT-Creation-Date: 2025-12-08 19:48+0300\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language: en\n"
@@ -30,8 +30,9 @@ msgid ""
"набора функций."
msgstr ""
"``Router`` is the main building block for organizing logic in an "
"application. Its purpose is to group related commands and their handlers. "
"Each router represents a logical container for a specific set of functions."
"application. Its purpose is to group related commands and their handlers."
" Each router represents a logical container for a specific set of "
"functions."
#: ../../root/api/router.rst:8
msgid ""
@@ -56,8 +57,8 @@ msgid ""
"``title``: Необязательный заголовок для группы команд. Отображается в "
"списке доступных команд, помогая пользователю ориентироваться."
msgstr ""
"``title``: Optional title for the command group. Displayed in the list of "
"available commands to help users navigate."
"``title``: Optional title for the command group. Displayed in the list of"
" available commands to help users navigate."
#: ../../root/api/router.rst:24
msgid ""
@@ -67,11 +68,11 @@ msgid ""
"используется статическая разделительная линия. Подробнее см. в разделе "
":ref:`Переопределение стандартного вывода <root_redirect_stdout>`."
msgstr ""
"``disable_redirect_stdout``: If ``True``, disables ``stdout`` capture for "
"all commands in this router. This is necessary for interactive commands "
"(e.g., with ``input()``). When capture is disabled, a static separator line "
"is automatically used. See :ref:`Overriding standard output <root_redirect_stdout>` "
"for more details."
"``disable_redirect_stdout``: If ``True``, disables ``stdout`` capture for"
" all commands in this router. This is necessary for interactive commands "
"(e.g., with ``input()``). When capture is disabled, a static separator "
"line is automatically used. See :ref:`Overriding standard output "
"<root_redirect_stdout>` for more details."
#: ../../root/api/router.rst:29
msgid "Регистрация команд"
@@ -82,7 +83,8 @@ msgid ""
"Для регистрации команды и привязки к ней обработчика используется "
"декоратор ``@command``."
msgstr ""
"The ``@command`` decorator is used to register a command and bind a handler to it."
"The ``@command`` decorator is used to register a command and bind a "
"handler to it."
#: ../../root/api/router.rst:35
msgid "Декоратор для регистрации функции как обработчика команды."
@@ -98,9 +100,9 @@ msgid ""
"Может быть строкой, которая станет триггером (без возможности настройки "
"флагов и описания)."
msgstr ""
"A ``Command`` instance describing the trigger, flags, and command description. "
"Can be a string that will become the trigger (without the ability to configure "
"flags and description)."
"A ``Command`` instance describing the trigger, flags, and command "
"description. Can be a string that will become the trigger (without the "
"ability to configure flags and description)."
#: ../../root/api/router.rst:39
msgid "**Пример использования:**"
@@ -130,12 +132,13 @@ msgstr ""
#: ../../root/api/router.rst:57
msgid ""
"Вы можете добавлять свои команды в этот роутер. Для этого импортируйте "
"``argenta.router.defaults.system_router`` и используйте его декоратор "
"``@command``."
"Вы можете добавлять свои команды в этот роутер. Для этого используйте "
"атрибут ``.system_router`` у созданного экхемпляра ``Orchestrator`` и "
"используйте его декоратор ``@command``."
msgstr ""
"You can add your own commands to this router. To do this, import "
"``argenta.router.defaults.system_router`` and use its ``@command`` decorator."
"You can add your own commands to this router. To do this, use the "
"``.system_router`` attribute of the created ``Orchestrator`` instance and"
" use its ``@command`` decorator."
#: ../../root/api/router.rst:62
msgid "Возможные исключения"
@@ -146,15 +149,16 @@ msgid ""
"При регистрации команд и флагов в ``Router`` могут возникнуть следующие "
"исключения:"
msgstr ""
"The following exceptions may occur when registering commands and flags in ``Router``:"
"The following exceptions may occur when registering commands and flags in"
" ``Router``:"
#: ../../root/api/router.rst:68
msgid ""
"Выбрасывается, если триггер команды в ``Command`` содержит пробелы. "
"Триггеры должны быть одним словом."
msgstr ""
"Raised if the command trigger in ``Command`` contains spaces. "
"Triggers must be a single word."
"Raised if the command trigger in ``Command`` contains spaces. Triggers "
"must be a single word."
#: ../../root/api/router.rst:70
msgid "**Неправильно:** ``Command(\"add user\")``"
@@ -173,7 +177,8 @@ msgstr ""
"Raised if duplicate names were used when defining flags for a command. "
"Flag names within a single command must be unique."
#: ../../root/api/router.rst:78
#: ../../root/api/router.rst:78 ../../root/api/router.rst:96
#: ../../root/api/router.rst:115
msgid "**Пример, вызывающий исключение:**"
msgstr "**Example that raises an exception:**"
@@ -182,5 +187,23 @@ msgid ""
"Возникает, если обработчик команды не принимает обязательный аргумент "
"``Response``."
msgstr ""
"Raised if the command handler does not accept the required ``Response`` argument."
"Raised if the command handler does not accept the required ``Response`` "
"argument."
#: ../../root/api/router.rst:94
msgid ""
"Возникает, если при регистрации команд в роутере были использованы "
"дублирующиеся триггеры. Каждая команда должна иметь уникальный триггер в "
"рамках приложения."
msgstr ""
"Raised if duplicate triggers were used when registering commands in the "
"router. Each command must have a unique trigger within a single router."
#: ../../root/api/router.rst:113
msgid ""
"Возникает, если при регистрации команд были использованы дублирующиеся "
"алиасы. Алиасы должны быть уникальны в рамках всего приложения."
msgstr ""
"Raised if duplicate aliases were used when registering commands. Aliases "
"must be unique within the entire router."
@@ -7,7 +7,7 @@ msgid ""
msgstr ""
"Project-Id-Version: Argenta \n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2025-12-04 20:39+0300\n"
"POT-Creation-Date: 2025-12-08 19:48+0300\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language: en\n"
@@ -46,17 +46,22 @@ msgstr ""
"``Router``) if your commands:"
#: ../../root/redirect_stdout.rst:15
msgid ""
"✓ Используют ``input()`` для интерактивного ввода данных от пользователя "
"✓ Используют прогресс-бары (``tqdm``, ``rich.progress``) ✓ Выводят данные"
" в реальном времени (streaming, логи) ✓ Используют библиотеки, которые "
"напрямую работают с ``stdout``"
msgstr ""
"✓ Use ``input()`` for interactive user input ✓ Use progress bars "
"(``tqdm``, ``rich.progress``) ✓ Output data in real-time (streaming, "
"logs) ✓ Use libraries that work directly with ``stdout``"
msgid "✓ Используют ``input()`` для интерактивного ввода данных от пользователя"
msgstr "✓ Use ``input()`` for interactive user input"
#: ../../root/redirect_stdout.rst:20
#: ../../root/redirect_stdout.rst:17
msgid "✓ Используют прогресс-бары (``tqdm``, ``rich.progress``)"
msgstr "✓ Use progress bars (``tqdm``, ``rich.progress``)"
#: ../../root/redirect_stdout.rst:19
msgid "✓ Выводят данные в реальном времени (streaming, логи)"
msgstr "✓ Output data in real-time (streaming, logs)"
#: ../../root/redirect_stdout.rst:21
msgid "✓ Используют библиотеки, которые напрямую работают с ``stdout``"
msgstr "✓ Use libraries that work directly with ``stdout``"
#: ../../root/redirect_stdout.rst:23
msgid ""
"Для обычных команд с ``print()`` перехват можно оставить включённым — это"
" не влияет на их работу."
@@ -64,11 +69,11 @@ msgstr ""
"For regular commands with ``print()``, interception can be left enabled —"
" it does not affect their operation."
#: ../../root/redirect_stdout.rst:25
#: ../../root/redirect_stdout.rst:28
msgid "Механизм перехвата ``stdout``"
msgstr "``stdout`` Interception Mechanism"
#: ../../root/redirect_stdout.rst:27
#: ../../root/redirect_stdout.rst:30
msgid ""
"По умолчанию ``Argenta`` перехватывает весь текст, выводимый в ``stdout``"
" внутри обработчика команды. Это необходимо для реализации **динамических"
@@ -83,15 +88,15 @@ msgstr ""
"draw the top and bottom borders. This approach creates a neat interface "
"where the command output is \"wrapped\" in a frame fitted to its content."
#: ../../root/redirect_stdout.rst:29
#: ../../root/redirect_stdout.rst:32
msgid "Пример приложения с динамической разделительной линией:"
msgstr "Example of an application with a dynamic dividing line:"
#: ../../root/redirect_stdout.rst:31
#: ../../root/redirect_stdout.rst:34
msgid "Example of an application with a dynamic dividing line"
msgstr "Example of an application with a dynamic dividing line"
#: ../../root/redirect_stdout.rst:34
#: ../../root/redirect_stdout.rst:37
msgid ""
"Как вы можете заметить, разделительная линия ровно той же длины, что и "
"самая длинная строка в выводе."
@@ -99,15 +104,15 @@ msgstr ""
"As you can see, the dividing line is exactly the same length as the "
"longest line in the output."
#: ../../root/redirect_stdout.rst:36
#: ../../root/redirect_stdout.rst:39
msgid "То же приложение с статической линией:"
msgstr "The same application with a static line:"
#: ../../root/redirect_stdout.rst:38
#: ../../root/redirect_stdout.rst:41
msgid "Example of an application with a static dividing line"
msgstr "Example of an application with a static dividing line"
#: ../../root/redirect_stdout.rst:41
#: ../../root/redirect_stdout.rst:44
msgid ""
"В этом примере разделительная линия имеет фиксированную длину (по "
"умолчанию 25 символов)."
@@ -115,11 +120,11 @@ msgstr ""
"In this example, the dividing line has a fixed length (25 characters by "
"default)."
#: ../../root/redirect_stdout.rst:46
#: ../../root/redirect_stdout.rst:49
msgid "Побочные эффекты перехвата ``stdout``"
msgstr "Side Effects of ``stdout`` Interception"
#: ../../root/redirect_stdout.rst:48
#: ../../root/redirect_stdout.rst:51
msgid ""
"Побочный эффект этого механизма проявляется при использовании функций, "
"которые последовательно выводят текст в консоль и ожидают ввод от "
@@ -129,7 +134,7 @@ msgstr ""
"sequentially output text to the console and expect user input. A classic "
"example is the standard ``input()`` function."
#: ../../root/redirect_stdout.rst:57
#: ../../root/redirect_stdout.rst:60
msgid ""
"При включённом перехвате ``stdout`` текст (например, ``\"Введите ваше "
"имя: \"``) **не будет выведен в консоль немедленно**. Он попадёт в буфер "
@@ -141,11 +146,11 @@ msgstr ""
" into a buffer and appear only after the handler finishes, along with the"
" rest of the output. This can confuse the user."
#: ../../root/redirect_stdout.rst:62
#: ../../root/redirect_stdout.rst:65
msgid "Отключение перехвата ``stdout`` с помощью ``disable_redirect_stdout``"
msgstr "Disabling ``stdout`` Interception with ``disable_redirect_stdout``"
#: ../../root/redirect_stdout.rst:64
#: ../../root/redirect_stdout.rst:67
msgid ""
"Чтобы решить эту проблему, в конструкторе ``Router`` предусмотрен "
"специальный аргумент:"
@@ -153,11 +158,11 @@ msgstr ""
"To solve this problem, the ``Router`` constructor provides a special "
"argument:"
#: ../../root/redirect_stdout.rst:66
#: ../../root/redirect_stdout.rst:69
msgid "**disable_redirect_stdout** (``bool``, по умолчанию ``False``)"
msgstr "**disable_redirect_stdout** (``bool``, default ``False``)"
#: ../../root/redirect_stdout.rst:68
#: ../../root/redirect_stdout.rst:71
msgid ""
"Если при создании роутера установить ``disable_redirect_stdout=True``, "
"механизм перехвата ``stdout`` будет отключён для всех его обработчиков."
@@ -165,11 +170,11 @@ msgstr ""
"If you set ``disable_redirect_stdout=True`` when creating a router, the "
"``stdout`` interception mechanism will be disabled for all its handlers."
#: ../../root/redirect_stdout.rst:70 ../../root/redirect_stdout.rst:100
#: ../../root/redirect_stdout.rst:73 ../../root/redirect_stdout.rst:103
msgid "**Пример использования:**"
msgstr "**Usage example:**"
#: ../../root/redirect_stdout.rst:76
#: ../../root/redirect_stdout.rst:79
msgid ""
"В этом случае ``input()`` будет работать как обычно, и пользователь сразу"
" увидит приглашение к вводу."
@@ -177,11 +182,11 @@ msgstr ""
"In this case, ``input()`` will work as usual, and the user will "
"immediately see the input prompt."
#: ../../root/redirect_stdout.rst:81
#: ../../root/redirect_stdout.rst:84
msgid "Типы разделительных линий"
msgstr "Types of Dividing Lines"
#: ../../root/redirect_stdout.rst:83
#: ../../root/redirect_stdout.rst:86
msgid ""
"``Argenta`` поддерживает два типа разделителей, которые настраиваются при"
" инициализации ``App``:"
@@ -189,11 +194,11 @@ msgstr ""
"``Argenta`` supports two types of dividers, which are configured during "
"``App`` initialization:"
#: ../../root/redirect_stdout.rst:85
#: ../../root/redirect_stdout.rst:88
msgid "**``DynamicDividingLine()``**"
msgstr "**``DynamicDividingLine()``**"
#: ../../root/redirect_stdout.rst:86
#: ../../root/redirect_stdout.rst:89
msgid ""
"Поведение по умолчанию. Длина линии динамически подстраивается под самый "
"длинный текст в выводе."
@@ -201,7 +206,7 @@ msgstr ""
"Default behavior. The line length dynamically adjusts to the longest text"
" in the output."
#: ../../root/redirect_stdout.rst:87
#: ../../root/redirect_stdout.rst:90
msgid ""
"Требует включённого перехвата ``stdout`` "
"(``disable_redirect_stdout=False`` в роутере)."
@@ -209,11 +214,11 @@ msgstr ""
"Requires enabled ``stdout`` interception "
"(``disable_redirect_stdout=False`` in the router)."
#: ../../root/redirect_stdout.rst:89
#: ../../root/redirect_stdout.rst:92
msgid "**``StaticDividingLine(length: int = 25)``**"
msgstr "**``StaticDividingLine(length: int = 25)``**"
#: ../../root/redirect_stdout.rst:90
#: ../../root/redirect_stdout.rst:93
msgid ""
"Линия имеет фиксированную длину (по умолчанию 25 символов), которую можно"
" задать через аргумент ``length``."
@@ -221,7 +226,7 @@ msgstr ""
"The line has a fixed length (25 characters by default), which can be set "
"via the ``length`` argument."
#: ../../root/redirect_stdout.rst:91
#: ../../root/redirect_stdout.rst:94
msgid ""
"Используется принудительно для роутеров с "
"``disable_redirect_stdout=True``, так как без перехвата вывода невозможно"
@@ -230,11 +235,11 @@ msgstr ""
"Used forcibly for routers with ``disable_redirect_stdout=True``, as it is"
" impossible to determine dynamic length without output interception."
#: ../../root/redirect_stdout.rst:96
#: ../../root/redirect_stdout.rst:99
msgid "Настройка разделительной линии в ``App``"
msgstr "Configuring the Dividing Line in ``App``"
#: ../../root/redirect_stdout.rst:98
#: ../../root/redirect_stdout.rst:101
msgid ""
"Вы можете глобально задать тип разделителя для всего приложения через "
"аргумент ``dividing_line`` в конструкторе ``App``."
@@ -242,63 +247,63 @@ msgstr ""
"You can globally set the divider type for the entire application via the "
"``dividing_line`` argument in the ``App`` constructor."
#: ../../root/redirect_stdout.rst:109
#: ../../root/redirect_stdout.rst:112
msgid "Итоговое поведение"
msgstr "Resulting Behavior"
#: ../../root/redirect_stdout.rst:115
#: ../../root/redirect_stdout.rst:118
msgid "``disable_redirect_stdout`` на ``Router``"
msgstr "``disable_redirect_stdout`` on ``Router``"
#: ../../root/redirect_stdout.rst:116
#: ../../root/redirect_stdout.rst:119
msgid "Тип линии в ``App``"
msgstr "Line type in ``App``"
#: ../../root/redirect_stdout.rst:117
#: ../../root/redirect_stdout.rst:120
msgid "Фактическое поведение"
msgstr "Actual behavior"
#: ../../root/redirect_stdout.rst:118
#: ../../root/redirect_stdout.rst:121
msgid "``input()`` работает корректно?"
msgstr "Does ``input()`` work correctly?"
#: ../../root/redirect_stdout.rst:119 ../../root/redirect_stdout.rst:123
#: ../../root/redirect_stdout.rst:122 ../../root/redirect_stdout.rst:126
msgid "``False`` (по умолчанию)"
msgstr "``False`` (default)"
#: ../../root/redirect_stdout.rst:120 ../../root/redirect_stdout.rst:128
#: ../../root/redirect_stdout.rst:123 ../../root/redirect_stdout.rst:131
msgid "``DynamicDividingLine``"
msgstr "``DynamicDividingLine``"
#: ../../root/redirect_stdout.rst:121
#: ../../root/redirect_stdout.rst:124
msgid "Динамическая линия, длина по содержимому"
msgstr "Dynamic line, length by content"
#: ../../root/redirect_stdout.rst:122 ../../root/redirect_stdout.rst:126
#: ../../root/redirect_stdout.rst:125 ../../root/redirect_stdout.rst:129
msgid "Нет"
msgstr "No"
#: ../../root/redirect_stdout.rst:124 ../../root/redirect_stdout.rst:132
#: ../../root/redirect_stdout.rst:127 ../../root/redirect_stdout.rst:135
msgid "``StaticDividingLine``"
msgstr "``StaticDividingLine``"
#: ../../root/redirect_stdout.rst:125 ../../root/redirect_stdout.rst:133
#: ../../root/redirect_stdout.rst:128 ../../root/redirect_stdout.rst:136
msgid "Статическая линия указанной длины"
msgstr "Static line of specified length"
#: ../../root/redirect_stdout.rst:127 ../../root/redirect_stdout.rst:131
#: ../../root/redirect_stdout.rst:130 ../../root/redirect_stdout.rst:134
msgid "``True``"
msgstr "``True``"
#: ../../root/redirect_stdout.rst:129
#: ../../root/redirect_stdout.rst:132
msgid "**Принудительно статическая линия** (длина по умолч.)"
msgstr "**Forcibly static line** (default length)"
#: ../../root/redirect_stdout.rst:130 ../../root/redirect_stdout.rst:134
#: ../../root/redirect_stdout.rst:133 ../../root/redirect_stdout.rst:137
msgid "Да"
msgstr "Yes"
#: ../../root/redirect_stdout.rst:136
#: ../../root/redirect_stdout.rst:139
msgid ""
"Таким образом, для интерактивных команд, требующих ввода от пользователя,"
" отключайте перехват ``stdout`` на уровне роутера. Для всех остальных "
+3 -2
View File
@@ -28,7 +28,6 @@ App
farewell_message: str = "\nSee you\n",
exit_command: Command = DEFAULT_EXIT_COMMAND,
system_router_title: str | None = "System points:",
ignore_command_register: bool = True,
dividing_line: AVAILABLE_DIVIDING_LINES = DEFAULT_DIVIDING_LINE,
repeat_command_groups_printing: bool = False,
override_system_messages: bool = False,
@@ -42,7 +41,6 @@ App
* ``farewell_message``: Сообщение, выводимое при выходе из приложения.
* ``exit_command``: Команда, которая маркируется как триггер для выхода из приложения.
* ``system_router_title``: Заголовок для системного роутера (содержит команду выхода).
* ``ignore_command_register``: Если ``True``, регистр вводимых команд игнорируется при поиске обработчика.
* ``dividing_line``: Тип разделительной линии (``StaticDividingLine`` или ``DynamicDividingLine``).
* ``repeat_command_groups_printing``: Если ``True``, список доступных команд выводится перед каждым вводом.
* ``override_system_messages``: Если ``True``, стандартное форматирование (цвета, ASCII-арт) отключается.
@@ -51,6 +49,9 @@ App
-----
.. note::
В приложениях на Argenta регистр вводимых команд не важен, проверка на существование и роутинг команд производится на основании триггеров, приведённых к нижнему регистру.
Основные методы
---------------
+4 -4
View File
@@ -5,7 +5,7 @@ Command
``Command`` — это основная единица функциональности в приложении. Каждая команда связывает хэндлер с триггером, введя который он будет вызван для обработки.
``Command`` инкапсулирует всю информацию о команде: её триггер (ключевое слово для вызова), описание, набор флагов и список псевдонимов.
``Command`` инкапсулирует всю информацию о команде: её триггер (ключевое слово для вызова), описание, набор флагов и множество псевдонимов.
-----
@@ -18,14 +18,14 @@ Command
__init__(self, trigger: str, *,
description: str | None = None,
flags: Flag | Flags = DEFAULT_WITHOUT_FLAGS,
aliases: list[str] | list[Never] = DEFAULT_WITHOUT_ALIASES) -> None
aliases: set[str] = DEFAULT_WITHOUT_ALIASES) -> None
Создаёт новую команду для регистрации в роутере.
* ``trigger``: Строковый триггер, который пользователь вводит для вызова команды. Является основным идентификатором.
* ``description``: Необязательное описание, объясняющее назначение команды. Отображается в справке.
* ``flags``: Набор флагов для настройки поведения. Может быть одиночным объектом ``Flag`` или коллекцией ``Flags``.
* ``aliases``: Список строковых псевдонимов для основного триггера.
* ``aliases``: Множество строковых псевдонимов для основного триггера.
**Атрибуты:**
@@ -43,7 +43,7 @@ Command
.. py:attribute:: aliases
Список строковых псевдонимов. Пуст, если псевдонимы не заданы.
Множество строковых псевдонимов. Пуст, если псевдонимы не заданы.
**Пример использования:**
+39 -1
View File
@@ -54,7 +54,7 @@ Router
Предопределённый экземпляр ``Router`` с базовыми системными командами (по умолчанию — команда выхода). Имеет заголовок **«System points:»**, который можно переопределить в ``App``.
Вы можете добавлять свои команды в этот роутер. Для этого импортируйте ``argenta.router.defaults.system_router`` и используйте его декоратор ``@command``.
Вы можете добавлять свои команды в этот роутер. Для этого используйте атрибут ``.system_router`` у созданного экхемпляра ``Orchestrator`` и используйте его декоратор ``@command``.
-----
@@ -89,3 +89,41 @@ Router
Возникает, если обработчик команды не принимает обязательный аргумент ``Response``.
.. py:exception:: RepeatedTriggerNameException
Возникает, если при регистрации команд в роутере были использованы дублирующиеся триггеры. Каждая команда должна иметь уникальный триггер в рамках приложения.
**Пример, вызывающий исключение:**
.. code-block:: python
:linenos:
router = Router()
@router.command(Command("start"))
def start_handler(response: Response) -> None:
pass
@router.command(Command("start")) # Duplicate trigger!
def another_start_handler(response: Response) -> None:
pass
.. py:exception:: RepeatedAliasNameException
Возникает, если при регистрации команд были использованы дублирующиеся алиасы. Алиасы должны быть уникальны в рамках всего приложения.
**Пример, вызывающий исключение:**
.. code-block:: python
:linenos:
router = Router()
@router.command(Command("start", aliases={"s", "run"}))
def start_handler(response: Response) -> None:
pass
@router.command(Command("begin", aliases={"s"})) # Duplicate alias "s"!
def begin_handler(response: Response) -> None:
pass
+3
View File
@@ -13,8 +13,11 @@
Отключайте перехват ``stdout`` (``disable_redirect_stdout=True`` в ``Router``), если ваши команды:
✓ Используют ``input()`` для интерактивного ввода данных от пользователя
✓ Используют прогресс-бары (``tqdm``, ``rich.progress``)
✓ Выводят данные в реальном времени (streaming, логи)
✓ Используют библиотеки, которые напрямую работают с ``stdout``
Для обычных команд с ``print()`` перехват можно оставить включённым — это не влияет на их работу.
+1
View File
@@ -0,0 +1 @@
from .benchmarks import *
+49
View File
@@ -0,0 +1,49 @@
from concurrent.futures import ProcessPoolExecutor
import os
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.text import Text
from metrics.utils import run_benchmark, BenchmarkResult
from .registry import Benchmarks, Benchmark
def main():
console = Console()
all_benchmarks: list[Benchmark] = Benchmarks.get_benchmarks()
workers = os.cpu_count() or 1
with ProcessPoolExecutor(max_workers=workers) as executor:
results = executor.map(run_benchmark, all_benchmarks)
type_paired_benchmarks: dict[str, list[BenchmarkResult]] = {}
for result in results:
type_paired_benchmarks.setdefault(result.type_, []).append(result)
for type_, benchmarks in type_paired_benchmarks.items():
header_text = Text(f"TYPE: {type_.upper()}", style="bold magenta")
console.print(Panel(header_text, expand=False, border_style="magenta"))
table = Table(show_header=True, header_style="bold cyan", border_style="blue", show_lines=True)
table.add_column("Name", style="green")
table.add_column("Description", style="dim")
table.add_column("Iterations", justify="right")
table.add_column("Avg Time (ms)", justify="right", style="bold yellow")
for benchmark in benchmarks:
table.add_row(
benchmark.name,
benchmark.description,
str(benchmark.iterations),
str(benchmark.avg_time)
)
console.print(table)
console.print()
if __name__ == "__main__":
main()
+1
View File
@@ -0,0 +1 @@
from .pre_cycle_setup import *
+125
View File
@@ -0,0 +1,125 @@
__all__ = [
"benchmark_no_aliases",
"benchmark_many_aliases",
"benchmark_few_aliases",
"benchmark_extreme_aliases",
"benchmark_very_many_aliases"
]
from argenta import App
from argenta.router import Router
from argenta.command.models import Command
from argenta.response import Response
from ..utils import get_time_of_pre_cycle_setup
from ..registry import benchmark
@benchmark(type_="pre_cycle_setup", description="With no aliases")
def benchmark_no_aliases() -> float:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1'))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2'))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3'))
def handler3(_res: Response) -> None:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
@benchmark(type_="pre_cycle_setup", description="With few aliases (6 total)")
def benchmark_few_aliases() -> float:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1', aliases={'c1', 'cmd1'}))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2', aliases={'c2', 'cmd2'}))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3', aliases={'c3', 'cmd3'}))
def handler3(_res: Response) -> None:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
@benchmark(type_="pre_cycle_setup", description="With many aliases (15 total)")
def benchmark_many_aliases() -> float:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1', aliases={'c1', 'cmd1', 'com1', 'first', 'one'}))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2', aliases={'c2', 'cmd2', 'com2', 'second', 'two'}))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3', aliases={'c3', 'cmd3', 'com3', 'third', 'three'}))
def handler3(_res: Response) -> None:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
@benchmark(type_="pre_cycle_setup", description="With very many aliases (60 total)")
def benchmark_very_many_aliases() -> float:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1', aliases={f'alias1_{i}' for i in range(20)}))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2', aliases={f'alias2_{i}' for i in range(20)}))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3', aliases={f'alias3_{i}' for i in range(20)}))
def handler3(_res: Response) -> None:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
@benchmark(type_="pre_cycle_setup", description="With extreme aliases (300 total)")
def benchmark_extreme_aliases() -> float:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1', aliases={f'alias1_{i}' for i in range(100)}))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2', aliases={f'alias2_{i}' for i in range(100)}))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3', aliases={f'alias3_{i}' for i in range(100)}))
def handler3(_res: Response) -> None:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
+98
View File
@@ -0,0 +1,98 @@
__all__ = [
"Benchmark",
"Benchmarks",
"benchmark"
]
from typing import Callable, ClassVar, overload, override
BenchmarkAsFunc = Callable[[], float]
class Benchmark:
def __init__(
self,
func: BenchmarkAsFunc,
*,
type_: str,
name: str,
description: str,
iterations: int
) -> None:
self.func = func
self.type_ = type_
self.name = name
self.description = description
self.iterations = iterations
def run(self) -> float:
return self.func()
@override
def __repr__(self) -> str:
return f'Benchmark<{self.type_=}, {self.name=}, {self.description=}, {self.iterations=}>'
@override
def __str__(self) -> str:
return f'Benchmark({self.type_=}, {self.name=}, {self.description=}, {self.iterations=})'
class Benchmarks:
_benchmarks: ClassVar[list[Benchmark]] = []
@overload
@classmethod
def register(
cls,
call: BenchmarkAsFunc,
*,
type_: str = "",
description: str = "",
iterations: int = 100,
) -> BenchmarkAsFunc:
...
@overload
@classmethod
def register(
cls,
call: None = None,
*,
type_: str = "",
description: str = "",
iterations: int = 100,
) -> Callable[[BenchmarkAsFunc], BenchmarkAsFunc]:
...
@classmethod
def register(
cls,
call: BenchmarkAsFunc | None = None,
*,
type_: str = "",
description: str = "",
iterations: int = 100,
) -> Callable[[BenchmarkAsFunc], BenchmarkAsFunc] | BenchmarkAsFunc:
def decorator(func: BenchmarkAsFunc) -> BenchmarkAsFunc:
cls._benchmarks.append(
Benchmark(
func,
type_=type_,
name=func.__name__,
description=description or f'description for {func.__name__} with {iterations} iterations',
iterations=iterations
)
)
return func
if call is None:
return decorator
else:
return decorator(call)
@classmethod
def get_benchmarks(cls) -> list[Benchmark]:
return cls._benchmarks
benchmark = Benchmarks.register
+44
View File
@@ -0,0 +1,44 @@
__all__ = [
"get_time_of_pre_cycle_setup",
"attempts_to_average",
"run_benchmark",
"BenchmarkResult"
]
import io
from contextlib import redirect_stdout
import time
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from argenta import App
from metrics.registry import Benchmark
def get_time_of_pre_cycle_setup(app: App) -> float:
start = time.perf_counter()
with redirect_stdout(io.StringIO()):
app._pre_cycle_setup() # pyright: ignore[reportPrivateUsage]
end = time.perf_counter()
return (end - start) * 1000 # as milliseconds
def attempts_to_average(bench_attempts: list[float], iterations: int) -> Decimal:
return Decimal(sum(bench_attempts) / iterations).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
@dataclass(frozen=True)
class BenchmarkResult:
type_: str
name: str
description: str
iterations: int
avg_time: Decimal
def run_benchmark(benchmark: Benchmark) -> BenchmarkResult:
bench_attempts: list[float] = []
for _ in range(benchmark.iterations):
bench_attempts.append(benchmark.run())
avg = attempts_to_average(bench_attempts, benchmark.iterations)
return BenchmarkResult(benchmark.type_, benchmark.name, benchmark.description, benchmark.iterations, avg)
View File
-32
View File
@@ -1,32 +0,0 @@
from argenta.app import App
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
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): # pyright: ignore[reportUnusedFunction, reportUnusedParameter]
pass
app = App()
app.include_router(router)
return get_time_of_pre_cycle_setup(app)
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): # pyright: ignore[reportUnusedFunction, reportUnusedParameter]
pass
app = App()
app.include_router(router)
return get_time_of_pre_cycle_setup(app)
+56 -14
View File
@@ -1,19 +1,61 @@
from argenta import App, DataBridge, Response, Router
from argenta.di import FromDishka
from argenta.di.integration import setup_dishka, _auto_inject_handlers
from argenta.di.providers import SystemProvider
from dishka import make_container
import math
container = make_container()
Response.patch_by_container(container)
def estimate_nth_prime_upper_bound(n: int):
if n < 6:
return 15
app = App()
router = Router()
log_n = math.log(n)
log_log_n = math.log(log_n)
@router.command('command')
def handler(res: Response, data_bridge: FromDishka[DataBridge]):
print(data_bridge)
if n < 100:
return int(n * (log_n + log_log_n) * 1.5)
elif n < 1000:
return int(n * (log_n + log_log_n) * 1.3)
elif n >= 8009824:
return int(n * (log_n + log_log_n - 1 + 1.8 * log_log_n / log_n))
else:
return int(n * (log_n + log_log_n - 1 + 2.0 * log_log_n / log_n))
_auto_inject_handlers(app)
_auto_inject_handlers(app)
def odd_dig_primes(n: int) -> list[int]:
nums = {k: True for k in range(2, n+1)}
for num, is_checkable in nums.items():
if not is_checkable:
continue
if nums[2]:
nums[2] = False
for x in range(num * num, n, num):
nums[x] = False
primes = len([x for x in nums.items() if x[1]])
max_prime = max([x[0] for x in nums.items() if x[1]])
upper_bound = estimate_nth_prime_upper_bound(primes+1)
print(upper_bound)
nums2 = {k: True for k in range(2, upper_bound)}
for num, is_checkable in nums2.items():
if not is_checkable:
continue
if nums2[2]:
nums2[2] = False
for x in range(num * num, upper_bound, num):
nums2[x] = False
print([x for x in nums2.items() if x[1]])
next_prime_after_max = [x[0] for x in nums2.items() if x[1]][-1]
return [
primes,
max_prime,
next_prime_after_max
]
print(odd_dig_primes(13))
+2 -1
View File
@@ -1,9 +1,10 @@
# main.py
from argenta import App, Orchestrator
from argenta.app import DynamicDividingLine
from .routers import router
app: App = App()
app: App = App(prompt='>>> ', dividing_line=DynamicDividingLine('~'))
orchestrator: Orchestrator = Orchestrator()
def main() -> None:
+2 -5
View File
@@ -1,16 +1,13 @@
from argenta import App, Orchestrator
from argenta.app import PredefinedMessages
from argenta.orchestrator.argparser import ArgParser, BooleanArgument
from argenta.app.dividing_line.models import DynamicDividingLine
from mock.mock_app.routers import work_router
app: App = App(
dividing_line=DynamicDividingLine('^'),
)
argparser = ArgParser([BooleanArgument('some')])
orchestrator: Orchestrator = Orchestrator(argparser)
orchestrator: Orchestrator = Orchestrator()
print(argparser.parsed_argspace.get_by_type(BooleanArgument))
def main():
app.include_router(work_router)
@@ -22,5 +19,5 @@ def main():
orchestrator.start_polling(app)
if __name__ == "__main__":
orchestrator.start_polling(app)
main()
+8 -1
View File
@@ -4,7 +4,14 @@ from argenta.command import Flag, Flags
work_router: Router = Router(title="Base points:", disable_redirect_stdout=True)
@work_router.command(Command("hello", flags=Flags(Flag("test")), description="Hello, world!"))
@work_router.command(
Command(
"hello",
flags=Flags([
Flag("test")
]),
description="Hello, world!")
)
def command_help(response: Response):
c = input("Enter your name: ")
print(f"Hello, {c}!")
+5
View File
@@ -55,6 +55,11 @@ root = "tests/"
reportPrivateUsage = false
reportUnusedFunction = false
[[tool.pyright.executionEnvironments]]
root = "metrics/"
reportPrivateUsage = false
reportUnusedFunction = false
[tool.coverage.run]
branch = true
omit = [
+18 -13
View File
@@ -6,7 +6,9 @@ from typing import Never
class AutoCompleter:
def __init__(self, history_filename: str | None = None, autocomplete_button: str = "tab") -> None:
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
@@ -23,12 +25,18 @@ class AutoCompleter:
: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 _get_history_items() if cmd.startswith(text))
matches: list[str] = sorted(
cmd for cmd in _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]:
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:
@@ -40,7 +48,7 @@ class AutoCompleter:
else:
return None
def initial_setup(self, all_commands: list[str]) -> None:
def initial_setup(self, all_commands: set[str]) -> None:
"""
Private. Initial setup function
:param all_commands: Registered commands for adding them to the autocomplete history
@@ -61,7 +69,7 @@ class AutoCompleter:
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], ignore_command_register: bool) -> None:
def exit_setup(self, all_commands: set[str]) -> None:
"""
Private. Exit setup function
:return: None
@@ -72,21 +80,18 @@ class AutoCompleter:
raw_history = history_file.read()
pretty_history: list[str] = []
for line in set(raw_history.strip().split("\n")):
if _is_command_exist(line.split()[0], all_commands, ignore_command_register):
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))
def _is_command_exist(command: str, existing_commands: list[str], ignore_command_register: bool) -> bool:
if ignore_command_register:
return command.lower() in existing_commands
return command in existing_commands
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)]
return [
readline.get_history_item(i)
for i in range(1, readline.get_current_history_length() + 1)
]
+116 -92
View File
@@ -3,7 +3,7 @@ __all__ = ["App"]
import io
import re
from contextlib import redirect_stdout
from typing import Never, TypeAlias
from typing import Callable, Never, TypeAlias
from art import text2art
from rich.console import Console
@@ -23,13 +23,15 @@ from argenta.command.exceptions import (
RepeatedInputFlagsException,
UnprocessedInputFlagException,
)
from argenta.router.exceptions import RepeatedAliasNameException, RepeatedTriggerNameException
from argenta.command.models import Command, InputCommand
from argenta.response import Response
from argenta.router import Router
from argenta.router.defaults import system_router
Matches: TypeAlias = list[str] | list[Never]
_ANSI_ESCAPE_RE: re.Pattern[str] = re.compile(r"\u001b\[[0-9;]*m")
class BaseApp:
def __init__(
@@ -40,7 +42,6 @@ class BaseApp:
farewell_message: str,
exit_command: Command,
system_router_title: str,
ignore_command_register: bool,
dividing_line: StaticDividingLine | DynamicDividingLine,
repeat_command_groups_printing: bool,
override_system_messages: bool,
@@ -50,43 +51,37 @@ class BaseApp:
self._prompt: str = prompt
self._print_func: Printer = print_func
self._exit_command: Command = exit_command
self._system_router_title: str = system_router_title
self._dividing_line: StaticDividingLine | DynamicDividingLine = dividing_line
self._ignore_command_register: bool = ignore_command_register
self._repeat_command_groups_printing_description: bool = repeat_command_groups_printing
self._repeat_command_groups_printing: bool = repeat_command_groups_printing
self._override_system_messages: bool = override_system_messages
self._autocompleter: AutoCompleter = autocompleter
self.system_router: Router = Router(title=system_router_title)
self._farewell_message: str = farewell_message
self._initial_message: str = initial_message
self._stdout_buffer: io.StringIO = io.StringIO()
self._description_message_gen: DescriptionMessageGenerator = (
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] = {}
self._current_matching_triggers_with_routers: dict[str, Router] = (
self._matching_lower_triggers_with_routers
if self._ignore_command_register
else self._matching_default_triggers_with_routers
self._incorrect_input_syntax_handler: NonStandardBehaviorHandler[str] = (
lambda _: print_func(f"Incorrect flag syntax: {_}")
)
self._incorrect_input_syntax_handler: NonStandardBehaviorHandler[str] = lambda _: print_func(
f"Incorrect flag syntax: {_}"
self._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = (
lambda _: print_func(f"Repeated input flags: {_}")
)
self._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = lambda _: print_func(
f"Repeated input flags: {_}"
self._empty_input_command_handler: EmptyCommandHandler = lambda: print_func(
"Empty input command"
)
self._empty_input_command_handler: EmptyCommandHandler = lambda: print_func("Empty input command")
self._unknown_command_handler: NonStandardBehaviorHandler[InputCommand] = lambda _: print_func(
f"Unknown command: {_.trigger}"
self._unknown_command_handler: NonStandardBehaviorHandler[InputCommand] = (
lambda _: print_func(f"Unknown command: {_.trigger}")
)
self._exit_command_handler: NonStandardBehaviorHandler[Response] = lambda _: print_func(
self._farewell_message
self._exit_command_handler: NonStandardBehaviorHandler[Response] = (
lambda _: print_func(self._farewell_message)
)
def set_description_message_pattern(self, _: DescriptionMessageGenerator, /) -> None:
@@ -97,7 +92,9 @@ class BaseApp:
"""
self._description_message_gen = _
def set_incorrect_input_syntax_handler(self, _: NonStandardBehaviorHandler[str], /) -> None:
def set_incorrect_input_syntax_handler(
self, _: NonStandardBehaviorHandler[str], /
) -> None:
"""
Public. Sets the handler for incorrect flags when entering a command
:param _: handler for incorrect flags when entering a command
@@ -105,7 +102,9 @@ class BaseApp:
"""
self._incorrect_input_syntax_handler = _
def set_repeated_input_flags_handler(self, _: NonStandardBehaviorHandler[str], /) -> None:
def set_repeated_input_flags_handler(
self, _: NonStandardBehaviorHandler[str], /
) -> None:
"""
Public. Sets the handler for repeated flags when entering a command
:param _: handler for repeated flags when entering a command
@@ -113,7 +112,9 @@ class BaseApp:
"""
self._repeated_input_flags_handler = _
def set_unknown_command_handler(self, _: NonStandardBehaviorHandler[InputCommand], /) -> None:
def set_unknown_command_handler(
self, _: NonStandardBehaviorHandler[InputCommand], /
) -> None:
"""
Public. Sets the handler for unknown commands when entering a command
:param _: handler for unknown commands when entering a command
@@ -129,7 +130,9 @@ class BaseApp:
"""
self._empty_input_command_handler = _
def set_exit_command_handler(self, _: NonStandardBehaviorHandler[Response], /) -> None:
def set_exit_command_handler(
self, _: NonStandardBehaviorHandler[Response], /
) -> None:
"""
Public. Sets the handler for exit command when entering a command
:param _: handler for exit command when entering a command
@@ -161,10 +164,14 @@ class BaseApp:
:return: None
"""
if isinstance(self._dividing_line, DynamicDividingLine):
clear_text = re.sub(r"\u001b\[[0-9;]*m", "", text)
clear_text = _ANSI_ESCAPE_RE.sub("", 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
max_length_line
if 10 <= max_length_line <= 80
else 80
if max_length_line > 80
else 10
)
self._print_func(
@@ -181,11 +188,15 @@ class BaseApp:
elif isinstance(self._dividing_line, StaticDividingLine): # pyright: ignore[reportUnnecessaryIsInstance]
self._print_func(
self._dividing_line.get_full_static_line(is_override=self._override_system_messages)
self._dividing_line.get_full_static_line(
is_override=self._override_system_messages
)
)
print(text.strip("\n"))
self._print_func(
self._dividing_line.get_full_static_line(is_override=self._override_system_messages)
self._dividing_line.get_full_static_line(
is_override=self._override_system_messages
)
)
else:
@@ -199,32 +210,28 @@ class BaseApp:
"""
trigger = command.trigger
exit_trigger = self._exit_command.trigger
if self._ignore_command_register:
if trigger.lower() == exit_trigger.lower():
return True
elif trigger.lower() in [x.lower() for x in self._exit_command.aliases]:
return True
else:
if trigger == exit_trigger:
return True
elif trigger in self._exit_command.aliases:
return False
def _is_unknown_command(self, input_command: InputCommand) -> bool:
if not self.registered_routers.get_router_by_trigger(input_command.trigger.lower()):
return True
return False
def _is_unknown_command(self, command: InputCommand) -> bool:
def _capture_stdout(self, func: Callable[[], None]) -> str:
"""
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
Private. Captures stdout from a function call using a reusable buffer
:param func: function to execute with captured stdout
:return: captured stdout as string
"""
input_command_trigger = command.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
self._stdout_buffer.seek(0)
self._stdout_buffer.truncate(0)
with redirect_stdout(self._stdout_buffer):
func()
return self._stdout_buffer.getvalue()
def _error_handler(self, error: InputCommandException, raw_command: str) -> None:
"""
@@ -245,17 +252,38 @@ class BaseApp:
Private. Sets up system router
:return: None
"""
system_router.title = self._system_router_title
@system_router.command(self._exit_command)
@self.system_router.command(self._exit_command)
def _(response: Response) -> None:
self._exit_command_handler(response)
system_router.command_register_ignore = self._ignore_command_register
self.registered_routers.add_registered_router(system_router)
self.registered_routers.add_registered_router(self.system_router)
def _validate_routers_for_collisions(self) -> None:
"""
Private. Validates that there are no trigger/alias collisions between routers
:return: None
:raises: RepeatedTriggerNameException or RepeatedAliasNameException if collision detected
"""
all_triggers: set[str] = set()
all_aliases: set[str] = set()
for router_entity in self.registered_routers:
union_units: set[str] = all_triggers | all_aliases
trigger_collisions: set[str] = union_units & router_entity.triggers
if trigger_collisions:
raise RepeatedTriggerNameException()
alias_collisions: set[str] = union_units & router_entity.aliases
if alias_collisions:
raise RepeatedAliasNameException(alias_collisions)
all_triggers.update(router_entity.triggers)
all_aliases.update(router_entity.aliases)
def _most_similar_command(self, unknown_command: str) -> str | None:
all_commands = list(self._current_matching_triggers_with_routers.keys())
all_commands = self.registered_routers.get_triggers()
matches_startswith_unknown_command: Matches = sorted(
cmd for cmd in all_commands if cmd.startswith(unknown_command)
@@ -279,7 +307,9 @@ class BaseApp:
: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._initial_message = (
"\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n"
)
self._farewell_message = (
"[bold red]\n\n"
+ str(text2art(self._farewell_message, font="chanky")) # pyright: ignore[reportUnknownArgumentType]
@@ -297,14 +327,20 @@ class BaseApp:
self._repeated_input_flags_handler = lambda raw_command: self._print_func(
f"[red bold]Repeated input flags: {escape(raw_command)}"
)
self._empty_input_command_handler = lambda: self._print_func("[red bold]Empty input command")
self._empty_input_command_handler = lambda: self._print_func(
"[red bold]Empty input command"
)
def unknown_command_handler(command: InputCommand) -> None:
cmd_trg: str = command.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]"
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 ""
("[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)
@@ -316,20 +352,9 @@ class BaseApp:
:return: None
"""
self._setup_system_router()
self._validate_routers_for_collisions()
for router_entity in self.registered_routers:
router_triggers = router_entity.triggers
router_aliases = router_entity.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()))
if not self._override_system_messages:
self._setup_default_view()
self._autocompleter.initial_setup(self.registered_routers.get_triggers())
self._print_func(self._initial_message)
@@ -337,11 +362,14 @@ class BaseApp:
self._print_func(message)
if self._messages_on_startup:
print("\n")
if not self._repeat_command_groups_printing_description:
if not self._repeat_command_groups_printing:
self._print_command_group_description()
def _process_exist_and_valid_command(self, input_command: InputCommand) -> None:
processing_router = self._current_matching_triggers_with_routers[input_command.trigger.lower()]
processing_router = self.registered_routers.get_router_by_trigger(input_command.trigger.lower())
if not processing_router:
raise RuntimeError(f"Router for '{input_command.trigger}' not found. Panic!")
if processing_router.disable_redirect_stdout:
dividing_line_unit_part: str = self._dividing_line.get_unit_part()
@@ -357,9 +385,9 @@ class BaseApp:
)
)
else:
with redirect_stdout(io.StringIO()) as stdout:
processing_router.finds_appropriate_handler(input_command)
stdout_result: str = stdout.getvalue()
stdout_result = self._capture_stdout(
lambda: processing_router.finds_appropriate_handler(input_command)
)
self._print_framed_text(stdout_result)
@@ -368,7 +396,7 @@ DEFAULT_DIVIDING_LINE: StaticDividingLine = StaticDividingLine()
DEFAULT_PRINT_FUNC: Printer = Console().print
DEFAULT_AUTOCOMPLETER: AutoCompleter = AutoCompleter()
DEFAULT_EXIT_COMMAND: Command = Command("Q", description="Exit command")
DEFAULT_EXIT_COMMAND: Command = Command("q", description="Exit command")
class App(BaseApp):
@@ -380,7 +408,6 @@ class App(BaseApp):
farewell_message: str = "\nSee you\n",
exit_command: Command = DEFAULT_EXIT_COMMAND,
system_router_title: str = "System points:",
ignore_command_register: bool = True,
dividing_line: AVAILABLE_DIVIDING_LINES = DEFAULT_DIVIDING_LINE,
repeat_command_groups_printing: bool = False,
override_system_messages: bool = False,
@@ -395,7 +422,6 @@ class App(BaseApp):
: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_printing: whether to repeat the available commands and their description
:param override_system_messages: whether to redefine the default formatting of system messages
@@ -409,13 +435,14 @@ class App(BaseApp):
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_printing=repeat_command_groups_printing,
override_system_messages=override_system_messages,
autocompleter=autocompleter,
print_func=print_func,
)
if not self._override_system_messages:
self._setup_default_view()
def run_polling(self) -> None:
"""
@@ -424,31 +451,29 @@ class App(BaseApp):
"""
self._pre_cycle_setup()
while True:
if self._repeat_command_groups_printing_description:
if self._repeat_command_groups_printing:
self._print_command_group_description()
raw_command: str = Console().input(self._prompt)
try:
input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
except InputCommandException as error:
with redirect_stdout(io.StringIO()) as stderr:
self._error_handler(error, raw_command)
stderr_result: str = stderr.getvalue()
except InputCommandException as error: # noqa F841
stderr_result = self._capture_stdout(
lambda: self._error_handler(error, raw_command) # noqa F821
)
self._print_framed_text(stderr_result)
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()), self._ignore_command_register
)
self.system_router.finds_appropriate_handler(input_command)
self._autocompleter.exit_setup(self.registered_routers.get_triggers())
return
if self._is_unknown_command(input_command):
with redirect_stdout(io.StringIO()) as stdout:
self._unknown_command_handler(input_command)
stdout_res: str = stdout.getvalue()
stdout_res = self._capture_stdout(
lambda: self._unknown_command_handler(input_command)
)
self._print_framed_text(stdout_res)
continue
@@ -460,7 +485,6 @@ class App(BaseApp):
:param router: registered router
:return: None
"""
router.command_register_ignore = self._ignore_command_register
self.registered_routers.add_registered_router(router)
def include_routers(self, *routers: Router) -> None:
+10 -4
View File
@@ -1,9 +1,10 @@
__all__ = ["NonStandardBehaviorHandler", "EmptyCommandHandler", "Printer", "DescriptionMessageGenerator"]
__all__ = ["NonStandardBehaviorHandler", "EmptyCommandHandler", "Printer", "DescriptionMessageGenerator", "HandlerFunc"]
from typing import Protocol, TypeVar
from typing import ParamSpec, Protocol, TypeVar
from argenta.response import Response
T = TypeVar("T", contravariant=True) # noqa: WPS111
T = TypeVar("T", contravariant=True)
P = ParamSpec("P")
class NonStandardBehaviorHandler(Protocol[T]):
@@ -24,3 +25,8 @@ class Printer(Protocol):
class DescriptionMessageGenerator(Protocol):
def __call__(self, _command: str, _description: str, /) -> str:
raise NotImplementedError
class HandlerFunc(Protocol):
def __call__(self, response: Response) -> None:
raise NotImplementedError
+12 -3
View File
@@ -1,18 +1,19 @@
__all__ = ["RegisteredRouters"]
from typing import Iterator, Optional
from typing import Iterator
from argenta.router import Router
class RegisteredRouters:
def __init__(self, registered_routers: Optional[list[Router]] = None) -> None:
def __init__(self) -> None:
"""
Private. Combines registered routers
:param registered_routers: list of the registered routers
:return: None
"""
self.registered_routers: list[Router] = registered_routers if registered_routers else []
self.registered_routers: list[Router] = []
self._paired_trigger_router: dict[str, Router] = {}
def add_registered_router(self, router: Router, /) -> None:
"""
@@ -21,6 +22,14 @@ class RegisteredRouters:
:return: None
"""
self.registered_routers.append(router)
for trigger in (router.aliases | router.triggers):
self._paired_trigger_router[trigger] = router
def get_router_by_trigger(self, trigger: str) -> Router | None:
return self._paired_trigger_router.get(trigger)
def get_triggers(self) -> set[str]:
return set(self._paired_trigger_router.keys())
def __iter__(self) -> Iterator[Router]:
return iter(self.registered_routers)
-1
View File
@@ -4,7 +4,6 @@ from enum import Enum
from re import Pattern
from typing import Literal, override
PREFIX_TYPE = Literal["-", "--", "---"]
+18 -8
View File
@@ -1,7 +1,7 @@
__all__ = ["Command", "InputCommand"]
import shlex
from typing import Never, Self, cast, Literal
from typing import Literal, Never, Self, cast
from argenta.command.exceptions import (
EmptyInputCommandException,
@@ -38,20 +38,23 @@ class Command:
:param flags: processed commands
:param aliases: string synonyms for the main trigger
"""
self.registered_flags: Flags = flags if isinstance(flags, Flags) else Flags([flags])
pretty_flags = flags if isinstance(flags, Flags) else Flags([flags])
self.registered_flags: Flags = pretty_flags
self.trigger: str = trigger
self.description: str = description
self.aliases: set[str] | set[Never] = aliases
self._paired_string_entity_flag: dict[str, Flag] = {
flag.string_entity: flag for flag in pretty_flags
}
def validate_input_flag(self, flag: InputFlag) -> ValidationStatus:
"""
Private. Validates the input flag
:param flag: input flag for validation
:return: is input flag valid as bool
"""
registered_flags: Flags = self.registered_flags
for registered_flag in registered_flags:
if registered_flag.string_entity == flag.string_entity:
if registered_flag := self._paired_string_entity_flag.get(flag.string_entity):
is_valid = registered_flag.validate_input_flag_value(flag.input_value)
if is_valid:
return ValidationStatus.VALID
@@ -61,7 +64,12 @@ class Command:
class InputCommand:
def __init__(self, trigger: str, *, input_flags: InputFlag | InputFlags = DEFAULT_WITHOUT_INPUT_FLAGS):
def __init__(
self,
trigger: str,
*,
input_flags: InputFlag | InputFlags = DEFAULT_WITHOUT_INPUT_FLAGS,
):
"""
Private. The model of the input command, after parsing
:param trigger:the trigger of the command
@@ -70,7 +78,9 @@ class InputCommand:
"""
self.trigger: str = trigger
self.input_flags: InputFlags = (
input_flags if isinstance(input_flags, InputFlags) else InputFlags([input_flags])
input_flags
if isinstance(input_flags, InputFlags)
else InputFlags([input_flags])
)
@classmethod
@@ -115,7 +125,7 @@ class InputCommand:
name=name,
prefix=cast(PREFIX_TYPE, prefix), # pyright: ignore[reportUnnecessaryCast]
input_value=input_value,
status=None
status=None,
)
if input_flag in flags:
-1
View File
@@ -1 +0,0 @@
from argenta.metrics.main import get_time_of_pre_cycle_setup as get_time_of_pre_cycle_setup
-22
View File
@@ -1,22 +0,0 @@
__all__ = [
"get_time_of_pre_cycle_setup",
]
import io
from contextlib import redirect_stdout
from time import time
from argenta 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() # pyright: ignore[reportPrivateUsage]
end = time()
return end - start
@@ -1,3 +1,4 @@
from argenta.orchestrator.argparser.arguments.models import BooleanArgument as BooleanArgument
from argenta.orchestrator.argparser.arguments.models import \
BooleanArgument as BooleanArgument
from argenta.orchestrator.argparser.arguments.models import InputArgument as InputArgument
from argenta.orchestrator.argparser.arguments.models import ValueArgument as ValueArgument
+2 -5
View File
@@ -7,12 +7,9 @@ import sys
from argparse import ArgumentParser, Namespace
from typing import Never, Self
from argenta.orchestrator.argparser.arguments.models import (
BaseArgument,
from argenta.orchestrator.argparser.arguments.models import (BaseArgument,
BooleanArgument,
InputArgument,
ValueArgument,
)
InputArgument, ValueArgument)
class ArgSpace:
+2 -2
View File
@@ -14,7 +14,7 @@ class Orchestrator:
def __init__(
self,
arg_parser: ArgParser = DEFAULT_ARGPARSER,
custom_providers: list[Provider] = [],
custom_providers: list[Provider] | None = None,
auto_inject_handlers: bool = True,
):
"""
@@ -23,7 +23,7 @@ class Orchestrator:
:return: None
"""
self._arg_parser: ArgParser = arg_parser
self._custom_providers: list[Provider] = custom_providers
self._custom_providers: list[Provider] = custom_providers or []
self._auto_inject_handlers: bool = auto_inject_handlers
self._arg_parser._parse_args() # pyright: ignore[reportPrivateUsage]
+13 -5
View File
@@ -1,20 +1,21 @@
__all__ = ["CommandHandler", "CommandHandlers"]
from collections.abc import Iterator
from typing import Callable
from typing import Never
from argenta.app.protocols import HandlerFunc
from argenta.command import Command
from argenta.response import Response
class CommandHandler:
def __init__(self, handler_as_func: Callable[..., None], handled_command: Command):
def __init__(self, handler_as_func: HandlerFunc, 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_as_func: Callable[..., None] = handler_as_func
self.handler_as_func: HandlerFunc = handler_as_func
self.handled_command: Command = handled_command
def handling(self, response: Response) -> None:
@@ -27,12 +28,13 @@ class CommandHandler:
class CommandHandlers:
def __init__(self, command_handlers: list[CommandHandler] | None = None):
def __init__(self, command_handlers: tuple[CommandHandler] | tuple[Never, ...] = tuple()):
"""
Private. The model that unites all CommandHandler of the routers
:param command_handlers: list of CommandHandlers for register
"""
self.command_handlers: list[CommandHandler] = command_handlers if command_handlers else []
self.command_handlers: list[CommandHandler] = list(command_handlers) if command_handlers else []
self.paired_command_handler_trigger: dict[str, CommandHandler] = {x.handled_command.trigger: x for x in command_handlers}
def add_handler(self, command_handler: CommandHandler) -> None:
"""
@@ -41,6 +43,12 @@ class CommandHandlers:
:return: None
"""
self.command_handlers.append(command_handler)
self.paired_command_handler_trigger[command_handler.handled_command.trigger.lower()] = command_handler
for alias in command_handler.handled_command.aliases:
self.paired_command_handler_trigger[alias.lower()] = command_handler
def get_command_handler_by_trigger(self, trigger: str) -> CommandHandler | None:
return self.paired_command_handler_trigger.get(trigger)
def __iter__(self) -> Iterator[CommandHandler]:
return iter(self.command_handlers)
-5
View File
@@ -1,5 +0,0 @@
__all__ = ["system_router"]
from argenta.router import Router
system_router = Router(title="System points:")
+33 -27
View File
@@ -1,22 +1,21 @@
__all__ = ["Router"]
from inspect import get_annotations, getfullargspec, getsourcefile, getsourcelines
from typing import Callable, TypeAlias
from typing import Callable
from rich.console import Console
from argenta.app.protocols import HandlerFunc
from argenta.command import Command, InputCommand
from argenta.command.flag import ValidationStatus
from argenta.command.flag.flags import Flags, InputFlags
from argenta.command.flag.flags import InputFlags
from argenta.response import Response, ResponseStatus
from argenta.router.command_handler.entity import CommandHandler, CommandHandlers
from argenta.router.exceptions import (
from argenta.router.exceptions import (RepeatedAliasNameException,
RepeatedFlagNameException,
RepeatedTriggerNameException,
RequiredArgumentNotPassedException,
TriggerContainSpacesException,
)
HandlerFunc: TypeAlias = Callable[..., None]
TriggerContainSpacesException)
class Router:
@@ -40,8 +39,6 @@ class Router:
self.disable_redirect_stdout: bool = disable_redirect_stdout
self.command_handlers: CommandHandlers = CommandHandlers()
self.command_register_ignore: bool = False
self.aliases: set[str] = set()
self.triggers: set[str] = set()
@@ -57,12 +54,7 @@ class Router:
redefined_command = command
self._validate_command(redefined_command)
if overlapping := (self.aliases | self.triggers) & redefined_command.aliases:
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{overlapping}[/b blue]")
self.aliases.update(redefined_command.aliases)
self.triggers.add(redefined_command.trigger)
self._update_routing_keys(redefined_command)
def decorator(func: HandlerFunc) -> HandlerFunc:
_validate_func_args(func)
@@ -80,25 +72,39 @@ class Router:
command_name: str = command.trigger
if command_name.find(" ") != -1:
raise TriggerContainSpacesException()
flags: Flags = command.registered_flags
flags_name: list[str] = [flag.string_entity.lower() for flag in flags]
if command_name.lower() in self.triggers:
raise RepeatedTriggerNameException()
if command_name.lower() in self.aliases:
raise RepeatedAliasNameException({command_name.lower()})
if overlapping := (self.aliases | self.triggers) & set(map(lambda x: x.lower(), command.aliases)):
raise RepeatedAliasNameException(overlapping)
flags_name: list[str] = [flag.string_entity.lower() for flag in command.registered_flags]
if len(set(flags_name)) < len(flags_name):
raise RepeatedFlagNameException()
def _update_routing_keys(self, registered_command: Command) -> None:
redefined_command_aliases_in_lower = set(map(lambda x: x.lower(), registered_command.aliases))
self.aliases.update(redefined_command_aliases_in_lower)
self.triggers.add(registered_command.trigger.lower())
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.trigger
input_command_name: str = input_command.trigger.lower()
input_command_flags: InputFlags = input_command.input_flags
for command_handler in self.command_handlers:
handle_command = command_handler.handled_command
if input_command_name.lower() == handle_command.trigger.lower():
self.process_input_command(input_command_flags, command_handler)
if input_command_name.lower() in handle_command.aliases:
command_handler = self.command_handlers.get_command_handler_by_trigger(input_command_name)
if not command_handler:
raise RuntimeError(f"Handler for '{input_command.trigger}' command not found. Panic!")
else:
self.process_input_command(input_command_flags, command_handler)
def process_input_command(self, input_command_flags: InputFlags, command_handler: CommandHandler) -> None:
@@ -147,13 +153,14 @@ def _structuring_input_flags(handled_command: Command, input_flags: InputFlags)
undefined_flags = True
status = ResponseStatus.from_flags(
has_invalid_value_flags=invalid_value_flags, has_undefined_flags=undefined_flags
has_invalid_value_flags=invalid_value_flags,
has_undefined_flags=undefined_flags
)
return Response(status=status, input_flags=input_flags)
def _validate_func_args(func: Callable[..., None]) -> None:
def _validate_func_args(func: HandlerFunc) -> None:
"""
Private. Validates the arguments of the handler
:param func: entity of the handler func
@@ -168,8 +175,7 @@ def _validate_func_args(func: Callable[..., None]) -> None:
response_arg_annotation = func_annotations.get(response_arg)
if response_arg_annotation is not None:
if response_arg_annotation is not Response:
if response_arg_annotation is not None and response_arg_annotation is not Response:
source_line: int = getsourcelines(func)[1]
Console().print(
f'\nFile "{getsourcefile(func)}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
+31 -1
View File
@@ -1,4 +1,10 @@
__all__ = ["RepeatedFlagNameException", "RequiredArgumentNotPassedException", "TriggerContainSpacesException"]
__all__ = [
"RepeatedFlagNameException",
"RepeatedTriggerNameException",
"RepeatedAliasNameException",
"RequiredArgumentNotPassedException",
"TriggerContainSpacesException",
]
from typing import override
@@ -13,6 +19,30 @@ class RepeatedFlagNameException(Exception):
return "Repeated registered flag names in register command"
class RepeatedTriggerNameException(Exception):
"""
Private. Raised when a repeated trigger name is registered
"""
@override
def __str__(self) -> str:
return "Repeated trigger name in registered commands"
class RepeatedAliasNameException(Exception):
"""
Private. Raised when a repeated alias name is registered
"""
@override
def __init__(self, repeated_aliases: set[str]) -> None:
self.repeated_aliases = repeated_aliases
super().__init__()
@override
def __str__(self) -> str:
return f"Repeated aliases names: {self.repeated_aliases}"
class RequiredArgumentNotPassedException(Exception):
"""
Private. Raised when a required argument is not passed
@@ -72,27 +72,6 @@ def test_unknown_command_triggers_unknown_command_handler(monkeypatch: pytest.Mo
assert "\nUnknown command: help\n" in output
def test_case_sensitive_command_triggers_unknown_command_handler(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
inputs = iter(["TeSt", "Q"])
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('test command')
app = App(ignore_command_register=False, override_system_messages=True, print_func=print)
app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
orchestrator.start_polling(app)
output = capsys.readouterr().out
assert '\nUnknown command: TeSt\n' in output
def test_mixed_valid_and_unknown_commands_handled_correctly(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
inputs = iter(["test", "some", "q"])
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
@@ -46,26 +46,6 @@ def test_simple_command_executes_successfully(monkeypatch: pytest.MonkeyPatch, c
assert '\ntest command\n' in output
def test_case_insensitive_command_executes_when_enabled(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
inputs = iter(["TeSt", "q"])
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
router = Router()
orchestrator = Orchestrator()
@router.command(Command('test'))
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
print('test command')
app = App(ignore_command_register=True, override_system_messages=True, print_func=print)
app.include_router(router)
orchestrator.start_polling(app)
output = capsys.readouterr().out
assert '\ntest command\n' in output
def test_two_commands_execute_sequentially(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
inputs = iter(["test", "some", "q"])
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
+92 -50
View File
@@ -1,3 +1,4 @@
from argenta.router.exceptions import RepeatedAliasNameException
import pytest
from pytest import CaptureFixture
@@ -25,46 +26,21 @@ def test_default_exit_command_uppercase_q_is_recognized() -> None:
assert app._is_exit_command(InputCommand('Q')) is True
def test_exit_command_not_recognized_when_case_sensitivity_enabled() -> None:
app = App(ignore_command_register=False)
assert app._is_exit_command(InputCommand('q')) is False
def test_custom_exit_command_is_recognized() -> None:
app = App(exit_command=Command('quit'))
assert app._is_exit_command(InputCommand('quit')) is True
def test_custom_exit_command_case_insensitive_by_default() -> None:
app = App(exit_command=Command('quit'))
assert app._is_exit_command(InputCommand('qUIt')) is True
def test_custom_exit_command_case_sensitive_when_enabled() -> None:
app = App(ignore_command_register=False, exit_command=Command('quit'))
assert app._is_exit_command(InputCommand('qUIt')) is False
def test_exit_command_alias_is_recognized() -> None:
app = App(exit_command=Command('q', aliases={'exit'}))
assert app._is_exit_command(InputCommand('exit')) is True
def test_exit_command_alias_case_sensitive_when_enabled() -> None:
app = App(exit_command=Command('q', aliases={'exit'}), ignore_command_register=False)
assert app._is_exit_command(InputCommand('exit')) is True
def test_non_exit_command_is_not_recognized() -> None:
app = App(exit_command=Command('q', aliases={'exit'}))
assert app._is_exit_command(InputCommand('quit')) is False
def test_non_exit_command_with_wrong_case_is_not_recognized() -> None:
app = App(exit_command=Command('q', aliases={'exit'}), ignore_command_register=False)
assert app._is_exit_command(InputCommand('Exit')) is False
# ============================================================================
# Tests for unknown command detection
# ============================================================================
@@ -73,31 +49,22 @@ def test_non_exit_command_with_wrong_case_is_not_recognized() -> None:
def test_registered_command_is_not_unknown() -> None:
app = App()
app.set_unknown_command_handler(lambda command: None)
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
router = Router()
@router.command('fr')
def handler(res: Response):
pass
app.include_router(router)
assert app._is_unknown_command(InputCommand('fr')) is False
def test_unregistered_command_is_unknown() -> None:
app = App()
app.set_unknown_command_handler(lambda command: None)
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
assert app._is_unknown_command(InputCommand('cr')) is True
def test_command_with_wrong_case_is_unknown_when_case_sensitivity_enabled() -> None:
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()}
assert app._is_unknown_command(InputCommand('pr')) is True
def test_command_with_exact_case_is_not_unknown_when_case_sensitivity_enabled() -> None:
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()}
assert app._is_unknown_command(InputCommand('tW')) is False
# ============================================================================
# Tests for similar command suggestions
# ============================================================================
@@ -207,24 +174,101 @@ def test_include_routers_registers_multiple_routers() -> None:
assert app.registered_routers.registered_routers == [router, router2]
def test_overlapping_aliases_prints_warning(capsys: CaptureFixture[str]) -> None:
app = App(override_system_messages=True)
def test_overlapping_aliases_raises_exception() -> None:
router = Router()
@router.command(Command('test', aliases={'alias'}))
def handler(_res: Response) -> None:
pass
with pytest.raises(RepeatedAliasNameException):
@router.command(Command('test2', aliases={'alias'}))
def handler2(_res: Response) -> None:
pass
app.include_routers(router)
def test_app_detects_trigger_collision_between_routers() -> None:
from argenta.router.exceptions import RepeatedTriggerNameException
app = App()
router1 = Router()
router2 = Router()
@router1.command('hello')
def handler1(_res: Response) -> None:
pass
@router2.command('hello')
def handler2(_res: Response) -> None:
pass
app.include_router(router1)
app.include_router(router2)
with pytest.raises(RepeatedTriggerNameException):
app._pre_cycle_setup()
captured = capsys.readouterr()
assert "Overlapping" in captured.out
def test_app_detects_alias_collision_between_routers() -> None:
app = App()
router1 = Router()
router2 = Router()
@router1.command(Command('hello', aliases={'hi'}))
def handler1(_res: Response) -> None:
pass
@router2.command(Command('world', aliases={'hi'}))
def handler2(_res: Response) -> None:
pass
app.include_router(router1)
app.include_router(router2)
with pytest.raises(RepeatedAliasNameException):
app._pre_cycle_setup()
def test_app_detects_trigger_alias_collision_between_routers() -> None:
app = App()
router1 = Router()
router2 = Router()
@router1.command('hello')
def handler1(_res: Response) -> None:
pass
@router2.command(Command('world', aliases={'hello'}))
def handler2(_res: Response) -> None:
pass
app.include_router(router1)
app.include_router(router2)
with pytest.raises(RepeatedAliasNameException):
app._pre_cycle_setup()
def test_app_detects_collision_case_insensitive() -> None:
from argenta.router.exceptions import RepeatedTriggerNameException
app = App()
router1 = Router()
router2 = Router()
@router1.command('Hello')
def handler1(_res: Response) -> None:
pass
@router2.command('hELLo')
def handler2(_res: Response) -> None:
pass
app.include_router(router1)
app.include_router(router2)
with pytest.raises(RepeatedTriggerNameException):
app._pre_cycle_setup()
# ============================================================================
@@ -317,8 +361,6 @@ def test_set_exit_command_handler_stores_handler() -> None:
def test_setup_default_view_formats_prompt() -> None:
app = App(prompt='>>')
app._setup_default_view()
assert app._prompt == '[italic dim bold]>>'
@@ -554,7 +596,7 @@ def test_handler_can_be_replaced_multiple_times() -> None:
def test_handler_receives_correct_parameters() -> None:
app = App()
received_data = {'trigger': None}
received_data: dict[str, None | str] = {'trigger': None}
def custom_handler(command: InputCommand) -> None:
received_data['trigger'] = command.trigger
@@ -567,7 +609,7 @@ def test_handler_receives_correct_parameters() -> None:
def test_exit_handler_receives_response_object() -> None:
app = App()
received_data = {'response': None}
received_data: dict[str, None | Response] = {'response': None}
def custom_handler(response: Response) -> None:
received_data['response'] = response
+4 -21
View File
@@ -7,13 +7,12 @@ from pytest_mock import MockerFixture
from argenta.app.autocompleter.entity import (
AutoCompleter,
_get_history_items,
_is_command_exist,
_get_history_items
)
HISTORY_FILE: str = "test_history.txt"
COMMANDS: list[str] = ["start", "stop", "status"]
COMMANDS: set[str] = {"start", "stop", "status"}
# ============================================================================
@@ -119,7 +118,7 @@ def test_exit_setup_writes_and_filters_duplicate_commands(fs: FakeFilesystem, mo
fs.create_file(HISTORY_FILE, contents=raw_history_content) # pyright: ignore[reportUnknownMemberType]
completer: AutoCompleter = AutoCompleter(history_filename=HISTORY_FILE)
completer.exit_setup(all_commands=["start", "stop"], ignore_command_register=False)
completer.exit_setup(all_commands={"start", "stop"})
mock_readline.write_history_file.assert_called_once_with(HISTORY_FILE)
@@ -131,7 +130,7 @@ def test_exit_setup_writes_and_filters_duplicate_commands(fs: FakeFilesystem, mo
def test_exit_setup_skips_writing_when_no_history_filename(mock_readline: Any) -> None:
completer: AutoCompleter = AutoCompleter(history_filename=None)
completer.exit_setup(all_commands=COMMANDS, ignore_command_register=False)
completer.exit_setup(all_commands=COMMANDS)
mock_readline.write_history_file.assert_not_called()
@@ -182,22 +181,6 @@ def test_complete_inserts_common_prefix_for_multiple_matches(mock_readline: Any)
# ============================================================================
def test_is_command_exist_checks_case_sensitive_when_enabled() -> None:
existing: list[str] = ["start", "stop", "status"]
assert _is_command_exist("start", existing, ignore_command_register=False) is True
assert _is_command_exist("START", existing, ignore_command_register=False) is False
assert _is_command_exist("unknown", existing, ignore_command_register=False) is False
def test_is_command_exist_checks_case_insensitive_when_enabled() -> None:
existing: list[str] = ["start", "stop", "status"]
assert _is_command_exist("start", existing, ignore_command_register=True) is True
assert _is_command_exist("START", existing, ignore_command_register=True) is True
assert _is_command_exist("unknown", existing, ignore_command_register=True) is False
def test_get_history_items_returns_empty_list_initially(mock_readline: Any) -> None:
assert _get_history_items() == []
+192 -1
View File
@@ -9,9 +9,11 @@ from argenta.command.flag.flags import Flags, InputFlags
from argenta.command.flag.models import PossibleValues, ValidationStatus
from argenta.response.entity import Response
from argenta.router import Router
from argenta.router.entity import _structuring_input_flags, _validate_func_args # pyright: ignore[reportPrivateUsage]
from argenta.router.entity import _structuring_input_flags, _validate_func_args
from argenta.router.exceptions import (
RepeatedAliasNameException,
RepeatedFlagNameException,
RepeatedTriggerNameException,
RequiredArgumentNotPassedException,
TriggerContainSpacesException,
)
@@ -28,6 +30,19 @@ def test_validate_command_raises_error_for_trigger_with_spaces() -> None:
router._validate_command(Command(trigger='command with spaces'))
def test_validate_command_raises_error_for_same_trigger() -> None:
router = Router()
@router.command('comm')
def handler(res: Response):
pass
with pytest.raises(RepeatedTriggerNameException):
@router.command('comm')
def handler2(res: Response):
pass
def test_validate_command_raises_error_for_repeated_flag_names() -> None:
router = Router()
with pytest.raises(RepeatedFlagNameException):
@@ -193,6 +208,33 @@ def test_finds_appropriate_handler_executes_handler_by_alias(capsys: CaptureFixt
assert "Hello World!" in output.out
def test_finds_appropriate_handler_executes_handler_by_alias_case_insensitive(capsys: CaptureFixture[str]) -> None:
router = Router()
@router.command(Command('hello', aliases={'hI'}))
def handler(_res: Response) -> None:
print("Hello World!")
router.finds_appropriate_handler(InputCommand('HI'))
output = capsys.readouterr()
assert "Hello World!" in output.out
def test_finds_appropriate_handler_executes_handler_by_trigger_case_insensitive(capsys: CaptureFixture[str]) -> None:
router = Router()
@router.command(Command('heLLo'))
def handler(_res: Response) -> None:
print("Hello World!")
router.finds_appropriate_handler(InputCommand('HellO'))
output = capsys.readouterr()
assert "Hello World!" in output.out
def test_finds_appropriate_handler_executes_handler_with_flags_by_alias(capsys: CaptureFixture[str]) -> None:
router = Router()
@@ -206,3 +248,152 @@ def test_finds_appropriate_handler_executes_handler_with_flags_by_alias(capsys:
output = capsys.readouterr()
assert "Hello World!" in output.out
# ============================================================================
# Tests for alias and trigger collision detection
# ============================================================================
def test_validate_command_raises_error_for_alias_collision_with_existing_trigger() -> None:
router = Router()
@router.command('hello')
def handler(_res: Response) -> None:
pass
with pytest.raises(RepeatedAliasNameException):
@router.command(Command('world', aliases={'hello'}))
def handler2(_res: Response) -> None:
pass
def test_validate_command_raises_error_for_alias_collision_with_existing_alias() -> None:
router = Router()
@router.command(Command('hello', aliases={'hi'}))
def handler(_res: Response) -> None:
pass
with pytest.raises(RepeatedAliasNameException):
@router.command(Command('world', aliases={'hi'}))
def handler2(_res: Response) -> None:
pass
def test_validate_command_raises_error_for_trigger_collision_with_existing_alias() -> None:
router = Router()
@router.command(Command('hello', aliases={'hi'}))
def handler(_res: Response) -> None:
pass
with pytest.raises(RepeatedAliasNameException):
@router.command('hi')
def handler2(_res: Response) -> None:
pass
def test_validate_command_raises_error_for_alias_collision_case_insensitive() -> None:
router = Router()
@router.command(Command('hello', aliases={'Hi'}))
def handler(_res: Response) -> None:
pass
with pytest.raises(RepeatedAliasNameException):
@router.command(Command('world', aliases={'hI'}))
def handler2(_res: Response) -> None:
pass
# ============================================================================
# Tests for RegisteredRouters
# ============================================================================
def test_registered_routers_get_router_by_trigger() -> None:
from argenta.app.registered_routers.entity import RegisteredRouters
registered_routers = RegisteredRouters()
router = Router()
@router.command('hello')
def handler(_res: Response) -> None:
pass
registered_routers.add_registered_router(router)
assert registered_routers.get_router_by_trigger('hello') == router
def test_registered_routers_get_router_by_alias() -> None:
from argenta.app.registered_routers.entity import RegisteredRouters
registered_routers = RegisteredRouters()
router = Router()
@router.command(Command('hello', aliases={'hi'}))
def handler(_res: Response) -> None:
pass
registered_routers.add_registered_router(router)
assert registered_routers.get_router_by_trigger('hi') == router
def test_registered_routers_get_router_case_insensitive() -> None:
from argenta.app.registered_routers.entity import RegisteredRouters
registered_routers = RegisteredRouters()
router = Router()
@router.command(Command('HeLLo'))
def handler(_res: Response) -> None:
pass
registered_routers.add_registered_router(router)
# Trigger stored in lowercase, should match regardless of case
assert registered_routers.get_router_by_trigger('hello') == router
assert registered_routers.get_router_by_trigger('HELLO') is None # Exact match required in dict
def test_registered_routers_get_triggers_returns_all_triggers_and_aliases() -> None:
from argenta.app.registered_routers.entity import RegisteredRouters
registered_routers = RegisteredRouters()
router1 = Router()
router2 = Router()
@router1.command(Command('hello', aliases={'hi'}))
def handler1(_res: Response) -> None:
pass
@router2.command(Command('world', aliases={'w'}))
def handler2(_res: Response) -> None:
pass
registered_routers.add_registered_router(router1)
registered_routers.add_registered_router(router2)
triggers = registered_routers.get_triggers()
assert 'hello' in triggers
assert 'hi' in triggers
assert 'world' in triggers
assert 'w' in triggers
def test_registered_routers_returns_none_for_unknown_trigger() -> None:
from argenta.app.registered_routers.entity import RegisteredRouters
registered_routers = RegisteredRouters()
router = Router()
@router.command('hello')
def handler(_res: Response) -> None:
pass
registered_routers.add_registered_router(router)
assert registered_routers.get_router_by_trigger('unknown') is None