This commit is contained in:
2025-11-02 01:04:00 +03:00
parent 15f97eab61
commit 64c984a704
36 changed files with 170 additions and 42 deletions
@@ -0,0 +1,52 @@
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.di import FromDishka
from .repository import Priority, Task, TaskRepository
router = Router(title="Task Manager")
@router.command(
Command(
"add-task",
description="Add a new task",
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")
if not description_flag or not description_flag.input_value:
print("Error: --description flag is required.")
return
task_description = description_flag.input_value
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:
priority_value = "medium"
priority = cast(Priority, priority_value)
task = Task(description=task_description, priority=priority)
repo.add_task(task)
print(f"Added task: '{task.description}' with priority '{task.priority}'")
@router.command(Command("list-tasks", description="List all tasks"))
def list_tasks(response: Response, repo: FromDishka[TaskRepository]):
tasks = repo.get_all_tasks()
if not tasks:
print("No tasks found.")
return
print("Tasks:")
for i, task in enumerate(tasks, 1):
print(f" {i}. {task.description} (Priority: {task.priority})")
@@ -0,0 +1,18 @@
from argenta import App, Orchestrator
from .handlers import router
from .provider import TaskProvider
# 1. Создаем экземпляр приложения
app = App(
initial_message="Task Manager",
prompt="Enter a command: ",
)
# 2. Подключаем роутер с нашими командами
app.include_router(router)
# 3. Создаем и запускаем оркестратор
if __name__ == "__main__":
orchestrator = Orchestrator(custom_providers=[TaskProvider()])
orchestrator.start_polling(app)
@@ -0,0 +1,9 @@
from dishka import Provider, Scope, provide
from .repository import TaskRepository
class TaskProvider(Provider):
@provide(scope=Scope.APP)
def get_repository(self) -> TaskRepository:
return TaskRepository()
@@ -0,0 +1,19 @@
from dataclasses import dataclass
from typing import Literal
Priority = Literal["low", "medium", "high"]
@dataclass
class Task:
description: str
priority: Priority = "medium"
class TaskRepository:
def __init__(self):
self._tasks: list[Task] = []
def add_task(self, task: Task):
self._tasks.append(task)
def get_all_tasks(self) -> list[Task]:
return self._tasks