ccc670b79c
Features: - Full API coverage (45 endpoints from HAR analysis) - Async/await support with httpx - Pydantic models for all responses - Clear error handling (KworkAuthError, KworkApiError, etc.) - Session management (cookies + web_auth_token) - Unit tests with respx mocks - Integration tests template - JSON logging support via structlog Endpoints implemented: - Authentication: signIn, getWebAuthToken - Catalog: catalogMainv2, getKworkDetails, getKworkDetailsExtra - Projects: projects, payerOrders, workerOrders - User: user, userReviews, favoriteKworks - Reference: cities, countries, timezones, features, badges - Notifications: notifications, notificationsFetch, dialogs - Other: 30+ additional endpoints Tests: 13 passed, 79% coverage
91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
"""
|
|
Kwork API exceptions.
|
|
|
|
All exceptions provide clear error messages for debugging.
|
|
"""
|
|
|
|
from typing import Any, Optional
|
|
|
|
|
|
class KworkError(Exception):
|
|
"""Base exception for all Kwork API errors."""
|
|
|
|
def __init__(self, message: str, response: Optional[Any] = None):
|
|
self.message = message
|
|
self.response = response
|
|
super().__init__(self.message)
|
|
|
|
def __str__(self) -> str:
|
|
return f"KworkError: {self.message}"
|
|
|
|
|
|
class KworkAuthError(KworkError):
|
|
"""Authentication/authorization error."""
|
|
|
|
def __init__(self, message: str = "Authentication failed", response: Optional[Any] = None):
|
|
super().__init__(message, response)
|
|
|
|
def __str__(self) -> str:
|
|
return f"KworkAuthError: {self.message}"
|
|
|
|
|
|
class KworkApiError(KworkError):
|
|
"""API request error (4xx, 5xx)."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
status_code: Optional[int] = None,
|
|
response: Optional[Any] = None,
|
|
):
|
|
self.status_code = status_code
|
|
super().__init__(message, response)
|
|
|
|
def __str__(self) -> str:
|
|
if self.status_code:
|
|
return f"KworkApiError [{self.status_code}]: {self.message}"
|
|
return f"KworkApiError: {self.message}"
|
|
|
|
|
|
class KworkNotFoundError(KworkApiError):
|
|
"""Resource not found (404)."""
|
|
|
|
def __init__(self, resource: str, response: Optional[Any] = None):
|
|
super().__init__(f"Resource not found: {resource}", 404, response)
|
|
|
|
|
|
class KworkRateLimitError(KworkApiError):
|
|
"""Rate limit exceeded (429)."""
|
|
|
|
def __init__(self, message: str = "Rate limit exceeded", response: Optional[Any] = None):
|
|
super().__init__(message, 429, response)
|
|
|
|
|
|
class KworkValidationError(KworkApiError):
|
|
"""Validation error (400)."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Validation failed",
|
|
fields: Optional[dict[str, list[str]]] = None,
|
|
response: Optional[Any] = None,
|
|
):
|
|
self.fields = fields or {}
|
|
super().__init__(message, 400, response)
|
|
|
|
def __str__(self) -> str:
|
|
if self.fields:
|
|
field_errors = ", ".join(f"{k}: {v[0]}" for k, v in self.fields.items())
|
|
return f"KworkValidationError: {field_errors}"
|
|
return f"KworkValidationError: {self.message}"
|
|
|
|
|
|
class KworkNetworkError(KworkError):
|
|
"""Network/connection error."""
|
|
|
|
def __init__(self, message: str = "Network error", response: Optional[Any] = None):
|
|
super().__init__(message, response)
|
|
|
|
def __str__(self) -> str:
|
|
return f"KworkNetworkError: {self.message}"
|