client/client.py - lab4 - Pepegator/task-api

client/client.py
297 строк10 KB
Pol136
fix bugs
30 дек 2025, 14:46
30 дек 2025, 14:46c8d3e65
import requests
import uuid
import time
import os
from typing import Optional, Dict, List, Any
from dataclasses import dataclass


os.environ["HTTP_PROXY"] = ""
os.environ["HTTPS_PROXY"] = ""
os.environ["http_proxy"] = ""
os.environ["https_proxy"] = ""

NO_PROXIES = {"http": "", "https": ""}


@dataclass
class RateLimitInfo:
    limit: int
    remaining: int
    retry_after: int


class BookShareClient:
    
    def __init__(self, base_url: str = "http://localhost:8000", api_version: str = "v1"):
        self.base_url = base_url.rstrip('/')
        self.api_version = api_version
        self.token = None
        self.session = requests.Session()
        
    def _get_rate_limit_info(self, response: requests.Response) -> Optional[RateLimitInfo]:
        try:
            if 200 <= response.status_code < 300 or response.status_code == 429:
                remaining = int(response.headers.get('X-Limit-Remaining', 60))
                retry_after = int(response.headers.get('Retry-After', 60))
                limit = int(response.headers.get('X-Limit-Limit', 60))
                return RateLimitInfo(limit=limit, remaining=remaining, retry_after=retry_after)
        except:
            pass
        return None
    
    def _request(
        self,
        method: str,
        endpoint: str,
        data: Optional[Dict] = None,
        params: Optional[Dict] = None,
        auto_retry_429: bool = True,
        form_data: bool = False,
        **kwargs
    ) -> Dict[str, Any]:
        url = f"{self.base_url}/api/{self.api_version}{endpoint}"
        
        headers = kwargs.pop("headers", {}).copy() if "headers" in kwargs else {}
        
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        
        if method == "POST":
            idempotency_key = kwargs.pop("idempotency_key", None) or str(uuid.uuid4())
            headers["Idempotency-Key"] = idempotency_key
        
        if form_data and data:
            response = self.session.request(
                method,
                url,
                data=data,
                params=params,
                headers=headers,
                proxies=NO_PROXIES,
                **kwargs
            )
        else:
            response = self.session.request(
                method,
                url,
                json=data,
                params=params,
                headers=headers,
                proxies=NO_PROXIES,
                **kwargs
            )
        
        rate_limit = self._get_rate_limit_info(response)
        if rate_limit:
            if rate_limit.remaining <= 10:
                print(f"Rate limit: {rate_limit.remaining}/{rate_limit.limit} запросов осталось")
        
        if response.status_code == 429 and auto_retry_429:
            retry_after = rate_limit.retry_after if rate_limit else 60
            print(f"Rate limit exceeded. Waiting {retry_after} seconds...")
            time.sleep(retry_after)
            return self._request(
                method,
                endpoint,
                data=data,
                params=params,
                form_data=form_data,
                auto_retry_429=False,
                **kwargs
            )
        
        try:
            response.raise_for_status()
        except requests.HTTPError as e:
            try:
                error_data = response.json()
                raise requests.HTTPError(f"{e.response.status_code}: {error_data.get('detail', str(e))}")
            except:
                raise
        
        return response.json()
    
    def register(
        self,
        email: str,
        password: str,
        full_name: str = "Test User",
        bio: str = ""
    ) -> Dict[str, Any]:
        return self._request("POST", "/auth/register", data={
            "email": email,
            "password": password,
            "full_name": full_name,
            "bio": bio
        })
    
    def login(self, email: str, password: str) -> Dict[str, Any]:
        resp = self._request(
            "POST", 
            "/auth/token", 
            data={
                "username": email,
                "password": password
            },
            form_data=True
        )
        self.token = resp.get("access_token")
        return resp
    
    def get_profile(self, include: Optional[str] = None) -> Dict[str, Any]:
        params = {}
        if include:
            params["include"] = include
        return self._request("GET", "/profile/me", params=params)
    
    def update_profile(
        self, 
        include: Optional[str] = None,
        **kwargs
    ) -> Dict[str, Any]:
        params = {}
        if include:
            params["include"] = include
        return self._request("PATCH", "/profile/me", data=kwargs, params=params)
    
    def create_book(
        self,
        title: str,
        author: str,
        description: str = "",
        tags: Optional[List[str]] = None,
        condition: Optional[str] = None,
        include: Optional[str] = None
    ) -> Dict[str, Any]:
        payload = {
            "title": title,
            "author": author
        }
        
        if description:
            payload["description"] = description
        
        if tags:
            payload["tags"] = tags
        
        if condition:
            payload["condition"] = condition
        
        params = {}
        if include:
            params["include"] = include
        
        return self._request("POST", "/books", data=payload, params=params)
    
    def list_books(
        self,
        limit: int = 20,
        offset: int = 0,
        condition: Optional[str] = None,
        include: Optional[str] = None
    ) -> Dict[str, Any]:
        params = {
            "limit": min(limit, 100),
            "offset": max(offset, 0)
        }
        
        if condition:
            params["condition"] = condition
        
        if include:
            params["include"] = include
        
        return self._request("GET", "/books", params=params)
    
    def get_book(self, book_id: int, include: Optional[str] = None) -> Dict[str, Any]:
        params = {}
        if include:
            params["include"] = include
        return self._request("GET", f"/books/{book_id}", params=params)
    
    def generate_qr(self, book_id: int) -> Dict[str, Any]:
        return self._request("POST", f"/books/{book_id}/qr", data={})
    
    def borrow_book(self, book_id: int) -> Dict[str, Any]:
        return self._request("POST", f"/books/{book_id}/borrow", data={})
    
    def return_book(self, book_id: int) -> Dict[str, Any]:
        return self._request("POST", f"/books/{book_id}/return", data={})
    
    def wait_for_api(self, timeout: int = 30) -> bool:
        start = time.time()
        while time.time() - start < timeout:
            try:
                self.session.get(f"{self.base_url}/health", timeout=2, proxies=NO_PROXIES)
                print("API доступен!")
                return True
            except:
                print("Ждём API...", end="\r")
                time.sleep(1)
        
        print("API не ответил за отведённое время")
        return False


if __name__ == "__main__":
    client = BookShareClient("http://localhost:8000", "v1")
    
    if not client.wait_for_api():
        exit(1)
    
    try:
        print("\nРегистрация...")
        email = f"test{uuid.uuid4().hex[:8]}@example.com"
        user = client.register(email, "StrongPass123!", "Test User", "Test bio")
        print(f"Зарегистрирован: ID={user.get('id')}, email={user.get('email')}")
        
        print("\nВход...")
        token_resp = client.login(email, "StrongPass123!")
        print(f"Token получен: {token_resp['access_token'][:20]}...")
        
        print("\nПолучение профиля...")
        profile = client.get_profile(include="email,full_name")
        print(f"Профиль: {profile}")
        
        print("\nСоздание книги (v1)...")
        book_v1 = client.create_book(
            title="1984",
            author="George Orwell",
            description="Dystopian novel",
            tags=["fiction", "dystopian"],
            include="id,title,author"
        )
        print(f"Книга создана: ID={book_v1.get('id')}, title={book_v1.get('title')}")
        
        print("\nСписок книг (limit=5, include=id,title)...")
        books = client.list_books(limit=5, include="id,title")
        print(f"Получено {len(books['items'])} книг из {books['total']} всего")
        for item in books['items'][:3]:
            print(f"  - {item}")
        
        print("\nГенерация QR кода...")
        if book_v1.get('id'):
            qr_data = client.generate_qr(book_v1['id'])
            print(f"QR код создан (base64 длина: {len(qr_data.get('qr_png_base64', ''))})")
        
        print("\nПопытка создания книги (v2)...")
        client_v2 = BookShareClient("http://localhost:8000", "v2")
        client_v2.token = client.token
        try:
            book_v2 = client_v2.create_book(
                title="Brave New World",
                author="Aldous Huxley",
                condition="good",
                include="id,title,condition"
            )
            print(f"v2 книга создана: condition={book_v2.get('condition')}")
        except Exception as e:
            print(f"v2 может быть не полностью реализована: {e}")
        
        print("\nПример работает!")
        
    except Exception as e:
        print(f"\nОшибка: {e}")
        import traceback
        traceback.print_exc()