Coverage for human_requests / abstraction / http.py: 94%

1from __future__ import annotations 

2 

3import os 

4from dataclasses import dataclass, field 

5from enum import Enum 

6from typing import Any, Dict, Optional 

7from urllib.parse import parse_qs, urlparse, urlunparse 

8 

9 

10class HttpMethod(Enum): 

11 """Represents an HTTP method.""" 

12 

13 GET = "GET" 

14 """Retrieves data from a server. 

15 It only reads data and does not modify it.""" 

16 POST = "POST" 

17 """Submits data to a server to create a new resource. 

18 It can also be used to update existing resources.""" 

19 PUT = "PUT" 

20 """Updates a existing resource on a server. 

21 It can also be used to create a new resource.""" 

22 PATCH = "PATCH" 

23 """Updates a existing resource on a server. 

24 It only updates the fields that are provided in the request body.""" 

25 DELETE = "DELETE" 

26 """Deletes a resource from a server.""" 

27 HEAD = "HEAD" 

28 """Retrieves metadata from a server. 

29 It only reads the headers and does not return the response body.""" 

30 OPTIONS = "OPTIONS" 

31 """Provides information about the HTTP methods supported by a server. 

32 It can be used for Cross-Origin Resource Sharing (CORS) request.""" 

33 

34 

35@dataclass(frozen=True) 

36class URL: 

37 """A dataclass containing the parsed URL components.""" 

38 

39 full_url: str 

40 """The full URL.""" 

41 base_url: str = "" 

42 """The base URL, without query parameters.""" 

43 secure: bool = False 

44 """Whether the URL is secure (https/wss).""" 

45 protocol: str = "" 

46 """The protocol of the URL.""" 

47 path: str = "" 

48 """The path of the URL.""" 

49 domain_with_port: str = "" 

50 """The domain of the URL with port.""" 

51 domain: str = "" 

52 """The domain of the URL.""" 

53 port: Optional[int] = None 

54 """The port of the URL.""" 

55 params: dict[str, list[str]] = field(default_factory=dict) 

56 """A dictionary of query parameters.""" 

57 

58 def __post_init__(self) -> None: 

59 parsed_url = urlparse(self.full_url) 

60 

61 object.__setattr__(self, "base_url", parsed_url._replace(query="").geturl()) 

62 object.__setattr__(self, "secure", parsed_url.scheme in ["https", "wss"]) 

63 object.__setattr__(self, "protocol", parsed_url.scheme) 

64 

65 object.__setattr__(self, "path", parsed_url.path) 

66 

67 full_domen = parsed_url.netloc.split(":") 

68 object.__setattr__(self, "domain_with_port", parsed_url.netloc) 

69 object.__setattr__(self, "domain", full_domen[0]) 

70 if len(full_domen) > 1: 

71 object.__setattr__(self, "port", int(full_domen[1])) 

72 

73 object.__setattr__(self, "params", parse_qs(parsed_url.query)) 

74 

75 

76class Proxy: 

77 """ 

78 Универсальный класс для работы с прокси в двух форматах: 

79 1. Строковый: 'http://user:pass@host:port' или 'socks5://host:port' 

80 2. Playwright dict: { 

81 'server': 'http://host:port', 

82 'username': 'user', 

83 'password': 'pass' 

84 } 

85 """ 

86 

87 @staticmethod 

88 def from_env() -> Proxy: 

89 """ 

90 Создаёт Proxy из переменных окружения. 

91 

92 Приоритет: 

93 1) https_proxy / HTTPS_PROXY 

94 2) http_proxy / HTTP_PROXY 

95 3) all_proxy / ALL_PROXY 

96 """ 

97 env_keys = ( 

98 "https_proxy", 

99 "HTTPS_PROXY", 

100 "http_proxy", 

101 "HTTP_PROXY", 

102 "all_proxy", 

103 "ALL_PROXY", 

104 ) 

105 

106 for key in env_keys: 

107 value = os.getenv(key) 

108 if value and value.strip(): 

109 return Proxy(value.strip()) 

110 return Proxy() 

111 

112 def __init__( 

113 self, 

114 proxy: Optional[str | Dict[str, Any]] = None, 

115 *, 

116 server: Optional[str] = None, 

117 username: Optional[str] = None, 

118 password: Optional[str] = None, 

119 ): 

120 """ 

121 Инициализация через строку или dict (Playwright-формат). 

122 Можно также передать параметры напрямую. 

123 """ 

124 self._server: str | None = None 

125 self._username: str | None = None 

126 self._password: str | None = None 

127 

128 if proxy is not None: 

129 if isinstance(proxy, str): 

130 self._from_str(proxy) 

131 elif isinstance(proxy, dict): 

132 self._from_dict(proxy) 

133 else: 

134 raise ValueError("proxy должен быть str или dict") 

135 elif server is not None: 

136 self._server = server 

137 self._username = username 

138 self._password = password 

139 else: 

140 # Прокси не указан 

141 pass 

142 

143 def _from_str(self, proxy_str: str) -> None: 

144 """Парсит строку вида protocol://user:pass@host:port""" 

145 if not proxy_str.strip(): 

146 raise ValueError("Прокси-строка не может быть пустой") 

147 

148 original = proxy_str 

149 # Поддержка без схемы: host:port или user:pass@host:port → всегда добавляем http:// 

150 if "://" not in proxy_str: 

151 proxy_str = "http://" + proxy_str 

152 

153 parsed = urlparse(proxy_str) 

154 

155 if not parsed.hostname: 

156 raise ValueError(f"Некорректный хост в прокси: {original}") 

157 

158 self._server = urlunparse( 

159 (parsed.scheme, f"{parsed.hostname}:{parsed.port or ''}".rstrip(":"), "", "", "", "") 

160 ) 

161 

162 # parsed.username может быть '', но мы ставим None если пусто 

163 self._username = parsed.username if parsed.username else None 

164 self._password = parsed.password if parsed.password else None 

165 

166 def _from_dict(self, proxy_dict: Dict[str, Any]) -> None: 

167 """Принимает dict в формате Playwright""" 

168 server = proxy_dict.get("server") 

169 if not server: 

170 raise ValueError("В dict должен быть ключ 'server'") 

171 

172 parsed = urlparse(server) 

173 if not parsed.hostname: 

174 raise ValueError(f"Некорректный server в dict: {server}") 

175 

176 self._server = server 

177 self._username = proxy_dict.get("username") 

178 self._password = proxy_dict.get("password") 

179 

180 def as_dict(self) -> Optional[Dict[str, Any]]: 

181 """ 

182 Возвращает прокси в формате Playwright 

183 """ 

184 if not self._server: 

185 return None 

186 

187 result: Dict[str, Any] = {"server": self._server} 

188 if self._username: 

189 result["username"] = self._username 

190 if self._password: 

191 result["password"] = self._password 

192 return result 

193 

194 def as_str(self, include_auth: bool = True) -> Optional[str]: 

195 """ 

196 Возвращает прокси в строковом формате. 

197 Если include_auth=False — без логина и пароля. 

198 """ 

199 if not self._server: 

200 return None 

201 

202 parsed = urlparse(self._server) 

203 if not parsed.scheme or not parsed.hostname: 

204 raise ValueError(f"Некорректный server: {self._server}") 

205 

206 netloc = parsed.hostname 

207 if parsed.port: 

208 netloc += f":{parsed.port}" 

209 

210 if include_auth and (self._username or self._password): 

211 auth = f"{self._username or ''}" 

212 if self._password: 

213 auth += f":{self._password}" 

214 netloc = f"{auth}@{netloc}" 

215 

216 return urlunparse((parsed.scheme, netloc, "", "", "", "")) 

217 

218 def __repr__(self) -> str: 

219 return f"Proxy(server={self._server}, username={'***' if self._username else None})" 

220 

221 def __bool__(self) -> bool: 

222 return bool(self._server)