Coverage for gws-app/gws/lib/net/__init_

1from typing import Optional

3import re

4import requests

5import urllib.parse

6import certifi

8import gws

9import gws.lib.osx

12class Error(gws.Error):

13 pass

16class HTTPError(Error):

17 pass

20class Timeout(HTTPError):

21 pass

24class ConnectionError(HTTPError):

25 pass

28class GenericError(HTTPError):

29 pass

32_STATUS_CONNECTION_ERROR = 900

33_STATUS_TIMEOUT = 901

34_STATUS_GENERIC_ERROR = 999

37class Url(gws.Data):

38 fragment: str

39 hostname: str

40 netloc: str

41 params: dict

42 password: str

43 path: str

44 pathparts: gws.lib.osx.ParsePathResult

45 port: int

46 qsl: list

47 query: str

48 scheme: str

49 url: str

50 username: str

53def parse_url(url: str, **kwargs) -> Url:

54 """Parse a string url and return an Url object"""

56 if not is_abs_url(url):

57 url = '//' + url

59 us = urllib.parse.urlsplit(url)

61 u = Url(

62 fragment=us.fragment or '',

63 hostname=us.hostname or '',

64 netloc=us.netloc or '',

65 params={},

66 password=us.password or '',

67 path=us.path or '',

68 pathparts=gws.lib.osx.ParsePathResult(),

69 port=0,

70 qsl=[],

71 query=us.query or '',

72 scheme=us.scheme or '',

73 url=url,

74 username=us.username or '',

75 )

77 if us.port:

78 try:

79 u.port = int(us.port)

80 except ValueError:

81 pass

83 if u.path:

84 u.pathparts = gws.lib.osx.parse_path(u.path)

86 if u.query:

87 u.qsl = urllib.parse.parse_qsl(u.query)

88 for k, v in u.qsl:

89 u.params.setdefault(k, v)

91 if u.username:

92 u.username = unquote(u.username)

93 u.password = unquote(u.get('password') or '')

95 u.update(**kwargs)

96 return u

99_DEFAULT_PORTS = {'http': '80', 'https': '443'}

100

101

102def make_url(u: Optional[Url | dict] = None, **kwargs) -> str:

103 p = gws.u.merge({}, u, kwargs)

104

105 s = ''

106

107 scheme = p.get('scheme', '').lower()

108 if scheme:

109 s += scheme + ':'

110

111 host = p.get('hostname', '')

112 port = p.get('port', '')

113 path = p.get('path', '')

114

115 if scheme or host:

116 s += '//'

117

118 if host:

119 username = p.get('username', '')

120 if username:

121 s += quote_param(username) + ':' + quote_param(p.get('password', '')) + '@'

122

123 s += host

124 if port and str(port) != _DEFAULT_PORTS.get(scheme):

125 s += ':' + str(port)

126

127 if path:

128 s += '/' + quote_path(path.lstrip('/'))

129

130 params = p.get('params')

131 if params:

132 s += '?' + make_qs(params)

133

134 fragment = p.get('fragment', '')

135 if fragment:

136 s += '#' + fragment.lstrip('#')

137

138 return s

139

140

141def parse_qs(x) -> dict:

142 return urllib.parse.parse_qs(x)

143

144

145def make_qs(x) -> str:

146 """Convert a dict/list to a query string.

147

148 For each item in x, if it's a list, join it with a comma, stringify and in utf8.

149

150 Args:

151 x: Value, which can be a dict'able or a list of key,value pairs.

152

153 Returns:

154 The query string.

155 """

156

157 p = []

158 items = x if isinstance(x, (list, tuple)) else gws.u.to_dict(x).items()

159

160 def _value(v):

161 if isinstance(v, (bytes, bytearray)):

162 return v

163 if isinstance(v, str):

164 return v.encode('utf8')

165 if v is True:

166 return b'true'

167 if v is False:

168 return b'false'

169 try:

170 return b','.join(_value(y) for y in v)

171 except TypeError:

172 return str(v).encode('utf8')

173

174 for k, v in items:

175 k = urllib.parse.quote_from_bytes(_value(k))

176 v = urllib.parse.quote_from_bytes(_value(v))

177 p.append(k + '=' + v)

178

179 return '&'.join(p)

180

181

182def quote_param(s: str) -> str:

183 return urllib.parse.quote(s, safe='')

184

185

186def quote_path(s: str) -> str:

187 return urllib.parse.quote(s, safe='/')

188

189

190def unquote(s: str) -> str:

191 return urllib.parse.unquote(s)

192

193

194def add_params(url: str, params: dict = None, **kwargs) -> str:

195 u = parse_url(url)

196 if params:

197 u.params.update(params)

198 u.params.update(kwargs)

199 return make_url(u)

200

201

202def make_relative_url(path: str, params: dict = None, **kwargs) -> str:

203 s = '/' + quote_path(path.lstrip('/'))

204 p = {}

205 if params:

206 p.update(params)

207 p.update(kwargs)

208 if p:

209 s += '?' + make_qs(p)

210 return s

211

212

213def extract_params(url: str) -> tuple[str, dict]:

214 u = parse_url(url)

215 params = u.params

216 u.params = None

217 return make_url(u), params

218

219

220def is_abs_url(url):

221 return re.match(r'^([a-z]+:|)//', url)

224##

227class HTTPResponse:

228 def __init__(

229 self,

230 ok: bool,

231 url: str,

232 res: requests.Response = None,

233 text: str = None,

234 status_code=0,

235 ):

236 self.ok = ok

237 self.url = url

238 if res is not None:

239 self.content_type, self.content_encoding = _parse_content_type(res.headers)

240 self.content = res.content

241 self.status_code = res.status_code

242 else:

243 self.content_type, self.content_encoding = 'text/plain', 'utf8'

244 self.content = text.encode('utf8') if text is not None else b''

245 self.status_code = status_code

246

247 @property

248 def text(self) -> str:

249 if not hasattr(self, '_text'):

250 setattr(self, '_text', _get_text(self.content, self.content_encoding))

251 return getattr(self, '_text')

252

253 def raise_if_failed(self):

254 if self.ok:

255 return

256 if self.status_code == _STATUS_CONNECTION_ERROR:

257 raise ConnectionError(self.text)

258 if self.status_code == _STATUS_TIMEOUT:

259 raise Timeout(self.text)

260 if self.status_code == _STATUS_GENERIC_ERROR:

261 raise GenericError(self.text)

262 raise HTTPError(f'HTTP error: {self.status_code}')

263

264

265def _get_text(content, encoding) -> str:

266 if encoding:

267 try:

268 return str(content, encoding=encoding, errors='strict')

269 except UnicodeDecodeError:

270 pass

271

272 # some folks serve utf8 content without a header, in which case requests thinks it's ISO-8859-1

273 # (see http://docs.python-requests.org/en/master/user/advanced/#encodings)

274 #

275 # 'apparent_encoding' is not always reliable

276 #

277 # therefore when there's no header, we try utf8 first, and then ISO-8859-1

278

279 try:

280 return str(content, encoding='utf8', errors='strict')

281 except UnicodeDecodeError:

282 pass

283

284 try:

285 return str(content, encoding='ISO-8859-1', errors='strict')

286 except UnicodeDecodeError:

287 pass

288

289 # both failed, do utf8 with replace

290

291 gws.log.warning(f'decode failed')

292 return str(content, encoding='utf8', errors='replace')

293

294

295def _parse_content_type_header(header):

296 parts = header.split(';')

297 ctype = parts[0].strip().lower()

298 params: dict[str, str] = {}

299

300 for part in parts[1:]:

301 part = part.strip()

302 if '=' in part:

303 k, v = part.split('=', 1)

304 k = k.strip().lower()

305 v = v.strip()

306 # strip matched quotes (single or double)

307 if len(v) >= 2 and v[0] == v[-1] and v[0] in ('"', "'"):

308 v = v[1:-1]

309 params[k] = v

310

311 return ctype, params

312

313

314def _parse_content_type(headers):

315 # copied from requests.utils.get_encoding_from_headers, but with no ISO-8859-1 default

316

317 header = headers.get('content-type')

318 if not header:

319 # https://www.w3.org/Protocols/rfc2616/rfc2616-sec7.html#sec7.2.1

320 return 'application/octet-stream', None

321

322 ctype, params = _parse_content_type_header(header)

323 if 'charset' not in params:

324 return ctype, None

325

326 # make sure this is a valid python encoding

327 enc = params['charset']

328 try:

329 str(b'.', encoding=enc, errors='strict')

330 except LookupError:

331 gws.log.warning(f'invalid content-type encoding {enc!r}')

332 return ctype, None

333

334 return ctype, enc

335

336

337##

338

339# @TODO locking for caches

340

341

342def http_request(url, **kwargs) -> HTTPResponse:

343 kwargs = dict(kwargs)

344

345 if 'params' in kwargs:

346 url = add_params(url, kwargs.pop('params'))

347

348 method = kwargs.pop('method', 'GET').upper()

349 max_age = kwargs.pop('max_age', 0)

350 cache_path = _cache_path(url)

351

352 if method == 'GET' and max_age:

353 age = gws.lib.osx.file_age(cache_path)

354 if 0 <= age < max_age:

355 gws.log.debug(f'HTTP_CACHED_{method}: url={url!r} path={cache_path!r} age={age}')

356 return gws.u.unserialize_from_path(cache_path)

357

358 gws.debug.time_start(f'HTTP_{method}={url!r}')

359 res = _http_request(method, url, kwargs)

360 gws.debug.time_end()

361

362 if method == 'GET' and max_age and res.ok:

363 gws.u.serialize_to_path(res, cache_path)

364

365 return res

366

367

368_DEFAULT_CONNECT_TIMEOUT = 60

369_DEFAULT_READ_TIMEOUT = 60

370

371_USER_AGENT = f'GBD WebSuite (https://gbd-websuite.de)'

372

373

374def _http_request(method, url, kwargs) -> HTTPResponse:

375 kwargs['stream'] = False

376

377 if 'verify' not in kwargs:

378 kwargs['verify'] = certifi.where()

379

380 timeout = kwargs.get('timeout', (_DEFAULT_CONNECT_TIMEOUT, _DEFAULT_READ_TIMEOUT))

381 if isinstance(timeout, (int, float)):

382 timeout = int(timeout), int(timeout)

383 kwargs['timeout'] = timeout

384

385 if 'headers' not in kwargs:

386 kwargs['headers'] = {}

387 kwargs['headers'].setdefault('User-Agent', _USER_AGENT)

388

389 try:

390 res = requests.request(method, url, **kwargs)

391 if 200 <= res.status_code < 300:

392 gws.log.debug(f'HTTP_OK_{method}: url={url!r} status={res.status_code!r}')

393 return HTTPResponse(ok=True, url=url, res=res)

394 gws.log.error(f'HTTP_FAILED_{method}: ({res.status_code!r}) url={url!r}')

395 return HTTPResponse(ok=False, url=url, res=res)

396 except requests.ConnectionError as exc:

397 gws.log.error(f'HTTP_FAILED_{method}: (ConnectionError) url={url!r}')

398 return HTTPResponse(ok=False, url=url, text=repr(exc), status_code=_STATUS_CONNECTION_ERROR)

399 except requests.Timeout as exc:

400 gws.log.error(f'HTTP_FAILED_{method}: (Timeout) url={url!r}')

401 return HTTPResponse(ok=False, url=url, text=repr(exc), status_code=_STATUS_TIMEOUT)

402 except requests.RequestException as exc:

403 gws.log.error(f'HTTP_FAILED_{method}: (Generic: {exc!r}) url={url!r}')

404 return HTTPResponse(ok=False, url=url, text=repr(exc), status_code=_STATUS_GENERIC_ERROR)

405

406

407def _cache_path(url):

408 return gws.c.NET_CACHE_DIR + '/' + gws.u.sha256(url)

Coverage for gws-app/gws/lib/net/init.py: 87%

254 statements