client/main.py - master - tinypot/lab4

client/main.py
182 строки8 KB
Tim Polus
lab4: Task Management API через RabbitMQ
02 мар 2026, 06:17
02 мар 2026, 06:1752ab147
#!/usr/bin/env python3
from __future__ import annotations

import json
import uuid
import sys

from client import RpcClient


def pp(data) -> None:
    print(json.dumps(data, indent=2, ensure_ascii=False, default=str))


def prompt(msg: str, default: str = "") -> str:
    if default:
        val = input(f"{msg} [{default}]: ").strip()
        return val or default
    return input(f"{msg}: ").strip()


def show_menu() -> None:
    print(
        """
╔══════════════════════════════════════════════════╗
║     Task Management API — RabbitMQ Client        ║
╠══════════════════════════════════════════════════╣
║  Категории                                       ║
║   1 — Список категорий                           ║
║   2 — Создать категорию                          ║
║   3 — Обновить категорию                         ║
║   4 — Удалить категорию                          ║
╠══════════════════════════════════════════════════╣
║  Задачи                                          ║
║   5 — Список задач (пагинация)                   ║
║   6 — Получить задачу по ID                      ║
║   7 — Создать задачу                             ║
║   8 — Обновить задачу                            ║
║   9 — Удалить задачу                             ║
╠══════════════════════════════════════════════════╣
║  Прочее                                          ║
║  10 — Список пользователей                       ║
║  11 — Демо: идемпотентность                      ║
║  12 — Переключить версию (v1/v2)                 ║
║   0 — Выход                                      ║
╚══════════════════════════════════════════════════╝"""
    )


def demo_idempotency(client: RpcClient) -> None:
    req_id = str(uuid.uuid4())
    data = {"title": "Idempotent task", "description": "test"}
    print(f"\nrequest_id: {req_id}")

    print("\n>>> Первый запрос create_task:")
    r1 = client.call("create_task", data, request_id=req_id)
    pp(r1)

    print("\n>>> Повторный запрос с тем же id:")
    r2 = client.call("create_task", data, request_id=req_id)
    pp(r2)

    if r1 == r2:
        print("\nОтветы идентичны — дубликат НЕ создан.")
    else:
        print("\nОтветы различаются.")


def main() -> None:
    api_key = prompt("API-ключ", "master-api-key")
    version = "v1"

    try:
        client = RpcClient(api_key=api_key, version=version)
    except Exception as e:
        print(f"Не удалось подключиться к RabbitMQ: {e}")
        sys.exit(1)

    print(f"Подключено к RabbitMQ. Версия API: {version}")

    try:
        while True:
            show_menu()
            choice = prompt("\nКоманда", "0")

            try:
                if choice == "0":
                    break

                elif choice == "1":
                    pp(client.call("get_categories"))

                elif choice == "2":
                    name = prompt("Название")
                    desc = prompt("Описание (Enter — пропустить)") or None
                    color = prompt("Цвет hex (Enter — пропустить)") or None
                    pp(client.call("create_category", {
                        "name": name, "description": desc, "color": color,
                    }))

                elif choice == "3":
                    cid = int(prompt("ID категории"))
                    name = prompt("Новое название (Enter — не менять)") or None
                    data = {"category_id": cid}
                    if name:
                        data["name"] = name
                    pp(client.call("update_category", data))

                elif choice == "4":
                    cid = int(prompt("ID категории"))
                    pp(client.call("delete_category", {"category_id": cid}))

                elif choice == "5":
                    page = int(prompt("Страница", "1"))
                    size = int(prompt("Размер", "10"))
                    extra = {}
                    if version == "v2":
                        pr = prompt("Фильтр по приоритету (low/medium/high/urgent, Enter — все)") or None
                        if pr:
                            extra["priority"] = pr
                    pp(client.call("get_tasks", {"page": page, "page_size": size, **extra}))

                elif choice == "6":
                    tid = int(prompt("ID задачи"))
                    pp(client.call("get_task", {"task_id": tid}))

                elif choice == "7":
                    title = prompt("Название")
                    desc = prompt("Описание (Enter — пропустить)") or None
                    priority = prompt("Приоритет (low/medium/high/urgent)", "medium")
                    cat_raw = prompt("ID категории (Enter — без категории)") or None
                    pp(client.call("create_task", {
                        "title": title,
                        "description": desc,
                        "priority": priority,
                        "category_id": int(cat_raw) if cat_raw else None,
                    }))

                elif choice == "8":
                    tid = int(prompt("ID задачи"))
                    data = {"task_id": tid}
                    title = prompt("Новое название (Enter — не менять)") or None
                    status = prompt("Статус (pending/in_progress/completed/cancelled, Enter — не менять)") or None
                    priority = prompt("Приоритет (low/medium/high/urgent, Enter — не менять)") or None
                    if title:
                        data["title"] = title
                    if status:
                        data["status"] = status
                    if priority:
                        data["priority"] = priority
                    pp(client.call("update_task", data))

                elif choice == "9":
                    tid = int(prompt("ID задачи"))
                    pp(client.call("delete_task", {"task_id": tid}))

                elif choice == "10":
                    pp(client.call("get_users"))

                elif choice == "11":
                    demo_idempotency(client)

                elif choice == "12":
                    version = "v2" if version == "v1" else "v1"
                    client.version = version
                    print(f"Версия переключена на {version}")

                else:
                    print("Неизвестная команда.")

            except KeyboardInterrupt:
                break
            except Exception as e:
                print(f"Ошибка: {e}")

    finally:
        client.close()
        print("Отключено.")


if __name__ == "__main__":
    main()