This commit is contained in:
2025-11-04 11:21:35 +03:00
parent 7b85b0f08d
commit 4da876b774
16 changed files with 122 additions and 48 deletions
@@ -1,9 +1,7 @@
from argenta import App
def empty_command_handler():
print("Empty command handler called")
app: App = App()
app.set_empty_command_handler(empty_command_handler)
@@ -1,9 +1,7 @@
from argenta import App
def incorrect_input_syntax_handler(raw_command: str):
print(f"Incorrect input syntax for command: {raw_command}")
app: App = App()
app.set_incorrect_input_syntax_handler(incorrect_input_syntax_handler)
@@ -1,9 +1,7 @@
from argenta import App
def repeated_input_flags_handler(raw_command: str):
print(f"Repeated input flags: {raw_command}")
app: App = App()
app.set_repeated_input_flags_handler(repeated_input_flags_handler)
@@ -1,9 +1,7 @@
from argenta import App
def empty_command_handler():
print("Empty input command")
app: App = App()
app.set_empty_command_handler(empty_command_handler)
@@ -1,10 +1,8 @@
from argenta import App
from argenta.command import InputCommand
def unknown_command_handler(command: InputCommand):
print(f"Unknown input command with trigger: {command.trigger}")
app: App = App()
app.set_unknown_command_handler(unknown_command_handler)
@@ -1,9 +1,7 @@
from argenta import App, Response
def exit_command_handler(response: Response):
print("Exit command handler")
app: App = App()
app.set_exit_command_handler(exit_command_handler)
+7 -5
View File
@@ -12,12 +12,13 @@ orchestrator = Orchestrator()
# 2. Создание роутера для группировки команд
main_router = Router(title="Основные команды")
# 3. Определение команды и её обработчика
@main_router.command(Command(
"hello",
description="Печатает приветственное сообщение",
flags=Flag("name")
))
@main_router.command(
Command(
"hello", description="Печатает приветственное сообщение", flags=Flag("name")
)
)
def hello_handler(response: Response):
"""Этот обработчик будет вызван для команды 'hello'."""
name = response.input_flags.get_flag_by_name("name")
@@ -26,6 +27,7 @@ def hello_handler(response: Response):
else:
print("Привет, мир!")
# 4. Подключение роутера к приложению
app.include_router(main_router)
@@ -1,8 +1,7 @@
from typing import cast
from argenta import Command, Response, Router
from argenta.command import Flag, Flags
from argenta.command.flag.models import ValidationStatus
from argenta.command.flag import ValidationStatus, Flag, Flags
from argenta.di import FromDishka
from .repository import Priority, Task, TaskRepository
@@ -24,14 +23,12 @@ router = Router(title="Task Manager")
)
def add_task(response: Response, repo: FromDishka[TaskRepository]):
description_flag = response.input_flags.get_flag_by_name("description")
if not description_flag or not description_flag.input_value:
if not description_flag or not description_flag.status == ValidationStatus.VALID:
print("Error: --description flag is required.")
return
task_description = description_flag.input_value
task_description = description_flag.input_value or ""
priority_flag = response.input_flags.get_flag_by_name("priority")
if priority_flag and priority_flag.status == ValidationStatus.VALID:
priority_value = priority_flag.input_value
else:
@@ -3,16 +3,16 @@ from argenta import App, Orchestrator
from .handlers import router
from .provider import TaskProvider
# 1. Создаем экземпляр приложения
# 1. Создаем экземпляр приложения и оркестратора
app = App(
initial_message="Task Manager",
prompt="Enter a command: ",
)
orchestrator = Orchestrator(custom_providers=[TaskProvider()])
# 2. Подключаем роутер с нашими командами
app.include_router(router)
# 3. Создаем и запускаем оркестратор
# 3. Запускаем поллинг через оркестратор
if __name__ == "__main__":
orchestrator = Orchestrator(custom_providers=[TaskProvider()])
orchestrator.start_polling(app)
+15 -8
View File
@@ -3,19 +3,23 @@ from argenta.command import Flag
from argenta.di import FromDishka
# 1. Создаём роутер
router = Router(title='Authentication')
router = Router(title="Authentication")
# 2. Определяем сервис и обработчики
def authenticate_user(username: str) -> str:
"""Возвращает фиктивный токен для пользователя."""
return f"token_for_{username}"
@router.command(Command('login', flags=Flag('username')))
@router.command(Command("login", flags=Flag("username")))
def login_handler(response: Response, data_bridge: FromDishka[DataBridge]):
"""Обработчик для команды 'login'. Сохраняет токен в хранилище."""
username_flag = response.input_flags.get_flag_by_name('username')
username_flag = response.input_flags.get_flag_by_name("username")
if not username_flag or not username_flag.input_value:
print("[red]Ошибка:[/red] необходимо указать имя пользователя с помощью флага --username.")
print(
"[red]Ошибка:[/red] необходимо указать имя пользователя с помощью флага --username."
)
return
username = username_flag.input_value
@@ -25,19 +29,23 @@ def login_handler(response: Response, data_bridge: FromDishka[DataBridge]):
data_bridge.update({"auth_token": token})
print(f"[green]Успешный вход![/green] Пользователь '{username}' аутентифицирован.")
@router.command('get-profile')
@router.command("get-profile")
def get_profile_handler(response: Response, data_bridge: FromDishka[DataBridge]):
"""Обработчик для команды 'get-profile'. Использует токен из хранилища."""
session_data = data_bridge.get_all()
token = session_data.get("auth_token")
if not token:
print("[red]Ошибка:[/red] вы не аутентифицированы. Сначала выполните команду 'login'.")
print(
"[red]Ошибка:[/red] вы не аутентифицированы. Сначала выполните команду 'login'."
)
return
print(f"Загрузка профиля с использованием токена: [yellow]{token}[/yellow]")
@router.command('logout')
@router.command("logout")
def logout_handler(response: Response, data_bridge: FromDishka[DataBridge]):
"""Обработчик для команды 'logout'. Очищает токен."""
try:
@@ -45,4 +53,3 @@ def logout_handler(response: Response, data_bridge: FromDishka[DataBridge]):
print("[green]Выход выполнен.[/green] Данные сессии очищены.")
except KeyError:
print("Вы и так не были аутентифицированы.")
@@ -14,6 +14,7 @@ class TestAppIntegration(unittest.TestCase):
@self.router.command(Command("HELP", description="Show help"))
def help_cmd(response: Response):
print("Available commands: HELP")
_ = help_cmd # appease linter: function is registered via decorator
self.app.include_router(self.router)
@@ -22,5 +23,3 @@ class TestAppIntegration(unittest.TestCase):
with redirect_stdout(io.StringIO()) as stdout:
self.router.finds_appropriate_handler(InputCommand.parse("HELP"))
self.assertIn("Available commands:", stdout.getvalue())
@@ -15,6 +15,7 @@ class Service:
def hello(self) -> str:
return "world"
router = Router(title="DI")
@@ -56,5 +57,3 @@ class TestDIHandler(unittest.TestCase):
class _FakeApp:
# Minimal stub for setup_dishka; app object is not used in unit tests
registered_routers = []
@@ -20,5 +20,3 @@ class TestSimpleHandler(unittest.TestCase):
with redirect_stdout(io.StringIO()) as stdout:
router.finds_appropriate_handler(InputCommand.parse("PING"))
self.assertIn("PONG", stdout.getvalue())