client.py - master - 1Rrd0/3Lab

client.py
158 строк6 KB
dima
add
18 дек 2025, 16:26
18 дек 2025, 16:26ac92f21
from __future__ import annotations

import time
import uuid
from dataclasses import dataclass
from typing import Any, Dict, Optional, List

import httpx


@dataclass
class ApiError(Exception):
    status_code: int
    message: str
    details: Any = None

    def __str__(self) -> str:
        return f"API error {self.status_code}: {self.message} | details={self.details}"


class TaskManagerClient:
    """
    Устойчивый клиент к Task Manager API (v1/v2):
    - JWT auth
    - CRUD tasks
    - pagination (limit/offset)
    - idempotency via Idempotency-Key
    - rate limit (429 + Retry-After)
    """

    def __init__(self, base_url: str, api_v1_prefix: str, api_v2_prefix: str, timeout: float = 10.0):
        self.base_url = base_url.rstrip("/")
        self.v1 = f"{self.base_url}/{api_v1_prefix.strip('/')}"
        self.v2 = f"{self.base_url}/{api_v2_prefix.strip('/')}"
        self._token: Optional[str] = None
        self._http = httpx.Client(timeout=timeout)

    def close(self) -> None:
        self._http.close()

    def _auth_headers(self) -> Dict[str, str]:
        if not self._token:
            return {}
        return {"Authorization": f"Bearer {self._token}"}

    def _request(self, method: str, url: str, *, headers: Optional[Dict[str, str]] = None, **kwargs) -> Any:
        all_headers = {}
        all_headers.update(headers or {})

        # простая ретрай-логика на 429
        for attempt in range(2):
            r = self._http.request(method, url, headers=all_headers, follow_redirects=True, **kwargs)


            # покажем rate limit заголовки (пригодится для отчёта/демо)
            limit_left = r.headers.get("X-Limit-Remaining")
            retry_after = r.headers.get("Retry-After")

            if r.status_code == 429:
                wait_s = int(retry_after) if (retry_after and retry_after.isdigit()) else 1
                if attempt == 0:
                    time.sleep(wait_s)
                    continue

            if 200 <= r.status_code < 300:
                if r.status_code == 204:
                    return None
                # иногда может быть пустое тело
                if not r.text:
                    return None
                return r.json()

            # ошибка — пытаемся достать detail
            try:
                body = r.json()
            except Exception:
                body = r.text

            msg = "Request failed"
            if isinstance(body, dict) and "detail" in body:
                msg = str(body["detail"])

            raise ApiError(r.status_code, msg, details={"body": body, "x_limit_remaining": limit_left, "retry_after": retry_after})

        raise ApiError(429, "Rate limit exceeded (retries failed)")

    # -------- AUTH v1 --------

    def register(self, email: str, password: str, full_name: str) -> Dict[str, Any]:
        url = f"{self.v1}/auth/register"
        payload = {"email": email, "password": password, "full_name": full_name}
        return self._request("POST", url, json=payload)

    def login(self, email: str, password: str) -> str:
        url = f"{self.v1}/auth/login"
        payload = {"email": email, "password": password}
        data = self._request("POST", url, json=payload)
        token = data.get("access_token")
        if not token:
            raise ApiError(500, "No access_token in response", details=data)
        self._token = token
        return token

    # -------- TASKS v1 --------

    def list_tasks_v1(self, limit: int = 20, offset: int = 0) -> List[Dict[str, Any]]:
        url = f"{self.v1}/tasks"
        return self._request("GET", url, headers=self._auth_headers(), params={"limit": limit, "offset": offset})

    def create_task_v1(self, title: str, description: Optional[str] = None, status: str = "todo",
                       due_date: Optional[str] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]:
        url = f"{self.v1}/tasks"
        idem = idempotency_key or str(uuid.uuid4())

        headers = self._auth_headers()
        headers["Idempotency-Key"] = idem  # ключ идемпотентности

        payload: Dict[str, Any] = {"title": title, "description": description, "status": status, "due_date": due_date}
        # уберём None чтобы не ругаться на валидации
        payload = {k: v for k, v in payload.items() if v is not None}

        return self._request("POST", url, headers=headers, json=payload)

    def get_task_v1(self, task_id: int) -> Dict[str, Any]:
        url = f"{self.v1}/tasks/{task_id}"
        return self._request("GET", url, headers=self._auth_headers())

    def update_task_v1(self, task_id: int, **fields) -> Dict[str, Any]:
        url = f"{self.v1}/tasks/{task_id}"
        payload = {k: v for k, v in fields.items() if v is not None}
        return self._request("PUT", url, headers=self._auth_headers(), json=payload)

    def delete_task_v1(self, task_id: int) -> None:
        url = f"{self.v1}/tasks/{task_id}"
        return self._request("DELETE", url, headers=self._auth_headers())

    # -------- TASKS v2 --------

    def list_tasks_v2(self, limit: int = 20, offset: int = 0, status_filter: Optional[str] = None,
                     priority_filter: Optional[str] = None) -> List[Dict[str, Any]]:
        url = f"{self.v2}/tasks"
        params = {"limit": limit, "offset": offset}
        if status_filter:
            params["status_filter"] = status_filter
        if priority_filter:
            params["priority_filter"] = priority_filter
        return self._request("GET", url, headers=self._auth_headers(), params=params)

    def list_tasks_fields_v2(self, include: Optional[str] = None, limit: int = 20, offset: int = 0) -> List[Dict[str, Any]]:
        url = f"{self.v2}/tasks/fields"
        params = {"limit": limit, "offset": offset}
        if include:
            params["include"] = include
        return self._request("GET", url, headers=self._auth_headers(), params=params)

    def internal_stats_v2(self) -> Dict[str, Any]:
        url = f"{self.v2}/internal/stats"
        return self._request("GET", url, headers=self._auth_headers())