mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
Compare commits
28 Commits
v1.0.0-beta2
...
v1.0.7
| Author | SHA1 | Date | |
|---|---|---|---|
| c8e0729be8 | |||
| c2bbc5f15d | |||
| 0acbf54e44 | |||
| c3d9541330 | |||
| f6561de9b3 | |||
| bebd84969b | |||
| 365347ea7f | |||
| 33cb528b1d | |||
| fd287c5da0 | |||
| 45f410e3e8 | |||
| 8b06e9cd39 | |||
| c38fe10006 | |||
| 03cbc64f48 | |||
| cbf7d3c578 | |||
| ea2d068022 | |||
| 5991851207 | |||
| f628c3b5b5 | |||
| 05379712f4 | |||
| ed1cbf0fcf | |||
| 471f05369b | |||
| 13f7e33db1 | |||
| 9a78aa9263 | |||
| 58ccd6b26d | |||
| 73144f7ba4 | |||
| 650f4c9036 | |||
| 393f5c7d81 | |||
| 9eb2bb6c46 | |||
| 79b275eac7 |
@@ -27,4 +27,4 @@ jobs:
|
||||
pip install ruff
|
||||
|
||||
- name: Run linter
|
||||
run: ruff check ./argenta
|
||||
run: ruff check ./src
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
.venv
|
||||
*venv
|
||||
.idea
|
||||
dist
|
||||
uv.lock
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
# Argenta
|
||||
|
||||
### Bibliothek zum Erstellen modularer CLI-Anwendungen
|
||||
|
||||

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

|
||||
|
||||
---
|
||||
|
||||
# Установка
|
||||
```bash
|
||||
pip install argenta
|
||||
```
|
||||
or
|
||||
```bash
|
||||
poetry add argenta
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
# Быстрый старт
|
||||
|
||||
Пример простейшего приложения
|
||||
```python
|
||||
# routers.py
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
from argenta.response import Response
|
||||
|
||||
|
||||
router = Router()
|
||||
|
||||
@router.command(Command("hello"))
|
||||
def handler(response: Response):
|
||||
print("Hello, world!")
|
||||
```
|
||||
|
||||
```python
|
||||
# main.py
|
||||
from argenta.app import App
|
||||
from argenta.orchestrator import Orchestrator
|
||||
from routers import router
|
||||
|
||||
app: App = App()
|
||||
orchestrator: Orchestrator = Orchestrator()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
app.include_router(router)
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
# Фичи в разработке
|
||||
|
||||
- Полноценная поддержка автокомплитера на Linux
|
||||
|
||||
## Полная [документация](https://argenta-docs.vercel.app) | MIT 2025 kolo | made by [kolo](https://t.me/kolo_id)
|
||||
|
||||
|
||||
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 1.5 MiB |
Binary file not shown.
|
After Width: | Height: | Size: 1.9 MiB |
Binary file not shown.
|
After Width: | Height: | Size: 928 KiB |
@@ -0,0 +1,46 @@
|
||||
<?xml version="1.0" standalone="no"?>
|
||||
<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 20010904//EN"
|
||||
"http://www.w3.org/TR/2001/REC-SVG-20010904/DTD/svg10.dtd">
|
||||
<svg version="1.0" xmlns="http://www.w3.org/2000/svg"
|
||||
width="809.000000pt" height="809.000000pt" viewBox="0 0 809.000000 809.000000"
|
||||
preserveAspectRatio="xMidYMid meet">
|
||||
|
||||
<g transform="translate(0.000000,809.000000) scale(0.100000,-0.100000)"
|
||||
fill="#000000" stroke="none">
|
||||
<path d="M3845 7600 c-683 -38 -1333 -259 -1875 -639 -702 -492 -1223 -1251
|
||||
-1434 -2089 -83 -330 -111 -602 -103 -980 6 -273 19 -393 67 -632 167 -829
|
||||
642 -1593 1321 -2122 512 -399 1133 -656 1794 -745 165 -22 666 -25 825 -5
|
||||
779 99 1451 397 2020 898 687 603 1095 1401 1201 2344 17 147 17 571 0 715
|
||||
-59 518 -204 977 -442 1402 -180 322 -355 552 -629 823 -518 515 -1164 850
|
||||
-1887 979 -254 45 -594 66 -858 51z m655 -234 c545 -76 1005 -248 1445 -541
|
||||
195 -130 336 -245 506 -415 457 -456 758 -987 909 -1605 134 -547 132 -1119
|
||||
-5 -1665 -149 -593 -445 -1112 -885 -1550 -624 -623 -1414 -966 -2315 -1006
|
||||
-412 -18 -935 73 -1339 232 -656 259 -1228 720 -1610 1299 -298 450 -461 890
|
||||
-538 1451 -28 210 -31 620 -4 819 103 786 437 1477 975 2016 149 149 266 249
|
||||
417 357 315 225 692 405 1059 506 211 58 342 81 675 120 87 10 601 -3 710 -18z"/>
|
||||
<path d="M3691 6759 c-231 -17 -522 -67 -660 -114 -227 -77 -354 -211 -381
|
||||
-400 -14 -105 -13 -628 2 -643 9 -9 158 -12 611 -12 598 0 628 -2 649 -34 14
|
||||
-21 8 -66 -12 -86 -20 -20 -33 -20 -908 -20 -857 0 -892 -1 -967 -20 -294 -75
|
||||
-500 -321 -599 -715 -49 -195 -61 -309 -60 -585 0 -221 3 -272 23 -385 44
|
||||
-251 116 -418 232 -540 78 -82 143 -123 254 -162 78 -27 85 -27 351 -31 l272
|
||||
-4 16 23 c14 20 16 58 16 259 0 334 20 446 107 613 80 153 268 310 447 374
|
||||
128 45 138 46 881 52 683 7 713 8 786 28 201 56 353 177 437 348 80 162 86
|
||||
240 74 905 -11 608 -10 600 -84 743 -80 153 -240 271 -453 332 -252 73 -653
|
||||
101 -1034 74z m-383 -466 c126 -78 147 -245 44 -355 -144 -154 -396 -54 -395
|
||||
157 0 97 62 187 154 222 54 20 143 10 197 -24z"/>
|
||||
<path d="M5451 5491 c-20 -20 -21 -30 -21 -283 0 -276 -7 -351 -42 -458 -81
|
||||
-251 -281 -454 -523 -534 -149 -48 -151 -48 -890 -56 -678 -6 -703 -7 -780
|
||||
-28 -139 -38 -219 -84 -315 -181 -61 -61 -95 -105 -117 -151 -65 -134 -64
|
||||
-129 -75 -780 -12 -665 -10 -694 47 -817 122 -265 419 -429 895 -494 127 -18
|
||||
592 -18 730 -1 377 48 646 169 785 355 57 75 67 94 96 182 20 64 23 94 27 325
|
||||
4 238 3 257 -15 278 l-18 23 -551 -1 c-399 0 -559 3 -577 11 -51 23 -56 102
|
||||
-8 126 9 4 439 10 956 13 1044 7 981 2 1125 75 178 89 305 255 386 502 84 257
|
||||
111 618 73 973 -39 364 -167 655 -351 801 -63 50 -121 80 -213 110 -64 20 -93
|
||||
23 -337 27 -260 4 -267 3 -287 -17z m-2104 -1698 c78 -52 531 -413 561 -447
|
||||
17 -19 33 -48 37 -65 15 -71 3 -84 -323 -356 -275 -229 -310 -255 -342 -255
|
||||
-78 1 -112 75 -61 134 9 10 86 75 171 144 247 201 355 294 358 311 1 9 -15 28
|
||||
-38 44 -127 89 -503 393 -512 413 -23 50 20 103 82 104 15 0 45 -12 67 -27z
|
||||
m1388 -1244 c59 -15 129 -77 151 -134 36 -96 0 -204 -88 -262 -36 -24 -51 -28
|
||||
-118 -28 -67 0 -82 4 -118 28 -153 101 -124 338 48 392 57 17 71 18 125 4z"/>
|
||||
</g>
|
||||
</svg>
|
||||
|
After Width: | Height: | Size: 2.9 KiB |
@@ -0,0 +1,36 @@
|
||||
from argenta.command import Command
|
||||
from argenta.metrics import get_time_of_pre_cycle_setup
|
||||
from argenta.response import Response
|
||||
from argenta.router import Router
|
||||
from argenta.app import App
|
||||
|
||||
|
||||
|
||||
class TimeOfPreCycleSetup:
|
||||
@staticmethod
|
||||
def commands_with_two_aliases(num_of_commands: int):
|
||||
router = Router()
|
||||
|
||||
for i in range(num_of_commands):
|
||||
@router.command(Command(f'cmd{i}', aliases=[f'cdr{i}', f'prt{i}']))
|
||||
def handler(response: Response):
|
||||
pass
|
||||
|
||||
app = App()
|
||||
app.include_router(router)
|
||||
|
||||
return get_time_of_pre_cycle_setup(app)
|
||||
|
||||
@staticmethod
|
||||
def commands_with_one_aliases(num_of_commands: int):
|
||||
router = Router()
|
||||
|
||||
for i in range(num_of_commands):
|
||||
@router.command(Command(f'cmd{i}', aliases=[f'cdr{i}']))
|
||||
def handler(response: Response):
|
||||
pass
|
||||
|
||||
app = App()
|
||||
app.include_router(router)
|
||||
|
||||
return get_time_of_pre_cycle_setup(app)
|
||||
@@ -1,26 +0,0 @@
|
||||
from mock.mock_app.handlers.routers import work_router
|
||||
|
||||
from argenta.app import App
|
||||
from argenta.app.defaults import PredefinedMessages
|
||||
from argenta.app.autocompleter import AutoCompleter
|
||||
from argenta.orchestrator import Orchestrator
|
||||
from argenta.orchestrator.argparser import ArgParser
|
||||
from argenta.orchestrator.argparser.arguments import BooleanArgument
|
||||
|
||||
|
||||
arg_parser = ArgParser(processed_args=[BooleanArgument('repeat')])
|
||||
app: App = App(autocompleter=AutoCompleter('.hist'))
|
||||
orchestrator: Orchestrator = Orchestrator()
|
||||
|
||||
|
||||
def main():
|
||||
app.include_router(work_router)
|
||||
|
||||
app.add_message_on_startup(PredefinedMessages.USAGE)
|
||||
app.add_message_on_startup(PredefinedMessages.AUTOCOMPLETE)
|
||||
app.add_message_on_startup(PredefinedMessages.HELP)
|
||||
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+10
-15
@@ -1,23 +1,18 @@
|
||||
from argenta.response import Response, Status
|
||||
from argenta.app import App
|
||||
from argenta.app.dividing_line import StaticDividingLine, DynamicDividingLine
|
||||
from argenta.app.autocompleter import AutoCompleter
|
||||
from argenta.app.defaults import PredefinedMessages
|
||||
from argenta.command import Command
|
||||
from argenta.command.flags import Flags, InputFlags, InvalidValueInputFlags, UndefinedInputFlags, ValidInputFlags
|
||||
from argenta.command.flag import Flag, InputFlag
|
||||
from argenta.command.flag.defaults import PredefinedFlags
|
||||
from argenta.router import Router
|
||||
from argenta.orchestrator import Orchestrator
|
||||
|
||||
from argenta.command.models import InputCommand
|
||||
import inspect
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test(response):
|
||||
print('test command')
|
||||
|
||||
@router.command(Command('some'))
|
||||
def handler(res: Response) -> Response:
|
||||
pass
|
||||
|
||||
app = App(ignore_command_register=True,
|
||||
override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
orchestrator.start_polling(app)
|
||||
@@ -1,10 +0,0 @@
|
||||
from rich.console import Console
|
||||
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
def help_command():
|
||||
console.print("[italic bold]The main functionality of the script is to convert an expression from a string "
|
||||
"to a mathematical one and then calculate this expression. "
|
||||
"Project GitHub: https://github.com/koloideal/WordMath[/italic bold]")
|
||||
@@ -1,4 +1,4 @@
|
||||
from mock.mock_app.handlers.routers import work_router
|
||||
from mock.mock_app.routers import work_router
|
||||
|
||||
from argenta.app import App
|
||||
from argenta.app.defaults import PredefinedMessages
|
||||
@@ -9,10 +9,12 @@ from argenta.orchestrator.argparser import ArgParser
|
||||
from argenta.orchestrator.argparser.arguments import BooleanArgument
|
||||
|
||||
|
||||
arg_parser = ArgParser(processed_args=[BooleanArgument('repeat')])
|
||||
app: App = App(dividing_line=DynamicDividingLine(),
|
||||
autocompleter=AutoCompleter('./mock/.hist'),
|
||||
repeat_command_groups=False,)
|
||||
arg_parser = ArgParser(processed_args=[BooleanArgument("repeat")])
|
||||
app: App = App(
|
||||
dividing_line=DynamicDividingLine(),
|
||||
autocompleter=AutoCompleter(),
|
||||
repeat_command_groups=False,
|
||||
)
|
||||
orchestrator: Orchestrator = Orchestrator(arg_parser)
|
||||
|
||||
|
||||
@@ -25,5 +27,6 @@ def main():
|
||||
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -2,30 +2,38 @@ from rich.console import Console
|
||||
|
||||
from argenta.command import Command
|
||||
from argenta.command.flag.defaults import PredefinedFlags
|
||||
from argenta.command.flags import Flags
|
||||
from argenta.command.flag import Flags, Flag, PossibleValues
|
||||
from argenta.response import Response
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
work_router: Router = Router(title='Work points:')
|
||||
work_router: Router = Router(title="Work points:", disable_redirect_stdout=True)
|
||||
|
||||
console = Console()
|
||||
|
||||
flag = Flag('csdv', possible_values=PossibleValues.DISABLE)
|
||||
|
||||
@work_router.command(Command('get', 'Get Help', aliases=['help', 'Get_help'], flags=Flags(PredefinedFlags.PORT, PredefinedFlags.HOST)))
|
||||
|
||||
@work_router.command(
|
||||
Command(
|
||||
"get",
|
||||
"Get Help",
|
||||
aliases=["help", "Get_help"],
|
||||
flags=Flags(PredefinedFlags.PORT, PredefinedFlags.HOST),
|
||||
)
|
||||
)
|
||||
def command_help(response: Response):
|
||||
case = input("test > ")
|
||||
print(case)
|
||||
print(response.status)
|
||||
print(response.undefined_flags.get_flags())
|
||||
print(response.valid_flags.get_flags())
|
||||
print(response.invalid_value_flags.get_flags())
|
||||
|
||||
|
||||
@work_router.command('run')
|
||||
@work_router.command("run")
|
||||
def command_start_solving(response: Response):
|
||||
print(response.status)
|
||||
print(response.undefined_flags.get_flags())
|
||||
print(response.valid_flags.get_flags())
|
||||
print(response.invalid_value_flags.get_flags())
|
||||
|
||||
|
||||
|
||||
+7
-7
@@ -1,9 +1,9 @@
|
||||
[project]
|
||||
name = "argenta"
|
||||
version = "1.0.0-beta2"
|
||||
version = "1.0.7"
|
||||
description = "Python library for building modular CLI applications"
|
||||
authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }]
|
||||
requires-python = ">=3.11, <4.0"
|
||||
requires-python = ">=3.8"
|
||||
readme = "README.md"
|
||||
license = { text = "MIT" }
|
||||
dependencies = [
|
||||
@@ -12,11 +12,6 @@ dependencies = [
|
||||
"pyreadline3>=3.5.4",
|
||||
]
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"pydoc-markdown>=4.8.2,<5",
|
||||
]
|
||||
|
||||
[tool.ruff]
|
||||
exclude = [
|
||||
".idea",
|
||||
@@ -31,3 +26,8 @@ exclude = [
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"psutil>=7.0.0",
|
||||
]
|
||||
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
import os
|
||||
import readline
|
||||
from typing import Never
|
||||
|
||||
|
||||
class AutoCompleter:
|
||||
def __init__(self, history_filename: str = False, 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
|
||||
@@ -12,7 +15,6 @@ class AutoCompleter:
|
||||
"""
|
||||
self.history_filename = history_filename
|
||||
self.autocomplete_button = autocomplete_button
|
||||
self.matches: list[str] = []
|
||||
|
||||
def _complete(self, text, state) -> str | None:
|
||||
"""
|
||||
@@ -21,16 +23,22 @@ 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 self.get_history_items() if cmd.startswith(text))
|
||||
matches: list[str] = sorted(
|
||||
cmd for cmd in self.get_history_items() if cmd.startswith(text)
|
||||
)
|
||||
if len(matches) > 1:
|
||||
common_prefix = matches[0]
|
||||
for match in matches[1:]:
|
||||
i = 0
|
||||
while i < len(common_prefix) and i < len(match) and common_prefix[i] == match[i]:
|
||||
while (
|
||||
i < len(common_prefix)
|
||||
and i < len(match)
|
||||
and common_prefix[i] == match[i]
|
||||
):
|
||||
i += 1
|
||||
common_prefix = common_prefix[:i]
|
||||
if state == 0:
|
||||
readline.insert_text(common_prefix[len(text):])
|
||||
readline.insert_text(common_prefix[len(text) :])
|
||||
readline.redisplay()
|
||||
return None
|
||||
elif len(matches) == 1:
|
||||
@@ -52,21 +60,32 @@ class AutoCompleter:
|
||||
readline.add_history(line)
|
||||
|
||||
readline.set_completer(self._complete)
|
||||
readline.set_completer_delims(readline.get_completer_delims().replace(' ', ''))
|
||||
readline.parse_and_bind(f'{self.autocomplete_button}: complete')
|
||||
readline.set_completer_delims(readline.get_completer_delims().replace(" ", ""))
|
||||
readline.parse_and_bind(f"{self.autocomplete_button}: complete")
|
||||
|
||||
def exit_setup(self) -> None:
|
||||
def exit_setup(self, all_commands: list[str]) -> None:
|
||||
"""
|
||||
Private. Exit setup function
|
||||
:return: None
|
||||
"""
|
||||
if self.history_filename:
|
||||
readline.write_history_file(self.history_filename)
|
||||
with open(self.history_filename, "r") as history_file:
|
||||
raw_history = history_file.read()
|
||||
pretty_history: list[str] = []
|
||||
for line in set(raw_history.strip().split("\n")):
|
||||
if line.split()[0] in all_commands:
|
||||
pretty_history.append(line)
|
||||
with open(self.history_filename, "w") as history_file:
|
||||
history_file.write("\n".join(pretty_history))
|
||||
|
||||
@staticmethod
|
||||
def get_history_items() -> list[str] | list:
|
||||
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]
|
||||
: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)
|
||||
]
|
||||
|
||||
@@ -1,12 +1,11 @@
|
||||
from dataclasses import dataclass
|
||||
from enum import StrEnum
|
||||
|
||||
|
||||
@dataclass
|
||||
class PredefinedMessages:
|
||||
class PredefinedMessages(StrEnum):
|
||||
"""
|
||||
Public. A dataclass with predetermined messages for quick use
|
||||
"""
|
||||
USAGE = '[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]'
|
||||
HELP = '[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]'
|
||||
AUTOCOMPLETE = '[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>'
|
||||
|
||||
USAGE = "[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]"
|
||||
HELP = "[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]"
|
||||
AUTOCOMPLETE = "[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>"
|
||||
|
||||
@@ -2,7 +2,7 @@ from abc import ABC
|
||||
|
||||
|
||||
class BaseDividingLine(ABC):
|
||||
def __init__(self, unit_part: str = '-') -> None:
|
||||
def __init__(self, unit_part: str = "-") -> None:
|
||||
"""
|
||||
Private. The basic dividing line
|
||||
:param unit_part: the single part of the dividing line
|
||||
@@ -16,13 +16,13 @@ class BaseDividingLine(ABC):
|
||||
:return: unit_part of dividing line as str
|
||||
"""
|
||||
if len(self._unit_part) == 0:
|
||||
return ' '
|
||||
return " "
|
||||
else:
|
||||
return self._unit_part[0]
|
||||
|
||||
|
||||
class StaticDividingLine(BaseDividingLine):
|
||||
def __init__(self, unit_part: str = '-', length: int = 25) -> None:
|
||||
def __init__(self, unit_part: str = "-", length: int = 25) -> None:
|
||||
"""
|
||||
Public. The static dividing line
|
||||
:param unit_part: the single part of the dividing line
|
||||
@@ -39,13 +39,13 @@ class StaticDividingLine(BaseDividingLine):
|
||||
:return: full line of dividing line as str
|
||||
"""
|
||||
if is_override:
|
||||
return f'\n{self.length * self.get_unit_part()}\n'
|
||||
return f"\n{self.length * self.get_unit_part()}\n"
|
||||
else:
|
||||
return f'\n[dim]{self.length * self.get_unit_part()}[/dim]\n'
|
||||
return f"\n[dim]{self.length * self.get_unit_part()}[/dim]\n"
|
||||
|
||||
|
||||
class DynamicDividingLine(BaseDividingLine):
|
||||
def __init__(self, unit_part: str = '-') -> None:
|
||||
def __init__(self, unit_part: str = "-") -> None:
|
||||
"""
|
||||
Public. The dynamic dividing line
|
||||
:param unit_part: the single part of the dividing line
|
||||
@@ -61,8 +61,6 @@ class DynamicDividingLine(BaseDividingLine):
|
||||
:return: full line of dividing line as str
|
||||
"""
|
||||
if is_override:
|
||||
return f'\n{length * self.get_unit_part()}\n'
|
||||
return f"\n{length * self.get_unit_part()}\n"
|
||||
else:
|
||||
return f'\n[dim]{self.get_unit_part() * length}[/dim]\n'
|
||||
|
||||
|
||||
return f"\n[dim]{self.get_unit_part() * length}[/dim]\n"
|
||||
|
||||
+157
-98
@@ -11,18 +11,19 @@ from argenta.router import Router
|
||||
from argenta.router.defaults import system_router
|
||||
from argenta.app.autocompleter import AutoCompleter
|
||||
from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
from argenta.command.exceptions import (
|
||||
UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException,
|
||||
BaseInputCommandException)
|
||||
BaseInputCommandException,
|
||||
)
|
||||
from argenta.app.registered_routers.entity import RegisteredRouters
|
||||
from argenta.response import Response
|
||||
|
||||
|
||||
|
||||
class BaseApp:
|
||||
def __init__(self,
|
||||
prompt: str,
|
||||
def __init__(self, prompt: str,
|
||||
initial_message: str,
|
||||
farewell_message: str,
|
||||
exit_command: Command,
|
||||
@@ -33,7 +34,6 @@ class BaseApp:
|
||||
override_system_messages: bool,
|
||||
autocompleter: AutoCompleter,
|
||||
print_func: Callable[[str], None]) -> None:
|
||||
|
||||
self._prompt = prompt
|
||||
self._print_func = print_func
|
||||
self._exit_command = exit_command
|
||||
@@ -47,20 +47,33 @@ class BaseApp:
|
||||
self._farewell_message = farewell_message
|
||||
self._initial_message = initial_message
|
||||
|
||||
|
||||
self._description_message_gen: Callable[[str, str], str] = lambda command, description: f'[{command}] *=*=* {description}'
|
||||
self._description_message_gen: Callable[[str, str], str] = lambda command, description: f"{command} *=*=* {description}"
|
||||
self._registered_routers: RegisteredRouters = RegisteredRouters()
|
||||
self._messages_on_startup: list[str] = []
|
||||
|
||||
self._all_registered_triggers_in_lower: list[str] = []
|
||||
self._all_registered_triggers_in_default_case: list[str] = []
|
||||
self._matching_lower_triggers_with_routers: dict[str, Router] = {}
|
||||
self._matching_default_triggers_with_routers: dict[str, Router] = {}
|
||||
|
||||
self._invalid_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Incorrect flag syntax: {raw_command}')
|
||||
self._repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Repeated input flags: {raw_command}')
|
||||
self._empty_input_command_handler: Callable[[], None] = lambda: print_func('Empty input command')
|
||||
self._unknown_command_handler: Callable[[InputCommand], None] = lambda command: print_func(f"Unknown command: {command.get_trigger()}")
|
||||
self._exit_command_handler: Callable[[Response], None] = lambda response: print_func(self._farewell_message)
|
||||
if self._ignore_command_register:
|
||||
self._current_matching_triggers_with_routers: dict[str, Router] = self._matching_lower_triggers_with_routers
|
||||
else:
|
||||
self._current_matching_triggers_with_routers: dict[str, Router] = self._matching_default_triggers_with_routers
|
||||
|
||||
self._incorrect_input_syntax_handler: Callable[[str], None] = (
|
||||
lambda raw_command: print_func(f"Incorrect flag syntax: {raw_command}")
|
||||
)
|
||||
self._repeated_input_flags_handler: Callable[[str], None] = (
|
||||
lambda raw_command: print_func(f"Repeated input flags: {raw_command}")
|
||||
)
|
||||
self._empty_input_command_handler: Callable[[], None] = lambda: print_func(
|
||||
"Empty input command"
|
||||
)
|
||||
self._unknown_command_handler: Callable[[InputCommand], None] = (
|
||||
lambda command: print_func(f"Unknown command: {command.get_trigger()}")
|
||||
)
|
||||
self._exit_command_handler: Callable[[Response], None] = (
|
||||
lambda response: print_func(self._farewell_message)
|
||||
)
|
||||
|
||||
def set_description_message_pattern(self, _: Callable[[str, str], str]) -> None:
|
||||
"""
|
||||
@@ -70,15 +83,13 @@ class BaseApp:
|
||||
"""
|
||||
self._description_message_gen: Callable[[str, str], str] = _
|
||||
|
||||
|
||||
def set_invalid_input_flags_handler(self, _: Callable[[str], None]) -> None:
|
||||
def set_incorrect_input_syntax_handler(self, _: Callable[[str], None]) -> None:
|
||||
"""
|
||||
Public. Sets the handler for incorrect flags when entering a command
|
||||
:param _: handler for incorrect flags when entering a command
|
||||
:return: None
|
||||
"""
|
||||
self._invalid_input_flags_handler = _
|
||||
|
||||
self._incorrect_input_syntax_handler = _
|
||||
|
||||
def set_repeated_input_flags_handler(self, _: Callable[[str], None]) -> None:
|
||||
"""
|
||||
@@ -88,7 +99,6 @@ class BaseApp:
|
||||
"""
|
||||
self._repeated_input_flags_handler = _
|
||||
|
||||
|
||||
def set_unknown_command_handler(self, _: Callable[[str], None]) -> None:
|
||||
"""
|
||||
Public. Sets the handler for unknown commands when entering a command
|
||||
@@ -97,7 +107,6 @@ class BaseApp:
|
||||
"""
|
||||
self._unknown_command_handler = _
|
||||
|
||||
|
||||
def set_empty_command_handler(self, _: Callable[[], None]) -> None:
|
||||
"""
|
||||
Public. Sets the handler for empty commands when entering a command
|
||||
@@ -106,7 +115,6 @@ class BaseApp:
|
||||
"""
|
||||
self._empty_input_command_handler = _
|
||||
|
||||
|
||||
def set_exit_command_handler(self, _: Callable[[], None]) -> None:
|
||||
"""
|
||||
Public. Sets the handler for exit command when entering a command
|
||||
@@ -115,21 +123,22 @@ class BaseApp:
|
||||
"""
|
||||
self._exit_command_handler = _
|
||||
|
||||
|
||||
def _print_command_group_description(self) -> None:
|
||||
"""
|
||||
Private. Prints the description of the available commands
|
||||
:return: None
|
||||
"""
|
||||
for registered_router in self._registered_routers:
|
||||
if registered_router.get_title():
|
||||
self._print_func(registered_router.get_title())
|
||||
if registered_router.title:
|
||||
self._print_func(registered_router.title)
|
||||
for command_handler in registered_router.get_command_handlers():
|
||||
self._print_func(self._description_message_gen(
|
||||
self._print_func(
|
||||
self._description_message_gen(
|
||||
command_handler.get_handled_command().get_trigger(),
|
||||
command_handler.get_handled_command().get_description()))
|
||||
self._print_func('')
|
||||
|
||||
command_handler.get_handled_command().get_description(),
|
||||
)
|
||||
)
|
||||
self._print_func("")
|
||||
|
||||
def _print_framed_text(self, text: str) -> None:
|
||||
"""
|
||||
@@ -138,19 +147,36 @@ class BaseApp:
|
||||
:return: None
|
||||
"""
|
||||
if isinstance(self._dividing_line, StaticDividingLine):
|
||||
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
|
||||
print(text.strip('\n'))
|
||||
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
|
||||
self._print_func(
|
||||
self._dividing_line.get_full_static_line(self._override_system_messages)
|
||||
)
|
||||
print(text.strip("\n"))
|
||||
self._print_func(
|
||||
self._dividing_line.get_full_static_line(self._override_system_messages)
|
||||
)
|
||||
|
||||
elif isinstance(self._dividing_line, DynamicDividingLine):
|
||||
clear_text = re.sub(r'\u001b\[[0-9;]*m', '', text)
|
||||
max_length_line = max([len(line) for line in clear_text.split('\n')])
|
||||
max_length_line = max_length_line if 10 <= max_length_line <= 80 else 80 if max_length_line > 80 else 10
|
||||
|
||||
self._print_func(self._dividing_line.get_full_dynamic_line(max_length_line, self._override_system_messages))
|
||||
print(text.strip('\n'))
|
||||
self._print_func(self._dividing_line.get_full_dynamic_line(max_length_line, self._override_system_messages))
|
||||
clear_text = re.sub(r"\u001b\[[0-9;]*m", "", text)
|
||||
max_length_line = max([len(line) for line in clear_text.split("\n")])
|
||||
max_length_line = (
|
||||
max_length_line
|
||||
if 10 <= max_length_line <= 80
|
||||
else 80
|
||||
if max_length_line > 80
|
||||
else 10
|
||||
)
|
||||
|
||||
self._print_func(
|
||||
self._dividing_line.get_full_dynamic_line(
|
||||
max_length_line, self._override_system_messages
|
||||
)
|
||||
)
|
||||
print(text.strip("\n"))
|
||||
self._print_func(
|
||||
self._dividing_line.get_full_dynamic_line(
|
||||
max_length_line, self._override_system_messages
|
||||
)
|
||||
)
|
||||
|
||||
def _is_exit_command(self, command: InputCommand) -> bool:
|
||||
"""
|
||||
@@ -159,9 +185,14 @@ class BaseApp:
|
||||
:return: is it an exit command or not as bool
|
||||
"""
|
||||
if self._ignore_command_register:
|
||||
if command.get_trigger().lower() == self._exit_command.get_trigger().lower():
|
||||
if (
|
||||
command.get_trigger().lower()
|
||||
== self._exit_command.get_trigger().lower()
|
||||
):
|
||||
return True
|
||||
elif command.get_trigger().lower() in [x.lower() for x in self._exit_command.get_aliases()]:
|
||||
elif command.get_trigger().lower() in [
|
||||
x.lower() for x in self._exit_command.get_aliases()
|
||||
]:
|
||||
return True
|
||||
else:
|
||||
if command.get_trigger() == self._exit_command.get_trigger():
|
||||
@@ -170,7 +201,6 @@ class BaseApp:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_unknown_command(self, command: InputCommand) -> bool:
|
||||
"""
|
||||
Private. Checks if the given command is an unknown command
|
||||
@@ -179,36 +209,35 @@ class BaseApp:
|
||||
"""
|
||||
input_command_trigger = command.get_trigger()
|
||||
if self._ignore_command_register:
|
||||
if input_command_trigger.lower() in self._all_registered_triggers_in_lower:
|
||||
if input_command_trigger.lower() in list(self._current_matching_triggers_with_routers.keys()):
|
||||
return False
|
||||
else:
|
||||
if input_command_trigger in self._all_registered_triggers_in_default_case:
|
||||
if input_command_trigger in list(self._current_matching_triggers_with_routers.keys()):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _error_handler(self, error: BaseInputCommandException, raw_command: str) -> None:
|
||||
def _error_handler(
|
||||
self, error: BaseInputCommandException, raw_command: str
|
||||
) -> None:
|
||||
"""
|
||||
Private. Handles parsing errors of the entered command
|
||||
:param error: error being handled
|
||||
:param raw_command: the raw input command
|
||||
:return: None
|
||||
"""
|
||||
match error:
|
||||
case UnprocessedInputFlagException():
|
||||
self._invalid_input_flags_handler(raw_command)
|
||||
case RepeatedInputFlagsException():
|
||||
if isinstance(error, UnprocessedInputFlagException):
|
||||
self._incorrect_input_syntax_handler(raw_command)
|
||||
elif isinstance(error, RepeatedInputFlagsException):
|
||||
self._repeated_input_flags_handler(raw_command)
|
||||
case EmptyInputCommandException():
|
||||
elif isinstance(error, EmptyInputCommandException):
|
||||
self._empty_input_command_handler()
|
||||
|
||||
|
||||
def _setup_system_router(self) -> None:
|
||||
"""
|
||||
Private. Sets up system router
|
||||
:return: None
|
||||
"""
|
||||
system_router.set_title(self._system_router_title)
|
||||
system_router.title = self._system_router_title
|
||||
|
||||
@system_router.command(self._exit_command)
|
||||
def exit_command(response: Response) -> None:
|
||||
@@ -218,12 +247,16 @@ class BaseApp:
|
||||
system_router.set_command_register_ignore(self._ignore_command_register)
|
||||
self._registered_routers.add_registered_router(system_router)
|
||||
|
||||
|
||||
def _most_similar_command(self, unknown_command: str) -> str | None:
|
||||
all_commands = self._all_registered_triggers_in_lower if self._ignore_command_register else self._all_registered_triggers_in_default_case
|
||||
matches: list[str] | list = sorted(cmd for cmd in all_commands if cmd.startswith(unknown_command))
|
||||
all_commands = list(self._current_matching_triggers_with_routers.keys())
|
||||
|
||||
matches: list[str] | list = sorted(
|
||||
cmd for cmd in all_commands if cmd.startswith(unknown_command)
|
||||
)
|
||||
if not matches:
|
||||
matches: list[str] | list = sorted(cmd for cmd in all_commands if unknown_command.startswith(cmd))
|
||||
matches: list[str] | list = sorted(
|
||||
cmd for cmd in all_commands if unknown_command.startswith(cmd)
|
||||
)
|
||||
if len(matches) == 1:
|
||||
return matches[0]
|
||||
elif len(matches) > 1:
|
||||
@@ -231,34 +264,42 @@ class BaseApp:
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
def _setup_default_view(self) -> None:
|
||||
"""
|
||||
Private. Sets up default app view
|
||||
:return: None
|
||||
"""
|
||||
self._prompt = '[italic dim bold]What do you want to do?\n'
|
||||
self._initial_message = f'\n[bold red]{text2art(self._initial_message, font="tarty1")}\n'
|
||||
self._farewell_message = (f'[bold red]\n{text2art(f"\n{self._farewell_message}\n", font="chanky")}[/bold red]\n'
|
||||
f'[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n')
|
||||
self._description_message_gen = lambda command, description: (f'[bold red]{escape("[" + command + "]")}[/bold red] '
|
||||
f'[blue dim]*=*=*[/blue dim] '
|
||||
f'[bold yellow italic]{escape(description)}')
|
||||
self._invalid_input_flags_handler = lambda raw_command: self._print_func(f'[red bold]Incorrect flag syntax: {escape(raw_command)}')
|
||||
self._repeated_input_flags_handler = lambda raw_command: self._print_func(f'[red bold]Repeated input flags: {escape(raw_command)}')
|
||||
self._empty_input_command_handler = lambda: self._print_func('[red bold]Empty input command')
|
||||
self._prompt = f"[italic dim bold]{self._prompt}"
|
||||
self._initial_message = ("\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n")
|
||||
self._farewell_message = (
|
||||
"[bold red]\n\n"
|
||||
+ text2art(self._farewell_message, font="chanky")
|
||||
+ "\n[/bold red]\n"
|
||||
"[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n"
|
||||
)
|
||||
self._description_message_gen = lambda command, description: (
|
||||
f"[bold red]{escape('[' + command + ']')}[/bold red] "
|
||||
f"[blue dim]*=*=*[/blue dim] "
|
||||
f"[bold yellow italic]{escape(description)}"
|
||||
)
|
||||
self._incorrect_input_syntax_handler = lambda raw_command: self._print_func(f"[red bold]Incorrect flag syntax: {escape(raw_command)}")
|
||||
self._repeated_input_flags_handler = lambda raw_command: self._print_func(f"[red bold]Repeated input flags: {escape(raw_command)}")
|
||||
self._empty_input_command_handler = lambda: self._print_func("[red bold]Empty input command")
|
||||
|
||||
def unknown_command_handler(command: InputCommand) -> None:
|
||||
cmd_trg: str = command.get_trigger()
|
||||
mst_sim_cmd: str | None = self._most_similar_command(cmd_trg)
|
||||
first_part_of_text = f"[red]Unknown command:[/red] [blue]{escape(cmd_trg)}[/blue]"
|
||||
second_part_of_text = ('[red], most similar:[/red] ' + ('[blue]' + mst_sim_cmd + '[/blue]')) if mst_sim_cmd else ''
|
||||
second_part_of_text = (
|
||||
("[red], most similar:[/red] " + ("[blue]" + mst_sim_cmd + "[/blue]"))
|
||||
if mst_sim_cmd
|
||||
else ""
|
||||
)
|
||||
self._print_func(first_part_of_text + second_part_of_text)
|
||||
|
||||
self._unknown_command_handler = unknown_command_handler
|
||||
|
||||
|
||||
def _pre_cycle_setup(self) -> None:
|
||||
def pre_cycle_setup(self) -> None:
|
||||
"""
|
||||
Private. Configures various aspects of the application before the start of the cycle
|
||||
:return: None
|
||||
@@ -266,13 +307,22 @@ class BaseApp:
|
||||
self._setup_system_router()
|
||||
|
||||
for router_entity in self._registered_routers:
|
||||
self._all_registered_triggers_in_default_case.extend(router_entity.get_triggers())
|
||||
self._all_registered_triggers_in_default_case.extend(router_entity.get_aliases())
|
||||
router_triggers = router_entity.get_triggers()
|
||||
router_aliases = router_entity.get_aliases()
|
||||
combined = router_triggers + router_aliases
|
||||
|
||||
self._all_registered_triggers_in_lower.extend([x.lower() for x in router_entity.get_triggers()])
|
||||
self._all_registered_triggers_in_lower.extend([x.lower() for x in router_entity.get_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(self._all_registered_triggers_in_lower)
|
||||
self._autocompleter.initial_setup(list(self._current_matching_triggers_with_routers.keys()))
|
||||
|
||||
seen = {}
|
||||
for item in list(self._current_matching_triggers_with_routers.keys()):
|
||||
if item in seen:
|
||||
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{item}[/b blue]")
|
||||
else:
|
||||
seen[item] = True
|
||||
|
||||
if not self._override_system_messages:
|
||||
self._setup_default_view()
|
||||
@@ -282,26 +332,26 @@ class BaseApp:
|
||||
for message in self._messages_on_startup:
|
||||
self._print_func(message)
|
||||
if self._messages_on_startup:
|
||||
print('\n')
|
||||
|
||||
print("\n")
|
||||
if not self._repeat_command_groups_description:
|
||||
self._print_command_group_description()
|
||||
|
||||
|
||||
|
||||
class App(BaseApp):
|
||||
def __init__(self,
|
||||
prompt: str = 'What do you want to do?\n',
|
||||
initial_message: str = '\nArgenta\n',
|
||||
farewell_message: str = '\nSee you\n',
|
||||
exit_command: Command = Command('Q', 'Exit command'),
|
||||
system_router_title: str | None = 'System points:',
|
||||
def __init__(
|
||||
self,
|
||||
prompt: str = "What do you want to do?\n",
|
||||
initial_message: str = "Argenta\n",
|
||||
farewell_message: str = "\nSee you\n",
|
||||
exit_command: Command = Command("Q", "Exit command"),
|
||||
system_router_title: str | None = "System points:",
|
||||
ignore_command_register: bool = True,
|
||||
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
|
||||
repeat_command_groups: bool = True,
|
||||
override_system_messages: bool = False,
|
||||
autocompleter: AutoCompleter = AutoCompleter(),
|
||||
print_func: Callable[[str], None] = Console().print) -> None:
|
||||
print_func: Callable[[str], None] = Console().print,
|
||||
) -> None:
|
||||
"""
|
||||
Public. The essence of the application itself.
|
||||
Configures and manages all aspects of the behavior and presentation of the user interacting with the user
|
||||
@@ -318,7 +368,8 @@ class App(BaseApp):
|
||||
:param print_func: system messages text output function
|
||||
:return: None
|
||||
"""
|
||||
super().__init__(prompt=prompt,
|
||||
super().__init__(
|
||||
prompt=prompt,
|
||||
initial_message=initial_message,
|
||||
farewell_message=farewell_message,
|
||||
exit_command=exit_command,
|
||||
@@ -328,15 +379,15 @@ class App(BaseApp):
|
||||
repeat_command_groups=repeat_command_groups,
|
||||
override_system_messages=override_system_messages,
|
||||
autocompleter=autocompleter,
|
||||
print_func=print_func)
|
||||
|
||||
print_func=print_func,
|
||||
)
|
||||
|
||||
def run_polling(self) -> None:
|
||||
"""
|
||||
Private. Starts the user input processing cycle
|
||||
:return: None
|
||||
"""
|
||||
self._pre_cycle_setup()
|
||||
self.pre_cycle_setup()
|
||||
while True:
|
||||
if self._repeat_command_groups_description:
|
||||
self._print_command_group_description()
|
||||
@@ -354,7 +405,7 @@ class App(BaseApp):
|
||||
|
||||
if self._is_exit_command(input_command):
|
||||
system_router.finds_appropriate_handler(input_command)
|
||||
self._autocompleter.exit_setup()
|
||||
self._autocompleter.exit_setup(list(self._current_matching_triggers_with_routers.keys()))
|
||||
return
|
||||
|
||||
if self._is_unknown_command(input_command):
|
||||
@@ -364,12 +415,23 @@ class App(BaseApp):
|
||||
self._print_framed_text(res)
|
||||
continue
|
||||
|
||||
with redirect_stdout(io.StringIO()) as f:
|
||||
for registered_router in self._registered_routers:
|
||||
registered_router.finds_appropriate_handler(input_command)
|
||||
res: str = f.getvalue()
|
||||
self._print_framed_text(res)
|
||||
processing_router = self._current_matching_triggers_with_routers[input_command.get_trigger().lower()]
|
||||
|
||||
if processing_router.disable_redirect_stdout:
|
||||
if isinstance(self._dividing_line, StaticDividingLine):
|
||||
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
|
||||
processing_router.finds_appropriate_handler(input_command)
|
||||
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
|
||||
else:
|
||||
self._print_func(StaticDividingLine(self._dividing_line.get_unit_part()).get_full_static_line(self._override_system_messages))
|
||||
processing_router.finds_appropriate_handler(input_command)
|
||||
self._print_func(StaticDividingLine(self._dividing_line.get_unit_part()).get_full_static_line(self._override_system_messages))
|
||||
else:
|
||||
with redirect_stdout(io.StringIO()) as f:
|
||||
processing_router.finds_appropriate_handler(input_command)
|
||||
res: str = f.getvalue()
|
||||
if res:
|
||||
self._print_framed_text(res)
|
||||
|
||||
def include_router(self, router: Router) -> None:
|
||||
"""
|
||||
@@ -380,7 +442,6 @@ class App(BaseApp):
|
||||
router.set_command_register_ignore(self._ignore_command_register)
|
||||
self._registered_routers.add_registered_router(router)
|
||||
|
||||
|
||||
def include_routers(self, *routers: Router) -> None:
|
||||
"""
|
||||
Public. Registers the routers in the application
|
||||
@@ -390,7 +451,6 @@ class App(BaseApp):
|
||||
for router in routers:
|
||||
self.include_router(router)
|
||||
|
||||
|
||||
def add_message_on_startup(self, message: str) -> None:
|
||||
"""
|
||||
Public. Adds a message that will be displayed when the application is launched
|
||||
@@ -398,4 +458,3 @@ class App(BaseApp):
|
||||
:return: None
|
||||
"""
|
||||
self._messages_on_startup.append(message)
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@ from argenta.router import Router
|
||||
|
||||
|
||||
class RegisteredRouters:
|
||||
def __init__(self, registered_routers: list[Router] = None) -> None:
|
||||
def __init__(self, registered_routers: list[Router] | None = None) -> None:
|
||||
"""
|
||||
Private. Combines registered routers
|
||||
:param registered_routers: list of the registered routers
|
||||
|
||||
@@ -5,6 +5,7 @@ class BaseInputCommandException(Exception):
|
||||
"""
|
||||
Private. Base exception class for all exceptions raised when parse input command
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
@@ -12,6 +13,7 @@ class UnprocessedInputFlagException(BaseInputCommandException):
|
||||
"""
|
||||
Private. Raised when an unprocessed input flag is detected
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Unprocessed Input Flags"
|
||||
|
||||
@@ -20,16 +22,21 @@ class RepeatedInputFlagsException(BaseInputCommandException):
|
||||
"""
|
||||
Private. Raised when repeated input flags are detected
|
||||
"""
|
||||
|
||||
def __init__(self, flag: Flag | InputFlag):
|
||||
self.flag = flag
|
||||
|
||||
def __str__(self):
|
||||
return ("Repeated Input Flags\n"
|
||||
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'")
|
||||
return (
|
||||
"Repeated Input Flags\n"
|
||||
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'"
|
||||
)
|
||||
|
||||
|
||||
class EmptyInputCommandException(BaseInputCommandException):
|
||||
"""
|
||||
Private. Raised when an empty input command is detected
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Input Command is empty"
|
||||
|
||||
@@ -1,4 +1,17 @@
|
||||
__all__ = ["Flag", "InputFlag"]
|
||||
__all__ = [
|
||||
"Flag",
|
||||
"InputFlag",
|
||||
"UndefinedInputFlags",
|
||||
"ValidInputFlags",
|
||||
"InvalidValueInputFlags",
|
||||
"Flags", "PossibleValues"
|
||||
]
|
||||
|
||||
|
||||
from argenta.command.flag.models import Flag, InputFlag
|
||||
from argenta.command.flag.models import Flag, InputFlag, PossibleValues
|
||||
from argenta.command.flag.flags.models import (
|
||||
UndefinedInputFlags,
|
||||
ValidInputFlags,
|
||||
Flags,
|
||||
InvalidValueInputFlags,
|
||||
)
|
||||
|
||||
@@ -1,24 +1,32 @@
|
||||
from dataclasses import dataclass
|
||||
from argenta.command.flag.models import Flag
|
||||
from argenta.command.flag.models import Flag, PossibleValues
|
||||
import re
|
||||
|
||||
|
||||
|
||||
@dataclass
|
||||
class PredefinedFlags:
|
||||
"""
|
||||
Public. A dataclass with predefined flags and most frequently used flags for quick use
|
||||
"""
|
||||
HELP = Flag(name='help', possible_values=False)
|
||||
SHORT_HELP = Flag(name='H', prefix='-', possible_values=False)
|
||||
|
||||
INFO = Flag(name='info', possible_values=False)
|
||||
SHORT_INFO = Flag(name='I', prefix='-', possible_values=False)
|
||||
HELP = Flag(name="help", possible_values=PossibleValues.DISABLE)
|
||||
SHORT_HELP = Flag(name="H", prefix="-", possible_values=PossibleValues.DISABLE)
|
||||
|
||||
ALL = Flag(name='all', possible_values=False)
|
||||
SHORT_ALL = Flag(name='A', prefix='-', possible_values=False)
|
||||
INFO = Flag(name="info", possible_values=PossibleValues.DISABLE)
|
||||
SHORT_INFO = Flag(name="I", prefix="-", possible_values=PossibleValues.DISABLE)
|
||||
|
||||
HOST = Flag(name='host', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
|
||||
SHORT_HOST = Flag(name='H', prefix='-', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
|
||||
ALL = Flag(name="all", possible_values=PossibleValues.DISABLE)
|
||||
SHORT_ALL = Flag(name="A", prefix="-", possible_values=PossibleValues.DISABLE)
|
||||
|
||||
PORT = Flag(name='port', possible_values=re.compile(r'^\d{1,5}$'))
|
||||
SHORT_PORT = Flag(name='P', prefix='-', possible_values=re.compile(r'^\d{1,5}$'))
|
||||
HOST = Flag(
|
||||
name="host", possible_values=re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
|
||||
)
|
||||
SHORT_HOST = Flag(
|
||||
name="H",
|
||||
prefix="-",
|
||||
possible_values=re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$"),
|
||||
)
|
||||
|
||||
PORT = Flag(name="port", possible_values=re.compile(r"^\d{1,5}$"))
|
||||
SHORT_PORT = Flag(name="P", prefix="-", possible_values=re.compile(r"^\d{1,5}$"))
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
__all__ = [
|
||||
"Flags",
|
||||
"InputFlags",
|
||||
"UndefinedInputFlags",
|
||||
"InvalidValueInputFlags",
|
||||
"ValidInputFlags",
|
||||
]
|
||||
|
||||
|
||||
from argenta.command.flag.flags.models import (
|
||||
Flags,
|
||||
InputFlags,
|
||||
UndefinedInputFlags,
|
||||
InvalidValueInputFlags,
|
||||
ValidInputFlags,
|
||||
)
|
||||
@@ -2,8 +2,7 @@ from argenta.command.flag.models import InputFlag, Flag
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
|
||||
|
||||
FlagType = TypeVar('FlagType')
|
||||
FlagType = TypeVar("FlagType")
|
||||
|
||||
|
||||
class BaseFlags(Generic[FlagType]):
|
||||
@@ -71,17 +70,21 @@ class BaseFlags(Generic[FlagType]):
|
||||
return True
|
||||
|
||||
|
||||
class Flags(BaseFlags[Flag]): pass
|
||||
class Flags(BaseFlags[Flag]):
|
||||
pass
|
||||
|
||||
|
||||
class InputFlags(BaseFlags[InputFlag]): pass
|
||||
class InputFlags(BaseFlags[InputFlag]):
|
||||
pass
|
||||
|
||||
|
||||
class ValidInputFlags(InputFlags): pass
|
||||
class ValidInputFlags(InputFlags):
|
||||
pass
|
||||
|
||||
|
||||
class UndefinedInputFlags(InputFlags): pass
|
||||
class UndefinedInputFlags(InputFlags):
|
||||
pass
|
||||
|
||||
|
||||
class InvalidValueInputFlags(InputFlags): pass
|
||||
|
||||
class InvalidValueInputFlags(InputFlags):
|
||||
pass
|
||||
@@ -1,10 +1,18 @@
|
||||
from enum import Enum
|
||||
from typing import Literal, Pattern
|
||||
|
||||
|
||||
|
||||
class PossibleValues(Enum):
|
||||
DISABLE: Literal[False] = False
|
||||
ALL: Literal[True] = True
|
||||
|
||||
def __eq__(self, other: bool) -> bool:
|
||||
return self.value == other
|
||||
|
||||
|
||||
class BaseFlag:
|
||||
def __init__(self, name: str,
|
||||
prefix: Literal['-', '--', '---'] = '--') -> None:
|
||||
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--") -> None:
|
||||
"""
|
||||
Private. Base class for flags
|
||||
:param name: the name of the flag
|
||||
@@ -41,9 +49,12 @@ class BaseFlag:
|
||||
|
||||
|
||||
class Flag(BaseFlag):
|
||||
def __init__(self, name: str,
|
||||
prefix: Literal['-', '--', '---'] = '--',
|
||||
possible_values: list[str] | Pattern[str] | False = True) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
prefix: Literal["-", "--", "---"] = "--",
|
||||
possible_values: list[str] | Pattern[str] | PossibleValues = PossibleValues.ALL,
|
||||
) -> None:
|
||||
"""
|
||||
Public. The entity of the flag being registered for subsequent processing
|
||||
:param name: The name of the flag
|
||||
@@ -60,7 +71,7 @@ class Flag(BaseFlag):
|
||||
:param input_flag_value: The input flag value to validate
|
||||
:return: whether the entered flag is valid as bool
|
||||
"""
|
||||
if self.possible_values is False:
|
||||
if self.possible_values == PossibleValues.DISABLE:
|
||||
if input_flag_value is None:
|
||||
return True
|
||||
else:
|
||||
@@ -85,9 +96,12 @@ class Flag(BaseFlag):
|
||||
|
||||
|
||||
class InputFlag(BaseFlag):
|
||||
def __init__(self, name: str,
|
||||
prefix: Literal['-', '--', '---'] = '--',
|
||||
value: str = None):
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
prefix: Literal["-", "--", "---"] = "--",
|
||||
value: str | None = None,
|
||||
):
|
||||
"""
|
||||
Public. The entity of the flag of the entered command
|
||||
:param name: the name of the input flag
|
||||
@@ -114,5 +128,7 @@ class InputFlag(BaseFlag):
|
||||
self._flag_value = value
|
||||
|
||||
def __eq__(self, other) -> bool:
|
||||
return self.get_string_entity() == other.get_string_entity() and self.get_value() == other.get_value()
|
||||
|
||||
return (
|
||||
self.get_string_entity() == other.get_string_entity()
|
||||
and self.get_value() == other.get_value()
|
||||
)
|
||||
|
||||
@@ -1,10 +0,0 @@
|
||||
__all__ = ["Flags", "InputFlags",
|
||||
"UndefinedInputFlags",
|
||||
"InvalidValueInputFlags",
|
||||
"ValidInputFlags"]
|
||||
|
||||
|
||||
from argenta.command.flags.models import (Flags, InputFlags,
|
||||
UndefinedInputFlags,
|
||||
InvalidValueInputFlags,
|
||||
ValidInputFlags)
|
||||
@@ -1,12 +1,11 @@
|
||||
from argenta.command.flag.models import Flag, InputFlag
|
||||
from argenta.command.flags.models import InputFlags, Flags
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
from argenta.command.flag.flags.models import InputFlags, Flags
|
||||
from argenta.command.exceptions import (
|
||||
UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
from typing import Generic, TypeVar, cast, Literal
|
||||
|
||||
|
||||
InputCommandType = TypeVar('InputCommandType')
|
||||
EmptyInputCommandException,
|
||||
)
|
||||
from typing import cast, Literal
|
||||
|
||||
|
||||
class BaseCommand:
|
||||
@@ -26,10 +25,13 @@ class BaseCommand:
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
def __init__(self, trigger: str,
|
||||
description: str = None,
|
||||
flags: Flag | Flags = None,
|
||||
aliases: list[str] = None):
|
||||
def __init__(
|
||||
self,
|
||||
trigger: str,
|
||||
description: str | None = None,
|
||||
flags: Flag | Flags | None = None,
|
||||
aliases: list[str] | None = None,
|
||||
):
|
||||
"""
|
||||
Public. The command that can and should be registered in the Router
|
||||
:param trigger: A string trigger, which, when entered by the user, indicates that the input corresponds to the command
|
||||
@@ -38,8 +40,14 @@ class Command(BaseCommand):
|
||||
:param aliases: string synonyms for the main trigger
|
||||
"""
|
||||
super().__init__(trigger)
|
||||
self._registered_flags: Flags = flags if isinstance(flags, Flags) else Flags(flags) if isinstance(flags, Flag) else Flags()
|
||||
self._description = f'Very useful command' if not description else description
|
||||
self._registered_flags: Flags = (
|
||||
flags
|
||||
if isinstance(flags, Flags)
|
||||
else Flags(flags)
|
||||
if isinstance(flags, Flag)
|
||||
else Flags()
|
||||
)
|
||||
self._description = "Very useful command" if not description else description
|
||||
self._aliases = aliases if isinstance(aliases, list) else []
|
||||
|
||||
def get_registered_flags(self) -> Flags:
|
||||
@@ -56,7 +64,9 @@ class Command(BaseCommand):
|
||||
"""
|
||||
return self._aliases
|
||||
|
||||
def validate_input_flag(self, flag: InputFlag) -> Literal['Undefined', 'Valid', 'Invalid']:
|
||||
def validate_input_flag(
|
||||
self, flag: InputFlag
|
||||
) -> Literal["Undefined", "Valid", "Invalid"]:
|
||||
"""
|
||||
Private. Validates the input flag
|
||||
:param flag: input flag for validation
|
||||
@@ -66,23 +76,28 @@ class Command(BaseCommand):
|
||||
if registered_flags:
|
||||
if isinstance(registered_flags, Flag):
|
||||
if registered_flags.get_string_entity() == flag.get_string_entity():
|
||||
is_valid = registered_flags.validate_input_flag_value(flag.get_value())
|
||||
is_valid = registered_flags.validate_input_flag_value(
|
||||
flag.get_value()
|
||||
)
|
||||
if is_valid:
|
||||
return 'Valid'
|
||||
return "Valid"
|
||||
else:
|
||||
return 'Invalid'
|
||||
return "Invalid"
|
||||
else:
|
||||
return 'Undefined'
|
||||
return "Undefined"
|
||||
else:
|
||||
for registered_flag in registered_flags:
|
||||
if registered_flag.get_string_entity() == flag.get_string_entity():
|
||||
is_valid = registered_flag.validate_input_flag_value(flag.get_value())
|
||||
is_valid = registered_flag.validate_input_flag_value(
|
||||
flag.get_value()
|
||||
)
|
||||
|
||||
if is_valid:
|
||||
return 'Valid'
|
||||
return "Valid"
|
||||
else:
|
||||
return 'Invalid'
|
||||
return 'Undefined'
|
||||
return 'Undefined'
|
||||
return "Invalid"
|
||||
return "Undefined"
|
||||
return "Undefined"
|
||||
|
||||
def get_description(self) -> str:
|
||||
"""
|
||||
@@ -92,10 +107,8 @@ class Command(BaseCommand):
|
||||
return self._description
|
||||
|
||||
|
||||
|
||||
class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
def __init__(self, trigger: str,
|
||||
input_flags: InputFlag | InputFlags = None):
|
||||
class InputCommand(BaseCommand):
|
||||
def __init__(self, trigger: str, input_flags: InputFlag | InputFlags | None = None):
|
||||
"""
|
||||
Private. The model of the input command, after parsing
|
||||
:param trigger:the trigger of the command
|
||||
@@ -103,7 +116,13 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
:return: None
|
||||
"""
|
||||
super().__init__(trigger)
|
||||
self._input_flags: InputFlags = input_flags if isinstance(input_flags, InputFlags) else InputFlags(input_flags) if isinstance(input_flags, InputFlag) else InputFlags()
|
||||
self._input_flags: InputFlags = (
|
||||
input_flags
|
||||
if isinstance(input_flags, InputFlags)
|
||||
else InputFlags(input_flags)
|
||||
if isinstance(input_flags, InputFlag)
|
||||
else InputFlags()
|
||||
)
|
||||
|
||||
def _set_input_flags(self, input_flags: InputFlags) -> None:
|
||||
"""
|
||||
@@ -120,9 +139,8 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
"""
|
||||
return self._input_flags
|
||||
|
||||
|
||||
@staticmethod
|
||||
def parse(raw_command: str) -> InputCommandType:
|
||||
def parse(raw_command: str) -> "InputCommand":
|
||||
"""
|
||||
Private. Parse the raw input command
|
||||
:param raw_command: raw input command
|
||||
@@ -138,8 +156,8 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
current_flag_name, current_flag_value = None, None
|
||||
|
||||
for k, _ in enumerate(list_of_tokens):
|
||||
if _.startswith('-'):
|
||||
if len(_) < 2 or len(_[:_.rfind('-')]) > 3:
|
||||
if _.startswith("-"):
|
||||
if len(_) < 2 or len(_[: _.rfind("-")]) > 3:
|
||||
raise UnprocessedInputFlagException()
|
||||
current_flag_name = _
|
||||
else:
|
||||
@@ -148,16 +166,22 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
current_flag_value = _
|
||||
|
||||
if current_flag_name:
|
||||
if not len(list_of_tokens) == k+1:
|
||||
if not list_of_tokens[k+1].startswith('-'):
|
||||
if not len(list_of_tokens) == k + 1:
|
||||
if not list_of_tokens[k + 1].startswith("-"):
|
||||
continue
|
||||
|
||||
input_flag = InputFlag(name=current_flag_name[current_flag_name.rfind('-') + 1:],
|
||||
prefix=cast(Literal['-', '--', '---'],
|
||||
current_flag_name[:current_flag_name.rfind('-')+1]),
|
||||
value=current_flag_value)
|
||||
input_flag = InputFlag(
|
||||
name=current_flag_name[current_flag_name.rfind("-") + 1 :],
|
||||
prefix=cast(
|
||||
Literal["-", "--", "---"],
|
||||
current_flag_name[: current_flag_name.rfind("-") + 1],
|
||||
),
|
||||
value=current_flag_value,
|
||||
)
|
||||
|
||||
all_flags = [flag.get_string_entity() for flag in input_flags.get_flags()]
|
||||
all_flags = [
|
||||
flag.get_string_entity() for flag in input_flags.get_flags()
|
||||
]
|
||||
if input_flag.get_string_entity() not in all_flags:
|
||||
input_flags.add_flag(input_flag)
|
||||
else:
|
||||
@@ -169,4 +193,3 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
raise UnprocessedInputFlagException()
|
||||
else:
|
||||
return InputCommand(trigger=command, input_flags=input_flags)
|
||||
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
__all__ = ["get_time_of_pre_cycle_setup"]
|
||||
|
||||
|
||||
from argenta.metrics.main import get_time_of_pre_cycle_setup
|
||||
@@ -0,0 +1,18 @@
|
||||
import io
|
||||
from contextlib import redirect_stdout
|
||||
from time import time
|
||||
|
||||
from argenta.app import App
|
||||
|
||||
|
||||
def get_time_of_pre_cycle_setup(app: App) -> float:
|
||||
"""
|
||||
Public. Return time of pre cycle setup
|
||||
:param app: app instance for testing time of pre cycle setup
|
||||
:return: time of pre cycle setup as float
|
||||
"""
|
||||
start = time()
|
||||
with redirect_stdout(io.StringIO()):
|
||||
app.pre_cycle_setup()
|
||||
end = time()
|
||||
return end - start
|
||||
@@ -1,6 +1,8 @@
|
||||
__all__ = ["BooleanArgument", "PositionalArgument", "OptionalArgument"]
|
||||
|
||||
|
||||
from argenta.orchestrator.argparser.arguments.models import (BooleanArgument,
|
||||
from argenta.orchestrator.argparser.arguments.models import (
|
||||
BooleanArgument,
|
||||
PositionalArgument,
|
||||
OptionalArgument)
|
||||
OptionalArgument,
|
||||
)
|
||||
|
||||
@@ -6,6 +6,7 @@ class BaseArgument(ABC):
|
||||
"""
|
||||
Private. Base class for all arguments
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get_string_entity(self) -> str:
|
||||
"""
|
||||
@@ -28,7 +29,7 @@ class PositionalArgument(BaseArgument):
|
||||
|
||||
|
||||
class OptionalArgument(BaseArgument):
|
||||
def __init__(self, name: str, prefix: Literal['-', '--', '---'] = '--'):
|
||||
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--"):
|
||||
"""
|
||||
Public. Optional argument, must have the value
|
||||
:param name: name of the argument
|
||||
@@ -42,7 +43,7 @@ class OptionalArgument(BaseArgument):
|
||||
|
||||
|
||||
class BooleanArgument(BaseArgument):
|
||||
def __init__(self, name: str, prefix: Literal['-', '--', '---'] = '--'):
|
||||
def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--"):
|
||||
"""
|
||||
Public. Boolean argument, does not require a value
|
||||
:param name: name of the argument
|
||||
|
||||
@@ -1,16 +1,20 @@
|
||||
from argparse import ArgumentParser
|
||||
|
||||
from argenta.orchestrator.argparser.arguments.models import (BooleanArgument,
|
||||
from argenta.orchestrator.argparser.arguments.models import (
|
||||
BooleanArgument,
|
||||
OptionalArgument,
|
||||
PositionalArgument)
|
||||
PositionalArgument,
|
||||
)
|
||||
|
||||
|
||||
class ArgParser:
|
||||
def __init__(self,
|
||||
def __init__(
|
||||
self,
|
||||
processed_args: list[PositionalArgument | OptionalArgument | BooleanArgument],
|
||||
name: str = 'Argenta',
|
||||
description: str = 'Argenta available arguments',
|
||||
epilog: str = 'github.com/koloideal/Argenta | made by kolo') -> None:
|
||||
name: str = "Argenta",
|
||||
description: str = "Argenta available arguments",
|
||||
epilog: str = "github.com/koloideal/Argenta | made by kolo",
|
||||
) -> None:
|
||||
"""
|
||||
Public. Cmd argument parser and configurator at startup
|
||||
:param name: the name of the ArgParse instance
|
||||
@@ -22,10 +26,16 @@ class ArgParser:
|
||||
self.description = description
|
||||
self.epilog = epilog
|
||||
|
||||
self.entity: ArgumentParser = ArgumentParser(prog=name, description=description, epilog=epilog)
|
||||
self.args: list[PositionalArgument | OptionalArgument | BooleanArgument] | None = processed_args
|
||||
self.entity: ArgumentParser = ArgumentParser(
|
||||
prog=name, description=description, epilog=epilog
|
||||
)
|
||||
self.args: (
|
||||
list[PositionalArgument | OptionalArgument | BooleanArgument] | None
|
||||
) = processed_args
|
||||
|
||||
def set_args(self, *args: PositionalArgument | OptionalArgument | BooleanArgument) -> None:
|
||||
def set_args(
|
||||
self, *args: PositionalArgument | OptionalArgument | BooleanArgument
|
||||
) -> None:
|
||||
"""
|
||||
Public. Sets the arguments to be processed
|
||||
:param args: processed arguments
|
||||
@@ -46,4 +56,4 @@ class ArgParser:
|
||||
elif type(arg) is OptionalArgument:
|
||||
self.entity.add_argument(arg.get_string_entity())
|
||||
elif type(arg) is BooleanArgument:
|
||||
self.entity.add_argument(arg.get_string_entity(), action='store_true')
|
||||
self.entity.add_argument(arg.get_string_entity(), action="store_true")
|
||||
|
||||
@@ -5,13 +5,13 @@ from argenta.orchestrator.argparser import ArgParser
|
||||
|
||||
|
||||
class Orchestrator:
|
||||
def __init__(self, arg_parser: ArgParser = False):
|
||||
def __init__(self, arg_parser: ArgParser | None = None):
|
||||
"""
|
||||
Public. An orchestrator and configurator that defines the behavior of an integrated system, one level higher than the App
|
||||
:param arg_parser: Cmd argument parser and configurator at startup
|
||||
:return: None
|
||||
"""
|
||||
self.arg_parser: ArgParser | False = arg_parser
|
||||
self.arg_parser: ArgParser | None = arg_parser
|
||||
if arg_parser:
|
||||
self.arg_parser.register_args()
|
||||
|
||||
@@ -33,4 +33,3 @@ class Orchestrator:
|
||||
return self.arg_parser.entity.parse_args()
|
||||
else:
|
||||
return None
|
||||
|
||||
@@ -1,19 +1,21 @@
|
||||
from argenta.response.status import Status
|
||||
from argenta.command.flags import (ValidInputFlags,
|
||||
from argenta.command.flag.flags import (
|
||||
ValidInputFlags,
|
||||
UndefinedInputFlags,
|
||||
InvalidValueInputFlags)
|
||||
InvalidValueInputFlags,
|
||||
)
|
||||
|
||||
|
||||
class Response:
|
||||
__slots__ = ('status',
|
||||
'valid_flags',
|
||||
'undefined_flags',
|
||||
'invalid_value_flags')
|
||||
__slots__ = ("status", "valid_flags", "undefined_flags", "invalid_value_flags")
|
||||
|
||||
def __init__(self, status: Status = None,
|
||||
def __init__(
|
||||
self,
|
||||
status: Status | None = None,
|
||||
valid_flags: ValidInputFlags = ValidInputFlags(),
|
||||
undefined_flags: UndefinedInputFlags = UndefinedInputFlags(),
|
||||
invalid_value_flags: InvalidValueInputFlags = InvalidValueInputFlags()):
|
||||
invalid_value_flags: InvalidValueInputFlags = InvalidValueInputFlags(),
|
||||
):
|
||||
"""
|
||||
Public. The entity of the user input sent to the handler
|
||||
:param status: the status of the response
|
||||
|
||||
@@ -2,8 +2,7 @@ from enum import Enum
|
||||
|
||||
|
||||
class Status(Enum):
|
||||
ALL_FLAGS_VALID = 'ALL_FLAGS_VALID'
|
||||
UNDEFINED_FLAGS = 'UNDEFINED_FLAGS'
|
||||
INVALID_VALUE_FLAGS = 'INVALID_VALUE_FLAGS'
|
||||
UNDEFINED_AND_INVALID_FLAGS = 'UNDEFINED_AND_INVALID_FLAGS'
|
||||
|
||||
ALL_FLAGS_VALID = "ALL_FLAGS_VALID"
|
||||
UNDEFINED_FLAGS = "UNDEFINED_FLAGS"
|
||||
INVALID_VALUE_FLAGS = "INVALID_VALUE_FLAGS"
|
||||
UNDEFINED_AND_INVALID_FLAGS = "UNDEFINED_AND_INVALID_FLAGS"
|
||||
|
||||
@@ -38,7 +38,7 @@ class CommandHandler:
|
||||
|
||||
|
||||
class CommandHandlers:
|
||||
def __init__(self, command_handlers: list[CommandHandler] = None):
|
||||
def __init__(self, command_handlers: list[CommandHandler] | None = None):
|
||||
"""
|
||||
Private. The model that unites all CommandHandler of the routers
|
||||
:param command_handlers: list of CommandHandlers for register
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
system_router = Router(title='System points:')
|
||||
system_router = Router(title="System points:")
|
||||
|
||||
@@ -6,50 +6,64 @@ from argenta.command import Command
|
||||
from argenta.command.models import InputCommand
|
||||
from argenta.response import Response, Status
|
||||
from argenta.router.command_handler.entity import CommandHandlers, CommandHandler
|
||||
from argenta.command.flags.models import (Flags, InputFlags,
|
||||
from argenta.command.flag.flags import (
|
||||
Flags,
|
||||
InputFlags,
|
||||
UndefinedInputFlags,
|
||||
ValidInputFlags,
|
||||
InvalidValueInputFlags)
|
||||
from argenta.router.exceptions import (RepeatedFlagNameException,
|
||||
InvalidValueInputFlags,
|
||||
)
|
||||
from argenta.router.exceptions import (
|
||||
RepeatedFlagNameException,
|
||||
TooManyTransferredArgsException,
|
||||
RequiredArgumentNotPassedException,
|
||||
TriggerContainSpacesException)
|
||||
TriggerContainSpacesException,
|
||||
)
|
||||
|
||||
|
||||
class Router:
|
||||
def __init__(self, title: str = None):
|
||||
def __init__(
|
||||
self, title: str | None = "Awesome title", disable_redirect_stdout: bool = False
|
||||
):
|
||||
"""
|
||||
Public. Directly configures and manages handlers
|
||||
:param title: the title of the router, displayed when displaying the available commands
|
||||
:param disable_redirect_stdout: Disables stdout forwarding, if the argument value is True,
|
||||
the StaticDividingLine will be forced to be used as a line separator for this router,
|
||||
disabled forwarding is needed when there is text output in conjunction with a text input request (for example, input()),
|
||||
if the argument value is True, the output of the input() prompt is intercepted and not displayed,
|
||||
which is ambiguous behavior and can lead to unexpected work
|
||||
:return: None
|
||||
"""
|
||||
self._title = title
|
||||
self.title = title
|
||||
self.disable_redirect_stdout = disable_redirect_stdout
|
||||
|
||||
self._command_handlers: CommandHandlers = CommandHandlers()
|
||||
self._ignore_command_register: bool = False
|
||||
|
||||
|
||||
def command(self, command: Command | str) -> Callable:
|
||||
"""
|
||||
Public. Registers handler
|
||||
:param command: Registered command
|
||||
:return: decorated handler as Callable
|
||||
"""
|
||||
self._validate_command(command)
|
||||
if isinstance(command, str):
|
||||
command = Command(command)
|
||||
redefined_command = Command(command)
|
||||
else:
|
||||
redefined_command = command
|
||||
self._validate_command(redefined_command)
|
||||
|
||||
def command_decorator(func):
|
||||
Router._validate_func_args(func)
|
||||
self._command_handlers.add_handler(CommandHandler(func, command))
|
||||
self._command_handlers.add_handler(CommandHandler(func, redefined_command))
|
||||
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return command_decorator
|
||||
|
||||
|
||||
def finds_appropriate_handler(self, input_command: InputCommand) -> None:
|
||||
"""
|
||||
Private. Finds the appropriate handler for given input command and passes control to it
|
||||
@@ -66,8 +80,9 @@ class Router:
|
||||
if input_command_name.lower() in handle_command.get_aliases():
|
||||
self.process_input_command(input_command_flags, command_handler)
|
||||
|
||||
|
||||
def process_input_command(self, input_command_flags: InputFlags, command_handler: CommandHandler) -> None:
|
||||
def process_input_command(
|
||||
self, input_command_flags: InputFlags, command_handler: CommandHandler
|
||||
) -> None:
|
||||
"""
|
||||
Private. Processes input command with the appropriate handler
|
||||
:param input_command_flags: input command flags as InputFlags
|
||||
@@ -78,7 +93,7 @@ class Router:
|
||||
response: Response = Response()
|
||||
if handle_command.get_registered_flags().get_flags():
|
||||
if input_command_flags.get_flags():
|
||||
response: Response = self._structuring_input_flags(handle_command, input_command_flags)
|
||||
response: Response = self._structuring_input_flags( handle_command, input_command_flags )
|
||||
command_handler.handling(response)
|
||||
else:
|
||||
response.status = Status.ALL_FLAGS_VALID
|
||||
@@ -93,9 +108,10 @@ class Router:
|
||||
response.status = Status.ALL_FLAGS_VALID
|
||||
command_handler.handling(response)
|
||||
|
||||
|
||||
@staticmethod
|
||||
def _structuring_input_flags(handled_command: Command, input_flags: InputFlags) -> Response:
|
||||
def _structuring_input_flags(
|
||||
handled_command: Command, input_flags: InputFlags
|
||||
) -> Response:
|
||||
"""
|
||||
Private. Validates flags of input command
|
||||
:param handled_command: entity of the handled command
|
||||
@@ -106,51 +122,56 @@ class Router:
|
||||
invalid_value_input_flags: InvalidValueInputFlags = InvalidValueInputFlags()
|
||||
undefined_input_flags: UndefinedInputFlags = UndefinedInputFlags()
|
||||
for flag in input_flags:
|
||||
flag_status: Literal['Undefined', 'Valid', 'Invalid'] = handled_command.validate_input_flag(flag)
|
||||
match flag_status:
|
||||
case 'Valid':
|
||||
flag_status: Literal["Undefined", "Valid", "Invalid"] = (
|
||||
handled_command.validate_input_flag(flag)
|
||||
)
|
||||
if flag_status == "Valid":
|
||||
valid_input_flags.add_flag(flag)
|
||||
case 'Undefined':
|
||||
elif flag_status == "Undefined":
|
||||
undefined_input_flags.add_flag(flag)
|
||||
case 'Invalid':
|
||||
elif flag_status == "Invalid":
|
||||
invalid_value_input_flags.add_flag(flag)
|
||||
|
||||
if not invalid_value_input_flags.get_flags() and not undefined_input_flags.get_flags():
|
||||
if (
|
||||
not invalid_value_input_flags.get_flags()
|
||||
and not undefined_input_flags.get_flags()
|
||||
):
|
||||
status = Status.ALL_FLAGS_VALID
|
||||
elif invalid_value_input_flags.get_flags() and not undefined_input_flags.get_flags():
|
||||
elif (
|
||||
invalid_value_input_flags.get_flags()
|
||||
and not undefined_input_flags.get_flags()
|
||||
):
|
||||
status = Status.INVALID_VALUE_FLAGS
|
||||
elif not invalid_value_input_flags.get_flags() and undefined_input_flags.get_flags():
|
||||
elif (
|
||||
not invalid_value_input_flags.get_flags()
|
||||
and undefined_input_flags.get_flags()
|
||||
):
|
||||
status = Status.UNDEFINED_FLAGS
|
||||
else:
|
||||
status = Status.UNDEFINED_AND_INVALID_FLAGS
|
||||
|
||||
return Response(invalid_value_flags=invalid_value_input_flags,
|
||||
return Response(
|
||||
invalid_value_flags=invalid_value_input_flags,
|
||||
valid_flags=valid_input_flags,
|
||||
status=status,
|
||||
undefined_flags=undefined_input_flags)
|
||||
|
||||
undefined_flags=undefined_input_flags,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _validate_command(command: Command | str) -> None:
|
||||
def _validate_command(command: Command) -> None:
|
||||
"""
|
||||
Private. Validates the command registered in handler
|
||||
:param command: validated command
|
||||
:return: None if command is valid else raise exception
|
||||
"""
|
||||
match type(command).__name__:
|
||||
case 'Command':
|
||||
command_name: str = command.get_trigger()
|
||||
if command_name.find(' ') != -1:
|
||||
if command_name.find(" ") != -1:
|
||||
raise TriggerContainSpacesException()
|
||||
flags: Flags = command.get_registered_flags()
|
||||
if flags:
|
||||
flags_name: list = [x.get_string_entity().lower() for x in flags]
|
||||
if len(set(flags_name)) < len(flags_name):
|
||||
raise RepeatedFlagNameException()
|
||||
case 'str':
|
||||
if command.find(' ') != -1:
|
||||
raise TriggerContainSpacesException()
|
||||
|
||||
|
||||
@staticmethod
|
||||
def _validate_func_args(func: Callable) -> None:
|
||||
@@ -172,14 +193,15 @@ class Router:
|
||||
if arg_annotation is Response:
|
||||
pass
|
||||
else:
|
||||
file_path: str = getsourcefile(func)
|
||||
source_line: int = getsourcelines(func)[1]+1
|
||||
file_path: str | None = getsourcefile(func)
|
||||
source_line: int = getsourcelines(func)[1]
|
||||
fprint = Console().print
|
||||
fprint(f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
|
||||
f'of argument([green]{transferred_arg}[/green]) passed to the handler is [/i][bold blue]{Response}[/bold blue],'
|
||||
f' [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]\n', highlight=False)
|
||||
|
||||
|
||||
fprint(
|
||||
f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
|
||||
f"of argument([green]{transferred_arg}[/green]) passed to the handler is [/i][bold blue]{Response}[/bold blue],"
|
||||
f" [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]",
|
||||
highlight=False,
|
||||
)
|
||||
|
||||
def set_command_register_ignore(self, _: bool) -> None:
|
||||
"""
|
||||
@@ -189,7 +211,6 @@ class Router:
|
||||
"""
|
||||
self._ignore_command_register = _
|
||||
|
||||
|
||||
def get_triggers(self) -> list[str]:
|
||||
"""
|
||||
Public. Gets registered triggers
|
||||
@@ -200,7 +221,6 @@ class Router:
|
||||
all_triggers.append(command_handler.get_handled_command().get_trigger())
|
||||
return all_triggers
|
||||
|
||||
|
||||
def get_aliases(self) -> list[str]:
|
||||
"""
|
||||
Public. Gets registered aliases
|
||||
@@ -212,27 +232,9 @@ class Router:
|
||||
all_aliases.extend(command_handler.get_handled_command().get_aliases())
|
||||
return all_aliases
|
||||
|
||||
|
||||
def get_command_handlers(self) -> CommandHandlers:
|
||||
"""
|
||||
Private. Gets registered command handlers
|
||||
:return: registered command handlers as CommandHandlers
|
||||
"""
|
||||
return self._command_handlers
|
||||
|
||||
|
||||
def get_title(self) -> str | None:
|
||||
"""
|
||||
Public. Gets title of the router
|
||||
:return: the title of the router as str or None
|
||||
"""
|
||||
return self._title
|
||||
|
||||
|
||||
def set_title(self, title: str) -> None:
|
||||
"""
|
||||
Public. Sets the title of the router
|
||||
:param title: title that will be setted
|
||||
:return: None
|
||||
"""
|
||||
self._title = title
|
||||
|
||||
@@ -2,6 +2,7 @@ class RepeatedFlagNameException(Exception):
|
||||
"""
|
||||
Private. Raised when a repeated flag name is registered
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Repeated registered flag names in register command"
|
||||
|
||||
@@ -10,6 +11,7 @@ class TooManyTransferredArgsException(Exception):
|
||||
"""
|
||||
Private. Raised when too many arguments are passed
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Too many transferred arguments"
|
||||
|
||||
@@ -18,6 +20,7 @@ class RequiredArgumentNotPassedException(Exception):
|
||||
"""
|
||||
Private. Raised when a required argument is not passed
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Required argument not passed"
|
||||
|
||||
@@ -26,5 +29,6 @@ class TriggerContainSpacesException(Exception):
|
||||
"""
|
||||
Private. Raised when there is a space in the trigger being registered
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "Command trigger cannot contain spaces"
|
||||
|
||||
@@ -7,7 +7,7 @@ import re
|
||||
from argenta.app import App
|
||||
from argenta.command import Command
|
||||
from argenta.router import Router
|
||||
from argenta.command.flags.models import Flags
|
||||
from argenta.command.flag.flags.models import Flags
|
||||
from argenta.command.flag.defaults import PredefinedFlags
|
||||
from argenta.orchestrator import Orchestrator
|
||||
from argenta.response import Response
|
||||
@@ -180,7 +180,7 @@ class TestSystemHandlerNormalWork(TestCase):
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_invalid_input_flags_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
|
||||
app.set_incorrect_input_syntax_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -10,7 +10,7 @@ from argenta.response import Response
|
||||
from argenta.router import Router
|
||||
from argenta.orchestrator import Orchestrator
|
||||
from argenta.command.flag import Flag
|
||||
from argenta.command.flags import Flags
|
||||
from argenta.command.flag.flags import Flags
|
||||
from argenta.command.flag.defaults import PredefinedFlags
|
||||
|
||||
|
||||
|
||||
@@ -3,6 +3,8 @@ from argenta.app import App
|
||||
|
||||
import unittest
|
||||
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
class MyTestCase(unittest.TestCase):
|
||||
def test_is_exit_command1(self):
|
||||
@@ -33,25 +35,25 @@ class MyTestCase(unittest.TestCase):
|
||||
def test_is_unknown_command1(self):
|
||||
app = App()
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._all_registered_triggers_in_lower = ['fr', 'tr', 'de']
|
||||
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
|
||||
self.assertEqual(app._is_unknown_command(InputCommand('fr')), False)
|
||||
|
||||
def test_is_unknown_command2(self):
|
||||
app = App()
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._all_registered_triggers_in_lower = ['fr', 'tr', 'de']
|
||||
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
|
||||
self.assertEqual(app._is_unknown_command(InputCommand('cr')), True)
|
||||
|
||||
def test_is_unknown_command3(self):
|
||||
app = App(ignore_command_register=False)
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._all_registered_triggers_in_default_case = ['Pr', 'tW', 'deQW']
|
||||
app._current_matching_triggers_with_routers = {'Pr': Router(), 'tW': Router(), 'deQW': Router()}
|
||||
self.assertEqual(app._is_unknown_command(InputCommand('pr')), True)
|
||||
|
||||
def test_is_unknown_command4(self):
|
||||
app = App(ignore_command_register=False)
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._all_registered_triggers_in_default_case = ['Pr', 'tW', 'deQW']
|
||||
app._current_matching_triggers_with_routers = {'Pr': Router(), 'tW': Router(), 'deQW': Router()}
|
||||
self.assertEqual(app._is_unknown_command(InputCommand('tW')), False)
|
||||
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from argenta.command.flag import Flag, InputFlag
|
||||
from argenta.command.flags import Flags
|
||||
from argenta.command.flag.flags import Flags
|
||||
from argenta.command.models import InputCommand, Command
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from argenta.command.flag import Flag, InputFlag
|
||||
from argenta.command.flags import InputFlags, Flags
|
||||
from argenta.command.flag import Flag, InputFlag, PossibleValues
|
||||
from argenta.command.flag.flags import InputFlags, Flags
|
||||
|
||||
import unittest
|
||||
import re
|
||||
@@ -54,19 +54,19 @@ class TestFlag(unittest.TestCase):
|
||||
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
|
||||
|
||||
def test_validate_correct_empty_flag_value_without_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=False)
|
||||
flag = Flag(name='test', possible_values=PossibleValues.DISABLE)
|
||||
self.assertEqual(flag.validate_input_flag_value(None), True)
|
||||
|
||||
def test_validate_correct_empty_flag_value_with_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=True)
|
||||
flag = Flag(name='test', possible_values=PossibleValues.DISABLE)
|
||||
self.assertEqual(flag.validate_input_flag_value(None), True)
|
||||
|
||||
def test_validate_incorrect_random_flag_value_without_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=False)
|
||||
flag = Flag(name='test', possible_values=PossibleValues.DISABLE)
|
||||
self.assertEqual(flag.validate_input_flag_value('random value'), False)
|
||||
|
||||
def test_validate_correct_random_flag_value_with_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=True)
|
||||
flag = Flag(name='test', possible_values=PossibleValues.ALL)
|
||||
self.assertEqual(flag.validate_input_flag_value('random value'), True)
|
||||
|
||||
def test_get_input_flag1(self):
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
from argenta.command.flag import InputFlag, Flag
|
||||
from argenta.command.flags import Flags, InputFlags, UndefinedInputFlags, InvalidValueInputFlags, ValidInputFlags
|
||||
from argenta.response import Response
|
||||
from argenta.command.flag.flags import Flags, InputFlags, UndefinedInputFlags, InvalidValueInputFlags, ValidInputFlags
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
from argenta.router.exceptions import (TriggerContainSpacesException,
|
||||
@@ -13,9 +12,6 @@ import re
|
||||
|
||||
|
||||
class TestRouter(unittest.TestCase):
|
||||
def test_get_router_title(self):
|
||||
self.assertEqual(Router(title='test title').get_title(), 'test title')
|
||||
|
||||
def test_register_command_with_spaces_in_trigger(self):
|
||||
router = Router()
|
||||
with self.assertRaises(TriggerContainSpacesException):
|
||||
|
||||
Reference in New Issue
Block a user