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

1from __future__ import annotations 

2 

3from dataclasses import dataclass, field 

4from enum import Enum 

5from typing import Any, Dict, Optional 

6from urllib.parse import parse_qs, urlparse, urlunparse 

7 

8 

9class HttpMethod(Enum): 

10 """Represents an HTTP method.""" 

11 

12 GET = "GET" 

13 """Retrieves data from a server. 

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

15 POST = "POST" 

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

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

18 PUT = "PUT" 

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

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

21 PATCH = "PATCH" 

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

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

24 DELETE = "DELETE" 

25 """Deletes a resource from a server.""" 

26 HEAD = "HEAD" 

27 """Retrieves metadata from a server. 

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

29 OPTIONS = "OPTIONS" 

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

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

32 

33 

34@dataclass(frozen=True) 

35class URL: 

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

37 

38 full_url: str 

39 """The full URL.""" 

40 base_url: str = "" 

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

42 secure: bool = False 

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

44 protocol: str = "" 

45 """The protocol of the URL.""" 

46 path: str = "" 

47 """The path of the URL.""" 

48 domain_with_port: str = "" 

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

50 domain: str = "" 

51 """The domain of the URL.""" 

52 port: Optional[int] = None 

53 """The port of the URL.""" 

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

55 """A dictionary of query parameters.""" 

56 

57 def __post_init__(self) -> None: 

58 parsed_url = urlparse(self.full_url) 

59 

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

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

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

63 

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

65 

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

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

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

69 if len(full_domen) > 1: 

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

71 

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

73 

74 

75class Proxy: 

76 """ 

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

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

79 2. Playwright dict: { 

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

81 'username': 'user', 

82 'password': 'pass' 

83 } 

84 """ 

85 

86 def __init__( 

87 self, 

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

89 *, 

90 server: Optional[str] = None, 

91 username: Optional[str] = None, 

92 password: Optional[str] = None, 

93 ): 

94 """ 

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

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

97 """ 

98 self._server: str | None = None 

99 self._username: str | None = None 

100 self._password: str | None = None 

101 

102 if proxy is not None: 

103 if isinstance(proxy, str): 

104 self._from_str(proxy) 

105 elif isinstance(proxy, dict): 

106 self._from_dict(proxy) 

107 else: 

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

109 elif server is not None: 

110 self._server = server 

111 self._username = username 

112 self._password = password 

113 else: 

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

115 pass 

116 

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

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

119 if not proxy_str.strip(): 

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

121 

122 original = proxy_str 

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

124 if "://" not in proxy_str: 

125 proxy_str = "http://" + proxy_str 

126 

127 parsed = urlparse(proxy_str) 

128 

129 if not parsed.hostname: 

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

131 

132 self._server = urlunparse( 

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

134 ) 

135 

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

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

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

139 

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

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

142 server = proxy_dict.get("server") 

143 if not server: 

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

145 

146 parsed = urlparse(server) 

147 if not parsed.hostname: 

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

149 

150 self._server = server 

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

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

153 

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

155 """ 

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

157 """ 

158 if not self._server: 

159 return None 

160 

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

162 if self._username: 

163 result["username"] = self._username 

164 if self._password: 

165 result["password"] = self._password 

166 return result 

167 

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

169 """ 

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

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

172 """ 

173 if not self._server: 

174 return None 

175 

176 parsed = urlparse(self._server) 

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

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

179 

180 netloc = parsed.hostname 

181 if parsed.port: 

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

183 

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

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

186 if self._password: 

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

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

189 

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

191 

192 def __repr__(self) -> str: 

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

194 

195 def __bool__(self) -> bool: 

196 return bool(self._server)