client/api_client.py - master - tinypot/integr3

client/api_client.py
337 строк11 KB
Tim Polus
lab3: клиент для интеграции с Task Tracker API
02 мар 2026, 05:56
02 мар 2026, 05:56c4962cb
"""
Клиент для интеграции с Task Tracker API (integr_DP/laba3).

Поддерживает:
  - JWT-аутентификацию
  - CRUD проектов и задач (v1 и v2)
  - Пагинацию задач
  - Идемпотентные POST-запросы (заголовок Idempotency-Key)
  - Обработку rate-limit (429) с автоматическим retry
  - Получение внутренней статистики (/api/internal/stats/summary)
"""

from __future__ import annotations

import time
import uuid
from typing import Any

import requests

from config import AUTH_URL, API_V1_URL, API_V2_URL, USERNAME, PASSWORD


class TaskTrackerClient:

    def __init__(
        self,
        username: str = USERNAME,
        password: str = PASSWORD,
        max_retries: int = 5,
        timeout: int = 10,
    ):
        self._username = username
        self._password = password
        self._max_retries = max_retries
        self._timeout = timeout
        self._token: str | None = None
        self._session = requests.Session()

    # ------------------------------------------------------------------
    # Transport
    # ------------------------------------------------------------------

    def _request(
        self,
        method: str,
        url: str,
        *,
        headers: dict[str, str] | None = None,
        **kwargs,
    ) -> requests.Response:
        """Отправка HTTP-запроса с автоматическим retry при 429."""
        merged = self._auth_headers()
        if headers:
            merged.update(headers)

        for attempt in range(1, self._max_retries + 1):
            try:
                resp = self._session.request(
                    method,
                    url,
                    headers=merged,
                    timeout=self._timeout,
                    **kwargs,
                )
            except requests.ConnectionError:
                raise ConnectionError(
                    f"Не удалось подключиться к {url}"
                )
            except requests.Timeout:
                raise TimeoutError(f"Таймаут запроса к {url}")

            if resp.status_code == 429:
                retry_after = self._parse_retry_after(resp)
                print(
                    f"[429] Превышен лимит запросов. "
                    f"Повтор через {retry_after}с (попытка {attempt}/{self._max_retries})"
                )
                time.sleep(retry_after)
                continue

            return resp

        raise RuntimeError(
            f"Превышено максимальное число повторов ({self._max_retries}) для {url}"
        )

    @staticmethod
    def _parse_retry_after(resp: requests.Response) -> int:
        try:
            return int(resp.json().get("retry_after_seconds", 5))
        except Exception:
            return 5

    def _auth_headers(self) -> dict[str, str]:
        h: dict[str, str] = {"Content-Type": "application/json"}
        if self._token:
            h["Authorization"] = f"Bearer {self._token}"
        return h

    @staticmethod
    def _idempotency_header(key: str | None = None) -> dict[str, str]:
        return {"Idempotency-Key": key or str(uuid.uuid4())}

    def _raise_for_status(self, resp: requests.Response) -> None:
        if resp.ok:
            return
        try:
            detail = resp.json()
        except Exception:
            detail = resp.text
        raise RuntimeError(
            f"HTTP {resp.status_code}: {detail}"
        )

    # ------------------------------------------------------------------
    # Аутентификация
    # ------------------------------------------------------------------

    def authenticate(self) -> dict[str, Any]:
        """Получить JWT-токен через POST /api/auth/token."""
        resp = self._request(
            "POST",
            AUTH_URL,
            json={"username": self._username, "password": self._password},
        )
        self._raise_for_status(resp)
        data = resp.json()

        token = data.get("access_Token") or data.get("access_token")
        if not token:
            raise RuntimeError(f"В ответе нет токена: {data}")
        self._token = token
        return data

    @property
    def is_authenticated(self) -> bool:
        return self._token is not None

    # ------------------------------------------------------------------
    # Проекты (v1)
    # ------------------------------------------------------------------

    def get_projects(self) -> list[dict]:
        """GET /api/v1/projects — список всех проектов."""
        resp = self._request("GET", f"{API_V1_URL}/projects")
        self._raise_for_status(resp)
        return resp.json()

    def get_project(self, project_id: int) -> dict:
        """GET /api/v1/projects/{id}."""
        resp = self._request("GET", f"{API_V1_URL}/projects/{project_id}")
        self._raise_for_status(resp)
        return resp.json()

    def create_project(self, name: str) -> dict:
        """POST /api/v1/projects — создать проект."""
        resp = self._request(
            "POST",
            f"{API_V1_URL}/projects",
            json={"name": name},
        )
        self._raise_for_status(resp)
        return resp.json()

    def update_project(self, project_id: int, name: str) -> None:
        """PUT /api/v1/projects/{id} — обновить проект. Возвращает 204."""
        resp = self._request(
            "PUT",
            f"{API_V1_URL}/projects/{project_id}",
            json={"name": name},
        )
        self._raise_for_status(resp)

    def delete_project(self, project_id: int) -> None:
        """DELETE /api/v1/projects/{id}."""
        resp = self._request("DELETE", f"{API_V1_URL}/projects/{project_id}")
        self._raise_for_status(resp)

    # ------------------------------------------------------------------
    # Задачи v1
    # ------------------------------------------------------------------

    def get_tasks(self, page: int = 1, page_size: int = 10) -> dict:
        """GET /api/v1/tasks — с пагинацией."""
        resp = self._request(
            "GET",
            f"{API_V1_URL}/tasks",
            params={"page": page, "pageSize": page_size},
        )
        self._raise_for_status(resp)
        return resp.json()

    def get_all_tasks(self, page_size: int = 20) -> list[dict]:
        """Получить все задачи постранично (v1)."""
        result: list[dict] = []
        page = 1
        while True:
            data = self.get_tasks(page=page, page_size=page_size)
            items = data.get("items", [])
            result.extend(items)
            if len(items) < page_size:
                break
            page += 1
            time.sleep(0.3)
        return result

    def get_task(self, task_id: int, include_project: bool = False) -> dict:
        """GET /api/v1/tasks/{id}."""
        params = {}
        if include_project:
            params["include"] = "project"
        resp = self._request(
            "GET",
            f"{API_V1_URL}/tasks/{task_id}",
            params=params,
        )
        self._raise_for_status(resp)
        return resp.json()

    def create_task(
        self,
        title: str,
        project_id: int,
        *,
        idempotency_key: str | None = None,
    ) -> dict:
        """POST /api/v1/tasks — создать задачу (идемпотентно)."""
        resp = self._request(
            "POST",
            f"{API_V1_URL}/tasks",
            json={"title": title, "projectId": project_id, "isDone": False},
            headers=self._idempotency_header(idempotency_key),
        )
        self._raise_for_status(resp)
        return resp.json()

    def update_task(
        self,
        task_id: int,
        *,
        title: str | None = None,
        is_done: bool | None = None,
    ) -> dict:
        """PUT /api/v1/tasks/{id}."""
        current = self.get_task(task_id)
        payload = {
            "title": title if title is not None else current.get("title"),
            "isDone": is_done if is_done is not None else current.get("isDone"),
        }
        resp = self._request(
            "PUT",
            f"{API_V1_URL}/tasks/{task_id}",
            json=payload,
        )
        self._raise_for_status(resp)
        return resp.json()

    def delete_task(self, task_id: int) -> None:
        """DELETE /api/v1/tasks/{id}."""
        resp = self._request("DELETE", f"{API_V1_URL}/tasks/{task_id}")
        self._raise_for_status(resp)

    # ------------------------------------------------------------------
    # Задачи v2 (расширенная версия с priority)
    # ------------------------------------------------------------------

    def get_tasks_v2(self, page: int = 1, page_size: int = 10) -> dict:
        """GET /api/v2/tasks — пагинация."""
        resp = self._request(
            "GET",
            f"{API_V2_URL}/tasks",
            params={"page": page, "pageSize": page_size},
        )
        self._raise_for_status(resp)
        return resp.json()

    def get_task_v2(self, task_id: int, include_project: bool = False) -> dict:
        """GET /api/v2/tasks/{id}."""
        params = {}
        if include_project:
            params["include"] = "project"
        resp = self._request(
            "GET",
            f"{API_V2_URL}/tasks/{task_id}",
            params=params,
        )
        self._raise_for_status(resp)
        return resp.json()

    def create_task_v2(
        self,
        title: str,
        project_id: int,
        priority: int | None = None,
        *,
        idempotency_key: str | None = None,
    ) -> dict:
        """POST /api/v2/tasks — создать задачу с приоритетом (идемпотентно)."""
        body: dict[str, Any] = {
            "title": title,
            "projectId": project_id,
            "isDone": False,
        }
        if priority is not None:
            body["priority"] = priority
        resp = self._request(
            "POST",
            f"{API_V2_URL}/tasks",
            json=body,
            headers=self._idempotency_header(idempotency_key),
        )
        self._raise_for_status(resp)
        return resp.json()

    # ------------------------------------------------------------------
    # Internal / Stats  (Admin)
    # ------------------------------------------------------------------

    def get_stats(self) -> dict:
        """GET /api/internal/stats/summary — статистика (Admin)."""
        from config import BASE_URL
        resp = self._request("GET", f"{BASE_URL}/api/internal/stats/summary")
        self._raise_for_status(resp)
        return resp.json()

    # ------------------------------------------------------------------
    # Утилиты
    # ------------------------------------------------------------------

    def rate_limit_info(self, resp: requests.Response) -> dict[str, str]:
        """Извлечь заголовки rate-limit из ответа."""
        return {
            k: v
            for k, v in resp.headers.items()
            if k.lower().startswith("x-limit") or k.lower() == "retry-after"
        }