mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 10:05:28 +03:00
Update documentation
This commit is contained in:
@@ -1,13 +1,17 @@
|
||||
from argenta import App, Orchestrator
|
||||
from argenta.orchestrator.argparser import ArgParser, BooleanArgument, ValueArgument
|
||||
from argenta.orchestrator.argparser import ArgParser, BooleanArgument
|
||||
|
||||
arg_parser = ArgParser(processed_args=[BooleanArgument("dev"), ValueArgument('some', possible_values=['fuck', 'cruck'])])
|
||||
arg_parser = ArgParser(
|
||||
processed_args=[
|
||||
BooleanArgument("dev")
|
||||
]
|
||||
)
|
||||
orchestrator = Orchestrator(
|
||||
arg_parser=arg_parser,
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
if arg_parser.parsed_argspace.get_by_name('dev'):
|
||||
orchestrator.start_polling(App(initial_message='ArgentaDev'))
|
||||
if arg_parser.parsed_argspace.get_by_name("dev"):
|
||||
orchestrator.start_polling(App(initial_message="ArgentaDev"))
|
||||
else:
|
||||
orchestrator.start_polling(App())
|
||||
|
||||
@@ -6,11 +6,7 @@ arguments = [
|
||||
ValueArgument("port", help="Server port", is_required=True),
|
||||
]
|
||||
|
||||
argparser = ArgParser(
|
||||
processed_args=arguments,
|
||||
name="WebServer",
|
||||
description="Simple web server"
|
||||
)
|
||||
argparser = ArgParser(processed_args=arguments, name="WebServer", description="Simple web server")
|
||||
|
||||
app = App()
|
||||
orchestrator = Orchestrator(argparser)
|
||||
|
||||
@@ -1,11 +1,20 @@
|
||||
config_arg = argspace.get_by_name("config")
|
||||
if config_arg:
|
||||
from argenta import Response, Router
|
||||
from argenta.di import FromDishka
|
||||
from argenta.orchestrator.argparser import ArgSpace
|
||||
|
||||
router = Router()
|
||||
|
||||
|
||||
@router.command("get_args")
|
||||
def get_args(response: Response, argspace: FromDishka[ArgSpace]):
|
||||
config_arg = argspace.get_by_name("config")
|
||||
if config_arg:
|
||||
print(f"Config path: {config_arg.value}")
|
||||
|
||||
verbose_arg = argspace.get_by_name("verbose")
|
||||
if verbose_arg and verbose_arg.value:
|
||||
verbose_arg = argspace.get_by_name("verbose")
|
||||
if verbose_arg and verbose_arg.value:
|
||||
print("Verbose mode enabled")
|
||||
|
||||
unknown_arg = argspace.get_by_name("nonexistent")
|
||||
if unknown_arg is None:
|
||||
unknown_arg = argspace.get_by_name("nonexistent")
|
||||
if unknown_arg is None:
|
||||
print("Argument not found")
|
||||
|
||||
@@ -1,28 +1,20 @@
|
||||
from argenta.orchestrator.argparser import ArgParser, ValueArgument
|
||||
|
||||
# Create arguments
|
||||
config_arg = ValueArgument(
|
||||
"config",
|
||||
help="Path to configuration file",
|
||||
default="config.yaml"
|
||||
)
|
||||
config_arg = ValueArgument("config", help="Path to configuration file", default="config.yaml")
|
||||
|
||||
log_level_arg = ValueArgument(
|
||||
"log-level",
|
||||
help="Logging level",
|
||||
possible_values=["DEBUG", "INFO", "WARNING", "ERROR"],
|
||||
default="INFO"
|
||||
default="INFO",
|
||||
)
|
||||
|
||||
host_arg = ValueArgument(
|
||||
"host",
|
||||
help="Server host address",
|
||||
is_required=True
|
||||
)
|
||||
host_arg = ValueArgument("host", help="Server host address", is_required=True)
|
||||
|
||||
# Register in ArgParser
|
||||
parser = ArgParser(
|
||||
processed_args=[config_arg, log_level_arg, host_arg],
|
||||
name="MyApp",
|
||||
description="My application with CLI arguments"
|
||||
description="My application with CLI arguments",
|
||||
)
|
||||
@@ -1,23 +1,9 @@
|
||||
from argenta.orchestrator.argparser import ArgParser, BooleanArgument
|
||||
|
||||
# Create boolean arguments
|
||||
verbose_arg = BooleanArgument(
|
||||
"verbose",
|
||||
help="Enable verbose output"
|
||||
)
|
||||
|
||||
debug_arg = BooleanArgument(
|
||||
"debug",
|
||||
help="Enable debug mode"
|
||||
)
|
||||
|
||||
no_cache_arg = BooleanArgument(
|
||||
"no-cache",
|
||||
help="Disable caching"
|
||||
)
|
||||
verbose_arg = BooleanArgument("verbose", help="Enable verbose output")
|
||||
debug_arg = BooleanArgument("debug", help="Enable debug mode")
|
||||
no_cache_arg = BooleanArgument("no-cache", help="Disable caching")
|
||||
|
||||
# Register in ArgParser
|
||||
parser = ArgParser(
|
||||
processed_args=[verbose_arg, debug_arg, no_cache_arg],
|
||||
name="MyApp"
|
||||
)
|
||||
parser = ArgParser(processed_args=[verbose_arg, debug_arg, no_cache_arg], name="MyApp")
|
||||
@@ -2,10 +2,13 @@ from argenta import Router, Command, Response
|
||||
|
||||
router = Router(title="System")
|
||||
|
||||
@router.command(Command(
|
||||
|
||||
@router.command(
|
||||
Command(
|
||||
"shutdown",
|
||||
description="Shutdown the system",
|
||||
aliases=["poweroff", "halt", "stop"]
|
||||
))
|
||||
)
|
||||
)
|
||||
def handle_shutdown(response: Response):
|
||||
print("Shutting down the system...")
|
||||
@@ -5,10 +5,7 @@ from argenta.command.flag import ValidationStatus
|
||||
router = Router()
|
||||
|
||||
|
||||
@router.command(Command(
|
||||
"deploy",
|
||||
flags=Flag("verbose", possible_values=PossibleValues.NEITHER)
|
||||
))
|
||||
@router.command(Command("deploy", flags=Flag("verbose", possible_values=PossibleValues.NEITHER)))
|
||||
def deploy_handler(response: Response):
|
||||
# Check for toggle flag presence
|
||||
verbose_flag = response.input_flags.get_flag_by_name("verbose")
|
||||
|
||||
@@ -8,10 +8,7 @@ router = Router(title="Example")
|
||||
Command(
|
||||
"example",
|
||||
description="Example command with flags",
|
||||
flags=Flags([
|
||||
Flag("name"),
|
||||
Flag("age")
|
||||
]),
|
||||
flags=Flags([Flag("name"), Flag("age")]),
|
||||
)
|
||||
)
|
||||
def example_handler(response: Response):
|
||||
|
||||
@@ -8,11 +8,7 @@ router = Router(title="Get Flag Example")
|
||||
Command(
|
||||
"config",
|
||||
description="Configure settings",
|
||||
flags=Flags([
|
||||
Flag("host"),
|
||||
Flag("port"),
|
||||
Flag("debug")
|
||||
]),
|
||||
flags=Flags([Flag("host"), Flag("port"), Flag("debug")]),
|
||||
)
|
||||
)
|
||||
def config_handler(response: Response):
|
||||
|
||||
@@ -5,17 +5,11 @@ from argenta.command import InputFlags
|
||||
flags = InputFlags()
|
||||
|
||||
# Create several flags
|
||||
flag1 = InputFlag(
|
||||
name="option1", prefix="--", input_value="value1", status=ValidationStatus.VALID
|
||||
)
|
||||
flag1 = InputFlag(name="option1", prefix="--", input_value="value1", status=ValidationStatus.VALID)
|
||||
|
||||
flag2 = InputFlag(
|
||||
name="option2", prefix="--", input_value="value2", status=ValidationStatus.VALID
|
||||
)
|
||||
flag2 = InputFlag(name="option2", prefix="--", input_value="value2", status=ValidationStatus.VALID)
|
||||
|
||||
flag3 = InputFlag(
|
||||
name="option3", prefix="---", input_value="value3", status=ValidationStatus.VALID
|
||||
)
|
||||
flag3 = InputFlag(name="option3", prefix="---", input_value="value3", status=ValidationStatus.VALID)
|
||||
|
||||
# Add all flags in one call
|
||||
flags.add_flags([flag1, flag2, flag3])
|
||||
|
||||
@@ -26,12 +26,8 @@ flags3 = InputFlags(
|
||||
)
|
||||
|
||||
print(f"flags1 == flags2: {flags1 == flags2}") # True (same names)
|
||||
print(
|
||||
f"flags1 == flags3: {flags1 == flags3}"
|
||||
) # True (same names, values are not considered)
|
||||
print(f"flags1 == flags3: {flags1 == flags3}") # True (same names, values are not considered)
|
||||
|
||||
# Different collections
|
||||
flags4 = InputFlags(
|
||||
[InputFlag(name="flag3", input_value="value3", status=ValidationStatus.VALID)]
|
||||
)
|
||||
flags4 = InputFlags([InputFlag(name="flag3", input_value="value3", status=ValidationStatus.VALID)])
|
||||
print(f"flags1 == flags4: {flags1 == flags4}") # False (different flags)
|
||||
|
||||
@@ -8,11 +8,8 @@ from argenta.response.status import ResponseStatus
|
||||
|
||||
router = Router("Calculator")
|
||||
|
||||
operations = {
|
||||
'mul': operator.mul,
|
||||
'sub': operator.sub,
|
||||
'add': operator.add
|
||||
}
|
||||
operations = {"mul": operator.mul, "sub": operator.sub, "add": operator.add}
|
||||
|
||||
|
||||
@router.command(
|
||||
Command(
|
||||
@@ -22,7 +19,9 @@ operations = {
|
||||
[
|
||||
Flag("a", possible_values=re.compile(r"^\d{,5}$")), # First number
|
||||
Flag("b", possible_values=re.compile(r"^\d{,5}$")), # Second number
|
||||
Flag("operation", possible_values=["add", "sub", "mul"]), # Operation: add, sub, mul
|
||||
Flag(
|
||||
"operation", possible_values=["add", "sub", "mul"]
|
||||
), # Operation: add, sub, mul
|
||||
]
|
||||
),
|
||||
)
|
||||
|
||||
@@ -6,7 +6,7 @@ app = App(
|
||||
prompt=">> ",
|
||||
initial_message="Simple App",
|
||||
farewell_message="Goodbye!",
|
||||
repeat_command_groups_printing=False
|
||||
repeat_command_groups_printing=False,
|
||||
)
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@@ -15,11 +15,7 @@ main_router = Router(title="Main commands")
|
||||
|
||||
|
||||
# 3. Define command and its handler
|
||||
@main_router.command(Command(
|
||||
"hello",
|
||||
description="Prints greeting message",
|
||||
flags=Flag("name")
|
||||
))
|
||||
@main_router.command(Command("hello", description="Prints greeting message", flags=Flag("name")))
|
||||
def hello_handler(response: Response):
|
||||
"""This handler will be called for 'hello' command."""
|
||||
name = response.input_flags.get_flag_by_name("name")
|
||||
|
||||
@@ -10,14 +10,18 @@ from .repository import Priority, Task, TaskRepository
|
||||
router = Router(title="Task Manager")
|
||||
|
||||
|
||||
@router.command(Command(
|
||||
@router.command(
|
||||
Command(
|
||||
"add-task",
|
||||
description="Add a new task",
|
||||
flags=Flags([
|
||||
flags=Flags(
|
||||
[
|
||||
Flag("description"),
|
||||
Flag("priority", possible_values=["low", "medium", "high"]),
|
||||
]),
|
||||
))
|
||||
]
|
||||
),
|
||||
)
|
||||
)
|
||||
def add_task(response: Response, repo: FromDishka[TaskRepository]):
|
||||
description_flag = response.input_flags.get_flag_by_name("description")
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ from argenta.di import FromDishka
|
||||
|
||||
router = Router(title="Authentication")
|
||||
|
||||
|
||||
def authenticate_user(username: str) -> str:
|
||||
return f"token_for_{username}"
|
||||
|
||||
|
||||
@@ -1,25 +0,0 @@
|
||||
from argenta import Command, Response, Router
|
||||
|
||||
router = Router(title="Data Example")
|
||||
|
||||
|
||||
@router.command(Command("set", description="Set data"))
|
||||
def set_handler(response: Response):
|
||||
# Update global data storage
|
||||
response.update_data(
|
||||
{
|
||||
"user_name": "John",
|
||||
"timestamp": "2024-01-01",
|
||||
"settings": {"theme": "dark", "language": "ru"},
|
||||
}
|
||||
)
|
||||
print("Data updated successfully")
|
||||
|
||||
|
||||
@router.command(Command("show", description="Show data"))
|
||||
def show_handler(response: Response):
|
||||
# Get data from global storage
|
||||
data = response.get_data()
|
||||
if "user_name" in data:
|
||||
print(f"User: {data['user_name']}")
|
||||
print(f"Settings: {data.get('settings', {})}")
|
||||
@@ -1,16 +0,0 @@
|
||||
from argenta import Command, Response, Router
|
||||
|
||||
router = Router(title="Get Data Example")
|
||||
|
||||
|
||||
@router.command(Command("info", description="Show all stored data"))
|
||||
def info_handler(response: Response):
|
||||
# Get all data from global storage
|
||||
all_data = response.get_data()
|
||||
|
||||
if all_data:
|
||||
print("Stored data:")
|
||||
for key, value in all_data.items():
|
||||
print(f" {key}: {value}")
|
||||
else:
|
||||
print("No data stored")
|
||||
@@ -1,19 +0,0 @@
|
||||
from argenta import Command, Response, Router
|
||||
|
||||
router = Router(title="Clear Data Example")
|
||||
|
||||
|
||||
@router.command(Command("clear", description="Clear all stored data"))
|
||||
def clear_handler(response: Response):
|
||||
# Clear all data storage
|
||||
response.clear_data()
|
||||
print("All data cleared")
|
||||
|
||||
|
||||
@router.command(Command("check", description="Check if data exists"))
|
||||
def check_handler(response: Response):
|
||||
data = response.get_data()
|
||||
if data:
|
||||
print(f"Storage contains {len(data)} item(s)")
|
||||
else:
|
||||
print("Storage is empty")
|
||||
@@ -1,29 +0,0 @@
|
||||
from argenta import Command, Response, Router
|
||||
|
||||
router = Router(title="Delete Data Example")
|
||||
|
||||
|
||||
@router.command(Command("store", description="Store data"))
|
||||
def store_handler(response: Response):
|
||||
response.update_data(
|
||||
{
|
||||
"temp_key": "temporary value",
|
||||
"important_key": "important value",
|
||||
"another_key": "another value",
|
||||
}
|
||||
)
|
||||
print("Data stored")
|
||||
|
||||
|
||||
@router.command(Command("remove", description="Remove specific key"))
|
||||
def remove_handler(response: Response):
|
||||
# Delete specific key from storage
|
||||
try:
|
||||
response.delete_from_data("temp_key")
|
||||
print("Key 'temp_key' deleted")
|
||||
|
||||
# Check what remains
|
||||
remaining = response.get_data()
|
||||
print(f"Remaining keys: {list(remaining.keys())}")
|
||||
except KeyError:
|
||||
print("Key not found")
|
||||
@@ -10,10 +10,7 @@ router = Router(title="Flags Example")
|
||||
Command(
|
||||
"process",
|
||||
description="Process with flags",
|
||||
flags=Flags([
|
||||
Flag("format", possible_values=["json", "xml"]),
|
||||
Flag("verbose")
|
||||
]),
|
||||
flags=Flags([Flag("format", possible_values=["json", "xml"]), Flag("verbose")]),
|
||||
)
|
||||
)
|
||||
def process_handler(response: Response):
|
||||
|
||||
@@ -8,22 +8,21 @@ from argenta import App, Orchestrator, Router, Command, Response
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def patched_argv():
|
||||
with patch.object(sys, 'argv', ['program.py']):
|
||||
with patch.object(sys, "argv", ["program.py"]):
|
||||
yield
|
||||
|
||||
|
||||
def test_input_incorrect_command(capsys: CaptureFixture[str]):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
@router.command(Command("test"))
|
||||
def test(response: Response) -> None:
|
||||
print('test command')
|
||||
print("test command")
|
||||
|
||||
app = App(override_system_messages=True, printer=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(
|
||||
lambda command: print(f'Unknown command: {command.trigger}')
|
||||
)
|
||||
app.set_unknown_command_handler(lambda command: print(f"Unknown command: {command.trigger}"))
|
||||
|
||||
with patch("builtins.input", side_effect=["help", "q"]):
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
@@ -12,12 +12,14 @@ class Service:
|
||||
def hello(self) -> str:
|
||||
return "world"
|
||||
|
||||
|
||||
def get_service() -> Service:
|
||||
return Service()
|
||||
|
||||
|
||||
router = Router(title="DI")
|
||||
|
||||
|
||||
@router.command("HELLO")
|
||||
def hello(response: Response, service: FromDishka[Service]) -> None:
|
||||
print(f"hello {service.hello()}")
|
||||
@@ -37,6 +39,6 @@ def test_hello_uses_service():
|
||||
|
||||
# Call handler
|
||||
with redirect_stdout(io.StringIO()) as stdout:
|
||||
router.finds_appropriate_handler(InputCommand.parse('HELLO'))
|
||||
router.finds_appropriate_handler(InputCommand.parse("HELLO"))
|
||||
|
||||
assert "hello world" in stdout.getvalue()
|
||||
|
||||
@@ -7,6 +7,7 @@ from argenta.command import InputCommand
|
||||
|
||||
router = Router(title="Demo")
|
||||
|
||||
|
||||
@router.command(Command("PING", description="Ping command"))
|
||||
def ping(response: Response):
|
||||
print("PONG")
|
||||
|
||||
+13
-19
@@ -13,26 +13,20 @@ App
|
||||
.. code-block:: python
|
||||
:linenos:
|
||||
|
||||
AVAILABLE_DIVIDING_LINES: TypeAlias = StaticDividingLine | DynamicDividingLine
|
||||
DEFAULT_DIVIDING_LINE: StaticDividingLine = StaticDividingLine()
|
||||
|
||||
DEFAULT_PRINT_FUNC: Printer = Console().print
|
||||
DEFAULT_AUTOCOMPLETER: AutoCompleter = AutoCompleter()
|
||||
DEFAULT_EXIT_COMMAND: Command = Command("Q", description="Exit command")
|
||||
|
||||
.. code-block:: python
|
||||
:linenos:
|
||||
|
||||
def __init__(self, *, prompt: str = "What do you want to do?\n\n",
|
||||
initial_message: str = "Argenta\n",
|
||||
farewell_message: str = "\nSee you\n",
|
||||
exit_command: Command = DEFAULT_EXIT_COMMAND,
|
||||
system_router_title: str | None = "System points:",
|
||||
dividing_line: AVAILABLE_DIVIDING_LINES = DEFAULT_DIVIDING_LINE,
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
prompt: str = ">>> ",
|
||||
initial_message: str = "Argenta",
|
||||
farewell_message: str = "See you",
|
||||
exit_command: Command = Command("q", description="Exit command"),
|
||||
system_router_title: str = "System points:",
|
||||
dividing_line: StaticDividingLine | DynamicDividingLine | None = None,
|
||||
repeat_command_groups_printing: bool = False,
|
||||
override_system_messages: bool = False,
|
||||
autocompleter: AutoCompleter = DEFAULT_AUTOCOMPLETER,
|
||||
print_func: Printer = DEFAULT_PRINT_FUNC) -> None
|
||||
autocompleter: AutoCompleter | None = None,
|
||||
printer: Printer = Console().print,
|
||||
) -> None:
|
||||
|
||||
Создаёт и настраивает экземпляр приложения.
|
||||
|
||||
@@ -45,7 +39,7 @@ App
|
||||
* ``repeat_command_groups_printing``: Если ``True``, список доступных команд выводится перед каждым вводом.
|
||||
* ``override_system_messages``: Если ``True``, стандартное форматирование (цвета, ASCII-арт) отключается.
|
||||
* ``autocompleter``: Экземпляр класса :ref:`AutoCompleter <root_api_app_autocompleter>`, отвечающий за автодополнение команд.
|
||||
* ``print_func``: Функция для вывода всех системных сообщений (по умолчанию ``rich.Console().print``).
|
||||
* ``printer``: Функция для вывода всех системных сообщений.
|
||||
|
||||
-----
|
||||
|
||||
|
||||
@@ -84,3 +84,8 @@ ArgParser
|
||||
|
||||
$ python app.py --old-param value
|
||||
Warning: argument --old-param is deprecated
|
||||
|
||||
.. warning::
|
||||
|
||||
Параметр поддерживается начиная с версии CPython 3.13, если версия ниже, то параметр будет игнорироваться.
|
||||
|
||||
|
||||
@@ -29,9 +29,9 @@
|
||||
Кастомизация вывода
|
||||
-------------------
|
||||
|
||||
Для полной замены логики вывода текста в конструкторе ``App`` предусмотрен параметр ``print_func``.
|
||||
Для полной замены логики вывода текста в конструкторе ``App`` предусмотрен параметр ``printer``.
|
||||
|
||||
* **print_func**: ``Callable[[str], None]``
|
||||
* **printer**: ``Callable[[str], None]``
|
||||
Этот параметр позволяет передать любую вызываемую сущность (например, функцию), которая будет использоваться для вывода всех системных сообщений. По умолчанию это ``rich.console.Console().print``. Вы можете передать сюда свою функцию, чтобы, например, логировать вывод в файл или отправлять его по сети.
|
||||
|
||||
.. important::
|
||||
|
||||
Reference in New Issue
Block a user