client.py - lab3 - DisVic/task-tracker

client.py
228 строк8 KB
DisappearedVictory
Final version of Lab 3
28 дек 2025, 14:47
28 дек 2025, 14:4703bd2c5
import httpx
import uuid
import logging
from typing import Optional, Dict, List, Any, Union

# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
logger = logging.getLogger(__name__)

class CrossBookingClient:
    def __init__(self, base_url: str = "http://127.0.0.1:8000", version: str = "v2"):
        self.base_url = base_url.rstrip("/")
        self.version = version
        self.api_prefix = f"/api/{version}"
        self.token = None

        self.http = httpx.Client(base_url=self.base_url, timeout=10.0)

    def _headers(self, idempotency_key: str = None, auth: bool = True) -> Dict[str, str]:
        headers = {}
        if auth and self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        if idempotency_key:
            headers["Idempotency-Key"] = idempotency_key
        return headers

    def _handle_response(self, response: httpx.Response) -> Any:
        """Обрабатывает ответ, проверяет ошибки и лимиты."""
        # Логируем лимиты, если они есть
        remaining = response.headers.get("X-Limit-Remaining")
        if remaining:
            logger.debug(f"Requests remaining: {remaining}")

        try:
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                retry_after = e.response.headers.get("Retry-After", "?")
                logger.error(f"Rate limit exceeded. Retry after {retry_after}s")
            elif e.response.status_code == 401:
                logger.error("Unauthorized. Please login again.")
            elif e.response.status_code == 422:
                logger.error(f"Validation Error: {e.response.text}")
            raise e
        
        if response.status_code == 204:
            return None
            
        return response.json()

    # --- Authentication ---

    def register(self, email: str, password: str, full_name: str, bio: str = "") -> Dict:
        """
        Регистрация пользователя.
        Автоматически генерирует Idempotency-Key.
        """
        idem_key = str(uuid.uuid4())
        payload = {
            "email": email,
            "password": password,
            "full_name": full_name,
            "bio": bio
        }
        
        resp = self.http.post(
            f"{self.api_prefix}/auth/register",
            json=payload,
            headers=self._headers(idempotency_key=idem_key, auth=False)
        )
        return self._handle_response(resp)

    def login(self, email: str, password: str) -> str:
        """
        Получение токена (OAuth2 Password Bearer).
        Сохраняет токен внутри клиента.
        """
        data = {
            "username": email,
            "password": password
        }
        
        resp = self.http.post(
            f"{self.api_prefix}/auth/token",
            data=data,
            headers=self._headers(auth=False)
        )
        
        result = self._handle_response(resp)
        self.token = result["access_token"]
        logger.info(f"Logged in as {email}")
        return self.token

    # --- Profile ---

    def get_profile(self, include: str = None) -> Dict:
        """Получить профиль текущего пользователя."""
        params = {}
        if include:
            params["include"] = include
            
        resp = self.http.get(
            f"{self.api_prefix}/profile/me",
            params=params,
            headers=self._headers()
        )
        return self._handle_response(resp)

    def update_profile(self, full_name: str = None, bio: str = None) -> Dict:
        """Обновить профиль (PATCH)."""
        payload = {}
        if full_name: payload["full_name"] = full_name
        if bio: payload["bio"] = bio
        
        resp = self.http.patch(
            f"{self.api_prefix}/profile/me",
            json=payload,
            headers=self._headers()
        )
        return self._handle_response(resp)

    # --- Books ---

    def create_book(self, title: str, author: str, 
                    description: str = "", tags: List[str] = [],
                    condition: str = None) -> Dict:
        """
        Создать книгу. Поддерживает поле condition для v2.
        Идемпотентность обеспечена автоматическим ключом.
        """
        idem_key = str(uuid.uuid4())
        payload = {
            "title": title,
            "author": author,
            "description": description,
            "tags": tags
        }
        
        if self.version == "v2" and condition:
            payload["condition"] = condition
            
        resp = self.http.post(
            f"{self.api_prefix}/books",
            json=payload,
            headers=self._headers(idempotency_key=idem_key)
        )
        return self._handle_response(resp)

    def get_books(self, limit: int = 10, offset: int = 0, 
                  condition: str = None, include: str = None) -> Dict:
        """Получить список книг с пагинацией."""
        params = {"limit": limit, "offset": offset}
        if include:
            params["include"] = include
        if self.version == "v2" and condition:
            params["condition"] = condition
            
        resp = self.http.get(
            f"{self.api_prefix}/books",
            params=params,
            headers=self._headers()
        )
        return self._handle_response(resp)

    def get_book(self, book_id: int, include: str = None) -> Dict:
        """Получить детали одной книги."""
        params = {}
        if include:
            params["include"] = include

        resp = self.http.get(
            f"{self.api_prefix}/books/{book_id}",
            params=params,
            headers=self._headers()
        )
        return self._handle_response(resp)

    def generate_qr(self, book_id: int) -> str:
        """Генерирует QR код для книги."""
        resp = self.http.post(
            f"{self.api_prefix}/books/{book_id}/qr",
            headers=self._headers()
        )
        data = self._handle_response(resp)
        return data.get("qr_png_base64")

    def borrow_book(self, book_id: int) -> Dict:
        """Взять книгу (Borrow)."""
        resp = self.http.post(
            f"{self.api_prefix}/books/{book_id}/borrow",
            headers=self._headers()
        )
        return self._handle_response(resp)

if __name__ == "__main__":
    client = CrossBookingClient(version="v2")
    
    try:
        # 1. Регистрация
        email = f"user_{uuid.uuid4().hex[:6]}@example.com"
        client.register(email, "StrongPass123!", "Alex Dev", "Python lover")
        
        # 2. Логин
        client.login(email, "StrongPass123!")
        
        # 3. Профиль
        me = client.get_profile(include="email,full_name")
        print(f"My Profile: {me}")
        
        # 4. Создание книги (v2 с состоянием)
        book = client.create_book(
            title="Harry Potter", 
            author="J.K. Rowling", 
            condition="new",
            tags=["fantasy", "magic"]
        )
        print(f"Created Book ID: {book['id']}")
        
        # 5. Получение QR
        qr_code = client.generate_qr(book['id'])
        print(f"QR Generated (len): {len(qr_code)}")
        
        # 6. Список книг
        books_page = client.get_books(limit=5, condition="new")
        print(f"Total books found: {books_page['total']}")

    except Exception as e:
        print(f"Error: {e}")