Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/httpx/_urls.py: 39%

1import typing 

2from urllib.parse import parse_qs, unquote 

3 

4import idna 

5 

6from ._types import QueryParamTypes, RawURL, URLTypes 

7from ._urlparse import urlencode, urlparse 

8from ._utils import primitive_value_to_str 

9 

10 

11class URL: 

12 """ 

13 url = httpx.URL("HTTPS://jo%40email.com:a%20secret@müller.de:1234/pa%20th?search=ab#anchorlink") 

14 

15 assert url.scheme == "https" 

16 assert url.username == "jo@email.com" 

17 assert url.password == "a secret" 

18 assert url.userinfo == b"jo%40email.com:a%20secret" 

19 assert url.host == "müller.de" 

20 assert url.raw_host == b"xn--mller-kva.de" 

21 assert url.port == 1234 

22 assert url.netloc == b"xn--mller-kva.de:1234" 

23 assert url.path == "/pa th" 

24 assert url.query == b"?search=ab" 

25 assert url.raw_path == b"/pa%20th?search=ab" 

26 assert url.fragment == "anchorlink" 

27 

28 The components of a URL are broken down like this: 

29 

30 https://jo%40email.com:a%20secret@müller.de:1234/pa%20th?search=ab#anchorlink 

31 [scheme] [ username ] [password] [ host ][port][ path ] [ query ] [fragment] 

32 [ userinfo ] [ netloc ][ raw_path ] 

33 

34 Note that: 

35 

36 * `url.scheme` is normalized to always be lowercased. 

37 

38 * `url.host` is normalized to always be lowercased. Internationalized domain 

39 names are represented in unicode, without IDNA encoding applied. For instance: 

40 

41 url = httpx.URL("http://中国.icom.museum") 

42 assert url.host == "中国.icom.museum" 

43 url = httpx.URL("http://xn--fiqs8s.icom.museum") 

44 assert url.host == "中国.icom.museum" 

45 

46 * `url.raw_host` is normalized to always be lowercased, and is IDNA encoded. 

47 

48 url = httpx.URL("http://中国.icom.museum") 

49 assert url.raw_host == b"xn--fiqs8s.icom.museum" 

50 url = httpx.URL("http://xn--fiqs8s.icom.museum") 

51 assert url.raw_host == b"xn--fiqs8s.icom.museum" 

52 

53 * `url.port` is either None or an integer. URLs that include the default port for 

54 "http", "https", "ws", "wss", and "ftp" schemes have their port normalized to `None`. 

55 

56 assert httpx.URL("http://example.com") == httpx.URL("http://example.com:80") 

57 assert httpx.URL("http://example.com").port is None 

58 assert httpx.URL("http://example.com:80").port is None 

59 

60 * `url.userinfo` is raw bytes, without URL escaping. Usually you'll want to work with 

61 `url.username` and `url.password` instead, which handle the URL escaping. 

62 

63 * `url.raw_path` is raw bytes of both the path and query, without URL escaping. 

64 This portion is used as the target when constructing HTTP requests. Usually you'll 

65 want to work with `url.path` instead. 

66 

67 * `url.query` is raw bytes, without URL escaping. A URL query string portion can only 

68 be properly URL escaped when decoding the parameter names and values themselves. 

69 """ 

70 

71 def __init__( 

72 self, url: typing.Union["URL", str] = "", **kwargs: typing.Any 

73 ) -> None: 

74 if kwargs: 

75 allowed = { 

76 "scheme": str, 

77 "username": str, 

78 "password": str, 

79 "userinfo": bytes, 

80 "host": str, 

81 "port": int, 

82 "netloc": bytes, 

83 "path": str, 

84 "query": bytes, 

85 "raw_path": bytes, 

86 "fragment": str, 

87 "params": object, 

88 } 

89 

90 # Perform type checking for all supported keyword arguments. 

91 for key, value in kwargs.items(): 

92 if key not in allowed: 

93 message = f"{key!r} is an invalid keyword argument for URL()" 

94 raise TypeError(message) 

95 if value is not None and not isinstance(value, allowed[key]): 

96 expected = allowed[key].__name__ 

97 seen = type(value).__name__ 

98 message = f"Argument {key!r} must be {expected} but got {seen}" 

99 raise TypeError(message) 

100 if isinstance(value, bytes): 

101 kwargs[key] = value.decode("ascii") 

102 

103 if "params" in kwargs: 

104 # Replace any "params" keyword with the raw "query" instead. 

105 # 

106 # Ensure that empty params use `kwargs["query"] = None` rather 

107 # than `kwargs["query"] = ""`, so that generated URLs do not 

108 # include an empty trailing "?". 

109 params = kwargs.pop("params") 

110 kwargs["query"] = None if not params else str(QueryParams(params)) 

111 

112 if isinstance(url, str): 

113 self._uri_reference = urlparse(url, **kwargs) 

114 elif isinstance(url, URL): 

115 self._uri_reference = url._uri_reference.copy_with(**kwargs) 

116 else: 

117 raise TypeError( 

118 f"Invalid type for url. Expected str or httpx.URL, got {type(url)}: {url!r}" 

119 ) 

120 

121 @property 

122 def scheme(self) -> str: 

123 """ 

124 The URL scheme, such as "http", "https". 

125 Always normalised to lowercase. 

126 """ 

127 return self._uri_reference.scheme 

128 

129 @property 

130 def raw_scheme(self) -> bytes: 

131 """ 

132 The raw bytes representation of the URL scheme, such as b"http", b"https". 

133 Always normalised to lowercase. 

134 """ 

135 return self._uri_reference.scheme.encode("ascii") 

136 

137 @property 

138 def userinfo(self) -> bytes: 

139 """ 

140 The URL userinfo as a raw bytestring. 

141 For example: b"jo%40email.com:a%20secret". 

142 """ 

143 return self._uri_reference.userinfo.encode("ascii") 

144 

145 @property 

146 def username(self) -> str: 

147 """ 

148 The URL username as a string, with URL decoding applied. 

149 For example: "jo@email.com" 

150 """ 

151 userinfo = self._uri_reference.userinfo 

152 return unquote(userinfo.partition(":")[0]) 

153 

154 @property 

155 def password(self) -> str: 

156 """ 

157 The URL password as a string, with URL decoding applied. 

158 For example: "a secret" 

159 """ 

160 userinfo = self._uri_reference.userinfo 

161 return unquote(userinfo.partition(":")[2]) 

162 

163 @property 

164 def host(self) -> str: 

165 """ 

166 The URL host as a string. 

167 Always normalized to lowercase, with IDNA hosts decoded into unicode. 

168 

169 Examples: 

170 

171 url = httpx.URL("http://www.EXAMPLE.org") 

172 assert url.host == "www.example.org" 

173 

174 url = httpx.URL("http://中国.icom.museum") 

175 assert url.host == "中国.icom.museum" 

176 

177 url = httpx.URL("http://xn--fiqs8s.icom.museum") 

178 assert url.host == "中国.icom.museum" 

179 

180 url = httpx.URL("https://[::ffff:192.168.0.1]") 

181 assert url.host == "::ffff:192.168.0.1" 

182 """ 

183 host: str = self._uri_reference.host 

184 

185 if host.startswith("xn--"): 

186 host = idna.decode(host) 

187 

188 return host 

189 

190 @property 

191 def raw_host(self) -> bytes: 

192 """ 

193 The raw bytes representation of the URL host. 

194 Always normalized to lowercase, and IDNA encoded. 

195 

196 Examples: 

197 

198 url = httpx.URL("http://www.EXAMPLE.org") 

199 assert url.raw_host == b"www.example.org" 

200 

201 url = httpx.URL("http://中国.icom.museum") 

202 assert url.raw_host == b"xn--fiqs8s.icom.museum" 

203 

204 url = httpx.URL("http://xn--fiqs8s.icom.museum") 

205 assert url.raw_host == b"xn--fiqs8s.icom.museum" 

206 

207 url = httpx.URL("https://[::ffff:192.168.0.1]") 

208 assert url.raw_host == b"::ffff:192.168.0.1" 

209 """ 

210 return self._uri_reference.host.encode("ascii") 

211 

212 @property 

213 def port(self) -> typing.Optional[int]: 

214 """ 

215 The URL port as an integer. 

216 

217 Note that the URL class performs port normalization as per the WHATWG spec. 

218 Default ports for "http", "https", "ws", "wss", and "ftp" schemes are always 

219 treated as `None`. 

220 

221 For example: 

222 

223 assert httpx.URL("http://www.example.com") == httpx.URL("http://www.example.com:80") 

224 assert httpx.URL("http://www.example.com:80").port is None 

225 """ 

226 return self._uri_reference.port 

227 

228 @property 

229 def netloc(self) -> bytes: 

230 """ 

231 Either `<host>` or `<host>:<port>` as bytes. 

232 Always normalized to lowercase, and IDNA encoded. 

233 

234 This property may be used for generating the value of a request 

235 "Host" header. 

236 """ 

237 return self._uri_reference.netloc.encode("ascii") 

238 

239 @property 

240 def path(self) -> str: 

241 """ 

242 The URL path as a string. Excluding the query string, and URL decoded. 

243 

244 For example: 

245 

246 url = httpx.URL("https://example.com/pa%20th") 

247 assert url.path == "/pa th" 

248 """ 

249 path = self._uri_reference.path or "/" 

250 return unquote(path) 

251 

252 @property 

253 def query(self) -> bytes: 

254 """ 

255 The URL query string, as raw bytes, excluding the leading b"?". 

256 

257 This is necessarily a bytewise interface, because we cannot 

258 perform URL decoding of this representation until we've parsed 

259 the keys and values into a QueryParams instance. 

260 

261 For example: 

262 

263 url = httpx.URL("https://example.com/?filter=some%20search%20terms") 

264 assert url.query == b"filter=some%20search%20terms" 

265 """ 

266 query = self._uri_reference.query or "" 

267 return query.encode("ascii") 

268 

269 @property 

270 def params(self) -> "QueryParams": 

271 """ 

272 The URL query parameters, neatly parsed and packaged into an immutable 

273 multidict representation. 

274 """ 

275 return QueryParams(self._uri_reference.query) 

276 

277 @property 

278 def raw_path(self) -> bytes: 

279 """ 

280 The complete URL path and query string as raw bytes. 

281 Used as the target when constructing HTTP requests. 

282 

283 For example: 

284 

285 GET /users?search=some%20text HTTP/1.1 

286 Host: www.example.org 

287 Connection: close 

288 """ 

289 path = self._uri_reference.path or "/" 

290 if self._uri_reference.query is not None: 

291 path += "?" + self._uri_reference.query 

292 return path.encode("ascii") 

293 

294 @property 

295 def fragment(self) -> str: 

296 """ 

297 The URL fragments, as used in HTML anchors. 

298 As a string, without the leading '#'. 

299 """ 

300 return unquote(self._uri_reference.fragment or "") 

301 

302 @property 

303 def raw(self) -> RawURL: 

304 """ 

305 Provides the (scheme, host, port, target) for the outgoing request. 

306 

307 In older versions of `httpx` this was used in the low-level transport API. 

308 We no longer use `RawURL`, and this property will be deprecated in a future release. 

309 """ 

310 return RawURL( 

311 self.raw_scheme, 

312 self.raw_host, 

313 self.port, 

314 self.raw_path, 

315 ) 

316 

317 @property 

318 def is_absolute_url(self) -> bool: 

319 """ 

320 Return `True` for absolute URLs such as 'http://example.com/path', 

321 and `False` for relative URLs such as '/path'. 

322 """ 

323 # We don't use `.is_absolute` from `rfc3986` because it treats 

324 # URLs with a fragment portion as not absolute. 

325 # What we actually care about is if the URL provides 

326 # a scheme and hostname to which connections should be made. 

327 return bool(self._uri_reference.scheme and self._uri_reference.host) 

328 

329 @property 

330 def is_relative_url(self) -> bool: 

331 """ 

332 Return `False` for absolute URLs such as 'http://example.com/path', 

333 and `True` for relative URLs such as '/path'. 

334 """ 

335 return not self.is_absolute_url 

336 

337 def copy_with(self, **kwargs: typing.Any) -> "URL": 

338 """ 

339 Copy this URL, returning a new URL with some components altered. 

340 Accepts the same set of parameters as the components that are made 

341 available via properties on the `URL` class. 

342 

343 For example: 

344 

345 url = httpx.URL("https://www.example.com").copy_with(username="jo@gmail.com", password="a secret") 

346 assert url == "https://jo%40email.com:a%20secret@www.example.com" 

347 """ 

348 return URL(self, **kwargs) 

349 

350 def copy_set_param(self, key: str, value: typing.Any = None) -> "URL": 

351 return self.copy_with(params=self.params.set(key, value)) 

352 

353 def copy_add_param(self, key: str, value: typing.Any = None) -> "URL": 

354 return self.copy_with(params=self.params.add(key, value)) 

355 

356 def copy_remove_param(self, key: str) -> "URL": 

357 return self.copy_with(params=self.params.remove(key)) 

358 

359 def copy_merge_params(self, params: QueryParamTypes) -> "URL": 

360 return self.copy_with(params=self.params.merge(params)) 

361 

362 def join(self, url: URLTypes) -> "URL": 

363 """ 

364 Return an absolute URL, using this URL as the base. 

365 

366 Eg. 

367 

368 url = httpx.URL("https://www.example.com/test") 

369 url = url.join("/new/path") 

370 assert url == "https://www.example.com/new/path" 

371 """ 

372 from urllib.parse import urljoin 

373 

374 return URL(urljoin(str(self), str(URL(url)))) 

375 

376 def __hash__(self) -> int: 

377 return hash(str(self)) 

378 

379 def __eq__(self, other: typing.Any) -> bool: 

380 return isinstance(other, (URL, str)) and str(self) == str(URL(other)) 

381 

382 def __str__(self) -> str: 

383 return str(self._uri_reference) 

384 

385 def __repr__(self) -> str: 

386 scheme, userinfo, host, port, path, query, fragment = self._uri_reference 

387 

388 if ":" in userinfo: 

389 # Mask any password component. 

390 userinfo = f'{userinfo.split(":")[0]}:[secure]' 

391 

392 authority = "".join( 

393 [ 

394 f"{userinfo}@" if userinfo else "", 

395 f"[{host}]" if ":" in host else host, 

396 f":{port}" if port is not None else "", 

397 ] 

398 ) 

399 url = "".join( 

400 [ 

401 f"{self.scheme}:" if scheme else "", 

402 f"//{authority}" if authority else "", 

403 path, 

404 f"?{query}" if query is not None else "", 

405 f"#{fragment}" if fragment is not None else "", 

406 ] 

407 ) 

408 

409 return f"{self.__class__.__name__}({url!r})" 

410 

411 

412class QueryParams(typing.Mapping[str, str]): 

413 """ 

414 URL query parameters, as a multi-dict. 

415 """ 

416 

417 def __init__( 

418 self, *args: typing.Optional[QueryParamTypes], **kwargs: typing.Any 

419 ) -> None: 

420 assert len(args) < 2, "Too many arguments." 

421 assert not (args and kwargs), "Cannot mix named and unnamed arguments." 

422 

423 value = args[0] if args else kwargs 

424 

425 if value is None or isinstance(value, (str, bytes)): 

426 value = value.decode("ascii") if isinstance(value, bytes) else value 

427 self._dict = parse_qs(value, keep_blank_values=True) 

428 elif isinstance(value, QueryParams): 

429 self._dict = {k: list(v) for k, v in value._dict.items()} 

430 else: 

431 dict_value: typing.Dict[typing.Any, typing.List[typing.Any]] = {} 

432 if isinstance(value, (list, tuple)): 

433 # Convert list inputs like: 

434 # [("a", "123"), ("a", "456"), ("b", "789")] 

435 # To a dict representation, like: 

436 # {"a": ["123", "456"], "b": ["789"]} 

437 for item in value: 

438 dict_value.setdefault(item[0], []).append(item[1]) 

439 else: 

440 # Convert dict inputs like: 

441 # {"a": "123", "b": ["456", "789"]} 

442 # To dict inputs where values are always lists, like: 

443 # {"a": ["123"], "b": ["456", "789"]} 

444 dict_value = { 

445 k: list(v) if isinstance(v, (list, tuple)) else [v] 

446 for k, v in value.items() 

447 } 

448 

449 # Ensure that keys and values are neatly coerced to strings. 

450 # We coerce values `True` and `False` to JSON-like "true" and "false" 

451 # representations, and coerce `None` values to the empty string. 

452 self._dict = { 

453 str(k): [primitive_value_to_str(item) for item in v] 

454 for k, v in dict_value.items() 

455 } 

456 

457 def keys(self) -> typing.KeysView[str]: 

458 """ 

459 Return all the keys in the query params. 

460 

461 Usage: 

462 

463 q = httpx.QueryParams("a=123&a=456&b=789") 

464 assert list(q.keys()) == ["a", "b"] 

465 """ 

466 return self._dict.keys() 

467 

468 def values(self) -> typing.ValuesView[str]: 

469 """ 

470 Return all the values in the query params. If a key occurs more than once 

471 only the first item for that key is returned. 

472 

473 Usage: 

474 

475 q = httpx.QueryParams("a=123&a=456&b=789") 

476 assert list(q.values()) == ["123", "789"] 

477 """ 

478 return {k: v[0] for k, v in self._dict.items()}.values() 

479 

480 def items(self) -> typing.ItemsView[str, str]: 

481 """ 

482 Return all items in the query params. If a key occurs more than once 

483 only the first item for that key is returned. 

484 

485 Usage: 

486 

487 q = httpx.QueryParams("a=123&a=456&b=789") 

488 assert list(q.items()) == [("a", "123"), ("b", "789")] 

489 """ 

490 return {k: v[0] for k, v in self._dict.items()}.items() 

491 

492 def multi_items(self) -> typing.List[typing.Tuple[str, str]]: 

493 """ 

494 Return all items in the query params. Allow duplicate keys to occur. 

495 

496 Usage: 

497 

498 q = httpx.QueryParams("a=123&a=456&b=789") 

499 assert list(q.multi_items()) == [("a", "123"), ("a", "456"), ("b", "789")] 

500 """ 

501 multi_items: typing.List[typing.Tuple[str, str]] = [] 

502 for k, v in self._dict.items(): 

503 multi_items.extend([(k, i) for i in v]) 

504 return multi_items 

505 

506 def get(self, key: typing.Any, default: typing.Any = None) -> typing.Any: 

507 """ 

508 Get a value from the query param for a given key. If the key occurs 

509 more than once, then only the first value is returned. 

510 

511 Usage: 

512 

513 q = httpx.QueryParams("a=123&a=456&b=789") 

514 assert q.get("a") == "123" 

515 """ 

516 if key in self._dict: 

517 return self._dict[str(key)][0] 

518 return default 

519 

520 def get_list(self, key: str) -> typing.List[str]: 

521 """ 

522 Get all values from the query param for a given key. 

523 

524 Usage: 

525 

526 q = httpx.QueryParams("a=123&a=456&b=789") 

527 assert q.get_list("a") == ["123", "456"] 

528 """ 

529 return list(self._dict.get(str(key), [])) 

530 

531 def set(self, key: str, value: typing.Any = None) -> "QueryParams": 

532 """ 

533 Return a new QueryParams instance, setting the value of a key. 

534 

535 Usage: 

536 

537 q = httpx.QueryParams("a=123") 

538 q = q.set("a", "456") 

539 assert q == httpx.QueryParams("a=456") 

540 """ 

541 q = QueryParams() 

542 q._dict = dict(self._dict) 

543 q._dict[str(key)] = [primitive_value_to_str(value)] 

544 return q 

545 

546 def add(self, key: str, value: typing.Any = None) -> "QueryParams": 

547 """ 

548 Return a new QueryParams instance, setting or appending the value of a key. 

549 

550 Usage: 

551 

552 q = httpx.QueryParams("a=123") 

553 q = q.add("a", "456") 

554 assert q == httpx.QueryParams("a=123&a=456") 

555 """ 

556 q = QueryParams() 

557 q._dict = dict(self._dict) 

558 q._dict[str(key)] = q.get_list(key) + [primitive_value_to_str(value)] 

559 return q 

560 

561 def remove(self, key: str) -> "QueryParams": 

562 """ 

563 Return a new QueryParams instance, removing the value of a key. 

564 

565 Usage: 

566 

567 q = httpx.QueryParams("a=123") 

568 q = q.remove("a") 

569 assert q == httpx.QueryParams("") 

570 """ 

571 q = QueryParams() 

572 q._dict = dict(self._dict) 

573 q._dict.pop(str(key), None) 

574 return q 

575 

576 def merge(self, params: typing.Optional[QueryParamTypes] = None) -> "QueryParams": 

577 """ 

578 Return a new QueryParams instance, updated with. 

579 

580 Usage: 

581 

582 q = httpx.QueryParams("a=123") 

583 q = q.merge({"b": "456"}) 

584 assert q == httpx.QueryParams("a=123&b=456") 

585 

586 q = httpx.QueryParams("a=123") 

587 q = q.merge({"a": "456", "b": "789"}) 

588 assert q == httpx.QueryParams("a=456&b=789") 

589 """ 

590 q = QueryParams(params) 

591 q._dict = {**self._dict, **q._dict} 

592 return q 

593 

594 def __getitem__(self, key: typing.Any) -> str: 

595 return self._dict[key][0] 

596 

597 def __contains__(self, key: typing.Any) -> bool: 

598 return key in self._dict 

599 

600 def __iter__(self) -> typing.Iterator[typing.Any]: 

601 return iter(self.keys()) 

602 

603 def __len__(self) -> int: 

604 return len(self._dict) 

605 

606 def __bool__(self) -> bool: 

607 return bool(self._dict) 

608 

609 def __hash__(self) -> int: 

610 return hash(str(self)) 

611 

612 def __eq__(self, other: typing.Any) -> bool: 

613 if not isinstance(other, self.__class__): 

614 return False 

615 return sorted(self.multi_items()) == sorted(other.multi_items()) 

616 

617 def __str__(self) -> str: 

618 """ 

619 Note that we use '%20' encoding for spaces, and treat '/' as a safe 

620 character. 

621 

622 See https://github.com/encode/httpx/issues/2536 and 

623 https://docs.python.org/3/library/urllib.parse.html#urllib.parse.urlencode 

624 """ 

625 return urlencode(self.multi_items()) 

626 

627 def __repr__(self) -> str: 

628 class_name = self.__class__.__name__ 

629 query_string = str(self) 

630 return f"{class_name}({query_string!r})" 

631 

632 def update(self, params: typing.Optional[QueryParamTypes] = None) -> None: 

633 raise RuntimeError( 

634 "QueryParams are immutable since 0.18.0. " 

635 "Use `q = q.merge(...)` to create an updated copy." 

636 ) 

637 

638 def __setitem__(self, key: str, value: str) -> None: 

639 raise RuntimeError( 

640 "QueryParams are immutable since 0.18.0. " 

641 "Use `q = q.set(key, value)` to create an updated copy." 

642 )