File size: 3,327 Bytes
e0b68d2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | #!/usr/bin/env python
# -*- coding: utf-8 -*-
"""HTTP 客户端异常类。
用于处理 HTTP 请求过程中的错误，如网络问题、SSL、代理等。
"""
from typing import Any, Dict, Optional
from app.exceptions.base import AppError
class HttpError(AppError):
"""HTTP 客户端错误基类。
当 HTTP 请求失败时抛出。
Attributes:
url: 请求的 URL
method: HTTP 方法
Example:
>>> raise HttpError("Request failed", url="https://api.example.com", method="POST")
"""
def __init__(
self,
message: str,
url: Optional[str] = None,
method: Optional[str] = None,
details: Optional[Dict[str, Any]] = None,
) -> None:
final_details = details or {}
if url:
final_details["url"] = url
if method:
final_details["method"] = method
super().__init__(
message=message,
code="http_error",
status_code=502,
details=final_details,
)
class HttpConnectionError(HttpError):
"""连接失败异常。
当无法建立 TCP 连接时抛出。
Example:
>>> raise HttpConnectionError("https://api.example.com", "Connection refused")
"""
def __init__(
self,
url: str,
reason: str,
) -> None:
super().__init__(
message=f"Connection failed: {reason}",
url=url,
)
self.code = "connection_error"
self.status_code = 503
class SSLError(HttpError):
"""SSL/TLS 错误异常。
当 SSL 握手失败或证书验证失败时抛出。
Example:
>>> raise SSLError("https://api.example.com", "Certificate verify failed")
"""
def __init__(
self,
url: str,
reason: str,
) -> None:
super().__init__(
message=f"SSL/TLS error: {reason}",
url=url,
)
self.code = "ssl_error"
class ProxyError(HttpError):
"""代理错误异常。
当代理连接失败或代理返回错误时抛出。
Example:
>>> raise ProxyError("https://api.example.com", "Proxy connection refused")
"""
def __init__(
self,
url: str,
reason: str,
) -> None:
super().__init__(
message=f"Proxy error: {reason}",
url=url,
)
self.code = "proxy_error"
class HttpTimeoutError(HttpError):
"""请求超时异常。
当请求超时时抛出。
Example:
>>> raise HttpTimeoutError("https://api.example.com", "POST", timeout=30.0)
"""
def __init__(
self,
url: str,
method: str,
timeout: Optional[float] = None,
) -> None:
message = f"Request timeout"
if timeout is not None:
message += f" after {timeout}s"
details = {}
if timeout is not None:
details["timeout_seconds"] = timeout
super().__init__(
message=message,
url=url,
method=method,
details=details,
)
self.code = "timeout_error"
self.status_code = 504 |