Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/uritools/__init__.py: 38%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

421 statements  

1"""RFC 3986 compliant, scheme-agnostic replacement for `urllib.parse`. 

2 

3This module defines RFC 3986 compliant replacements for the most 

4commonly used functions of the Python Standard Library 

5:mod:`urllib.parse` module. 

6 

7""" 

8 

9import collections 

10import collections.abc 

11import functools 

12import ipaddress 

13import numbers 

14import re 

15from string import hexdigits 

16 

17__all__ = ( 

18 "GEN_DELIMS", 

19 "RESERVED", 

20 "SUB_DELIMS", 

21 "UNRESERVED", 

22 "isabspath", 

23 "isabsuri", 

24 "isnetpath", 

25 "isrelpath", 

26 "issamedoc", 

27 "isuri", 

28 "uricompose", 

29 "uridecode", 

30 "uridefrag", 

31 "uriencode", 

32 "urijoin", 

33 "urisplit", 

34 "uriunsplit", 

35) 

36 

37__version__ = "7.0.0" 

38 

39 

40# RFC 3986 2.2. Reserved Characters 

41# 

42# reserved = gen-delims / sub-delims 

43# 

44# gen-delims = ":" / "/" / "?" / "#" / "[" / "]" / "@" 

45# 

46# sub-delims = "!" / "$" / "&" / "'" / "(" / ")" 

47# / "*" / "+" / "," / ";" / "=" 

48# 

49GEN_DELIMS = ":/?#[]@" 

50SUB_DELIMS = "!$&'()*+,;=" 

51RESERVED = GEN_DELIMS + SUB_DELIMS 

52 

53# RFC 3986 2.3. Unreserved Characters 

54# 

55# unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~" 

56# 

57UNRESERVED = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~" 

58 

59_unreserved = frozenset(UNRESERVED.encode()) 

60 

61# RFC 3986 2.1: For consistency, URI producers and normalizers should 

62# use uppercase hexadecimal digits for all percent-encodings. 

63_encoded = [ 

64 bytes([i]) if i in _unreserved else ("%%%02X" % i).encode() for i in range(256) 

65] 

66 

67_decoded = { 

68 (a + b).encode(): bytes.fromhex(a + b) for a in hexdigits for b in hexdigits 

69} 

70 

71 

72@functools.lru_cache 

73def _get_encoded(safe): 

74 # one lookup table per "safe" value; cache is bounded and thread-safe 

75 table = _encoded[:] 

76 for i in safe: 

77 table[i] = bytes([i]) 

78 return table 

79 

80 

81def uriencode(uristring, safe="", encoding="utf-8", errors="strict"): 

82 """Encode a URI string or string component.""" 

83 if isinstance(uristring, str): 

84 uristring = uristring.encode(encoding, errors) 

85 elif not isinstance(uristring, bytes): 

86 raise TypeError("Invalid URI string type: %r" % type(uristring)) 

87 if isinstance(safe, str): 

88 safe = safe.encode("ascii") 

89 return b"".join(map(_get_encoded(safe).__getitem__, uristring)) 

90 

91 

92def uridecode(uristring, encoding="utf-8", errors="strict"): 

93 """Decode a URI string or string component.""" 

94 if not isinstance(uristring, bytes): 

95 uristring = uristring.encode(encoding or "ascii", errors) 

96 parts = uristring.split(b"%") 

97 result = [parts[0]] 

98 append = result.append 

99 decode = _decoded.get 

100 for s in parts[1:]: 

101 append(decode(s[:2], b"%" + s[:2])) 

102 append(s[2:]) 

103 if encoding is not None: 

104 return b"".join(result).decode(encoding, errors) 

105 else: 

106 return b"".join(result) 

107 

108 

109class DefragResult(collections.namedtuple("DefragResult", "uri fragment")): 

110 """Class to hold :func:`uridefrag` results.""" 

111 

112 __slots__ = () # prevent creation of instance dictionary 

113 

114 def geturi(self): 

115 """Return the recombined version of the original URI as a string.""" 

116 fragment = self.fragment 

117 if fragment is None: 

118 return self.uri 

119 elif isinstance(fragment, bytes): 

120 return self.uri + b"#" + fragment 

121 else: 

122 return self.uri + "#" + fragment 

123 

124 def getfragment(self, default=None, encoding="utf-8", errors="strict"): 

125 """Return the decoded fragment identifier, or `default` if the 

126 original URI did not contain a fragment component. 

127 

128 """ 

129 # FIXME: by default, getfragment() should return bytes if geturi() returns bytes 

130 fragment = self.fragment 

131 if fragment is not None: 

132 return uridecode(fragment, encoding, errors) 

133 else: 

134 return default 

135 

136 

137class SplitResult( 

138 collections.namedtuple("SplitResult", "scheme authority path query fragment") 

139): 

140 """Base class to hold :func:`urisplit` results.""" 

141 

142 __slots__ = () # prevent creation of instance dictionary 

143 

144 # regex and delimiter constants, defined by SplitResultBytes and SplitResultString 

145 _RE: re.Pattern 

146 _COLON: str | bytes 

147 _SLASH: str | bytes 

148 _QUEST: str | bytes 

149 _HASH: str | bytes 

150 _LBRACKET: str | bytes 

151 _RBRACKET: str | bytes 

152 _AT: str | bytes 

153 _DOT: str | bytes 

154 _DOTDOT: str | bytes 

155 _EMPTY: str | bytes 

156 _EQ: str | bytes 

157 _DIGITS: str | bytes 

158 

159 @property 

160 def userinfo(self): 

161 authority = self.authority 

162 if authority is None: 

163 return None 

164 userinfo, present, _ = authority.rpartition(self._AT) 

165 if present: 

166 return userinfo 

167 else: 

168 return None 

169 

170 @property 

171 def host(self): 

172 authority = self.authority 

173 if authority is None: 

174 return None 

175 _, _, hostinfo = authority.rpartition(self._AT) 

176 host, present, port = hostinfo.rpartition(self._COLON) 

177 if not present or port.lstrip(self._DIGITS): 

178 return hostinfo 

179 else: 

180 return host 

181 

182 @property 

183 def port(self): 

184 authority = self.authority 

185 if authority is None: 

186 return None 

187 _, present, port = authority.rpartition(self._COLON) 

188 if present and not port.lstrip(self._DIGITS): 

189 return port 

190 else: 

191 return None 

192 

193 def geturi(self): 

194 """Return the re-combined version of the original URI reference as a 

195 string. 

196 

197 """ 

198 scheme, authority, path, query, fragment = self 

199 

200 # RFC 3986 5.3. Component Recomposition 

201 result = [] 

202 if scheme is not None: 

203 result.extend([scheme, self._COLON]) 

204 if authority is not None: 

205 result.extend([self._SLASH, self._SLASH, authority]) 

206 result.append(path) 

207 if query is not None: 

208 result.extend([self._QUEST, query]) 

209 if fragment is not None: 

210 result.extend([self._HASH, fragment]) 

211 return self._EMPTY.join(result) 

212 

213 def getscheme(self, default=None): 

214 """Return the URI scheme in canonical (lowercase) form, or `default` 

215 if the original URI reference did not contain a scheme component. 

216 

217 """ 

218 # FIXME: should getscheme() return bytes if geturi() returns bytes? 

219 scheme = self.scheme 

220 if scheme is None: 

221 return default 

222 elif isinstance(scheme, bytes): 

223 return scheme.decode("ascii").lower() 

224 else: 

225 return scheme.lower() 

226 

227 def getauthority(self, default=None, encoding="utf-8", errors="strict"): 

228 """Return the decoded userinfo, host and port subcomponents of the URI 

229 authority as a three-item tuple. 

230 

231 """ 

232 # TBD: (userinfo, host, port) kwargs, default string? 

233 if default is None: 

234 default = (None, None, None) 

235 elif not isinstance(default, collections.abc.Sequence): 

236 raise TypeError("Invalid default type") 

237 elif len(default) != 3: 

238 raise ValueError("Invalid default length") 

239 # TODO: this could be much more efficient by using a dedicated regex 

240 return ( 

241 self.getuserinfo(default[0], encoding, errors), 

242 self.gethost(default[1], errors), 

243 self.getport(default[2]), 

244 ) 

245 

246 def getuserinfo(self, default=None, encoding="utf-8", errors="strict"): 

247 """Return the decoded userinfo subcomponent of the URI authority, or 

248 `default` if the original URI reference did not contain a 

249 userinfo field. 

250 

251 """ 

252 userinfo = self.userinfo 

253 if userinfo is None: 

254 return default 

255 else: 

256 return uridecode(userinfo, encoding, errors) 

257 

258 def gethost(self, default=None, errors="strict"): 

259 """Return the decoded host subcomponent of the URI authority as a 

260 string or an :mod:`ipaddress` address object, or `default` if 

261 the original URI reference did not contain a host. 

262 

263 """ 

264 host = self.host 

265 if host is None or (not host and default is not None): 

266 return default 

267 elif host.startswith(self._LBRACKET) and host.endswith(self._RBRACKET): 

268 return self.__parse_ip_literal(host[1:-1]) 

269 elif host.startswith(self._LBRACKET) or host.endswith(self._RBRACKET): 

270 raise ValueError("Invalid host %r: mismatched brackets" % host) 

271 # TODO: faster check for IPv4 address? 

272 try: 

273 if isinstance(host, bytes): 

274 return ipaddress.IPv4Address(host.decode("ascii")) 

275 else: 

276 return ipaddress.IPv4Address(host) 

277 except ValueError: 

278 # RFC 3986 3.2.2. Host: Non-ASCII characters must first be 

279 # encoded according to UTF-8 [STD63], and then each octet 

280 # of the corresponding UTF-8 sequence must be 

281 # percent-encoded to be represented as URI characters. 

282 return uridecode(host, "utf-8", errors).lower() 

283 

284 def getport(self, default=None): 

285 """Return the port subcomponent of the URI authority as an 

286 :class:`int`, or `default` if the original URI reference did 

287 not contain a port or if the port was empty. 

288 

289 """ 

290 port = self.port 

291 if port: 

292 return int(port) 

293 else: 

294 return default 

295 

296 def getpath(self, encoding="utf-8", errors="strict"): 

297 """Return the normalized decoded URI path.""" 

298 path = self.__remove_dot_segments(self.path) 

299 return uridecode(path, encoding, errors) 

300 

301 def getquery(self, default=None, encoding="utf-8", errors="strict"): 

302 """Return the decoded query string, or `default` if the original URI 

303 reference did not contain a query component. 

304 

305 """ 

306 query = self.query 

307 if query is None: 

308 return default 

309 else: 

310 return uridecode(query, encoding, errors) 

311 

312 def getquerydict(self, sep="&", encoding="utf-8", errors="strict"): 

313 """Split the query component into individual `name=value` pairs 

314 separated by `sep` and return a dictionary of query variables. 

315 The dictionary keys are the unique query variable names and 

316 the values are lists of values for each name. 

317 

318 """ 

319 result = collections.defaultdict(list) 

320 for name, value in self.getquerylist(sep, encoding, errors): 

321 result[name].append(value) 

322 return result 

323 

324 def getquerylist(self, sep="&", encoding="utf-8", errors="strict"): 

325 """Split the query component into individual `name=value` pairs 

326 separated by `sep`, and return a list of `(name, value)` 

327 tuples. 

328 

329 """ 

330 if not self.query: 

331 return [] 

332 elif isinstance(sep, type(self.query)): 

333 qsl = self.query.split(sep) 

334 elif isinstance(sep, bytes): 

335 qsl = self.query.split(sep.decode("ascii")) 

336 else: 

337 qsl = self.query.split(sep.encode("ascii")) 

338 result = [] 

339 for parts in [qs.partition(self._EQ) for qs in qsl if qs]: 

340 name = uridecode(parts[0], encoding, errors) 

341 if parts[1]: 

342 value = uridecode(parts[2], encoding, errors) 

343 else: 

344 value = None 

345 result.append((name, value)) 

346 return result 

347 

348 def getfragment(self, default=None, encoding="utf-8", errors="strict"): 

349 """Return the decoded fragment identifier, or `default` if the 

350 original URI reference did not contain a fragment component. 

351 

352 """ 

353 fragment = self.fragment 

354 if fragment is None: 

355 return default 

356 else: 

357 return uridecode(fragment, encoding, errors) 

358 

359 def isuri(self): 

360 """Return :const:`True` if this is a URI.""" 

361 return self.scheme is not None 

362 

363 def isabsuri(self): 

364 """Return :const:`True` if this is an absolute URI.""" 

365 return self.scheme is not None and self.fragment is None 

366 

367 def isnetpath(self): 

368 """Return :const:`True` if this is a network-path reference.""" 

369 return self.scheme is None and self.authority is not None 

370 

371 def isabspath(self): 

372 """Return :const:`True` if this is an absolute-path reference.""" 

373 return ( 

374 self.scheme is None 

375 and self.authority is None 

376 and self.path.startswith(self._SLASH) 

377 ) 

378 

379 def isrelpath(self): 

380 """Return :const:`True` if this is a relative-path reference.""" 

381 return ( 

382 self.scheme is None 

383 and self.authority is None 

384 and not self.path.startswith(self._SLASH) 

385 ) 

386 

387 def issamedoc(self): 

388 """Return :const:`True` if this is a same-document reference.""" 

389 return ( 

390 self.scheme is None 

391 and self.authority is None 

392 and not self.path 

393 and self.query is None 

394 ) 

395 

396 def transform(self, ref, strict=False): 

397 """Transform a URI reference relative to `self` into a 

398 :class:`SplitResult` representing its target URI. 

399 

400 """ 

401 scheme, authority, path, query, fragment = self._match(ref).groups() 

402 

403 # RFC 3986 5.2.2. Transform References; scheme names are 

404 # case-insensitive (RFC 3986 3.1) 

405 if scheme is not None and ( 

406 strict or self.scheme is None or scheme.lower() != self.scheme.lower() 

407 ): 

408 path = self.__remove_dot_segments(path) 

409 elif authority is not None: 

410 scheme = self.scheme 

411 path = self.__remove_dot_segments(path) 

412 elif not path: 

413 scheme = self.scheme 

414 authority = self.authority 

415 path = self.path 

416 query = self.query if query is None else query 

417 elif path.startswith(self._SLASH): 

418 scheme = self.scheme 

419 authority = self.authority 

420 path = self.__remove_dot_segments(path) 

421 else: 

422 scheme = self.scheme 

423 authority = self.authority 

424 path = self.__remove_dot_segments(self.__merge(path)) 

425 return type(self)(scheme, authority, path, query, fragment) 

426 

427 @classmethod 

428 def _match(cls, ref): 

429 return cls._RE.match(ref) 

430 

431 def __merge(self, path): 

432 # RFC 3986 5.2.3. Merge Paths 

433 if self.authority is not None and not self.path: 

434 return self._SLASH + path 

435 else: 

436 parts = self.path.rpartition(self._SLASH) 

437 return parts[1].join((parts[0], path)) 

438 

439 @classmethod 

440 def __remove_dot_segments(cls, path): 

441 # RFC 3986 5.2.4. Remove Dot Segments 

442 pseg = [] 

443 for s in path.split(cls._SLASH): 

444 if s == cls._DOT: 

445 continue 

446 elif s != cls._DOTDOT: 

447 pseg.append(s) 

448 elif len(pseg) == 1 and not pseg[0]: 

449 continue 

450 elif pseg and pseg[-1] != cls._DOTDOT: 

451 pseg.pop() 

452 else: 

453 pseg.append(s) 

454 # adjust for trailing '/.' or '/..' 

455 if path.rpartition(cls._SLASH)[2] in (cls._DOT, cls._DOTDOT): 

456 pseg.append(cls._EMPTY) 

457 if path and len(pseg) == 1 and pseg[0] == cls._EMPTY: 

458 pseg.insert(0, cls._DOT) 

459 return cls._SLASH.join(pseg) 

460 

461 @classmethod 

462 def __parse_ip_literal(cls, address): 

463 # RFC 3986 3.2.2: In anticipation of future, as-yet-undefined 

464 # IP literal address formats, an implementation may use an 

465 # optional version flag to indicate such a format explicitly 

466 # rather than rely on heuristic determination. 

467 # 

468 # IP-literal = "[" ( IPv6address / IPvFuture ) "]" 

469 # 

470 # IPvFuture = "v" 1*HEXDIG "." 1*( unreserved / sub-delims / ":" ) 

471 # 

472 # If a URI containing an IP-literal that starts with "v" 

473 # (case-insensitive), indicating that the version flag is 

474 # present, is dereferenced by an application that does not 

475 # know the meaning of that version flag, then the application 

476 # should return an appropriate error for "address mechanism 

477 # not supported". 

478 if isinstance(address, bytes): 

479 address = address.decode("ascii") 

480 if address.startswith(("v", "V")): 

481 raise ValueError("Address mechanism not supported") 

482 return ipaddress.IPv6Address(address) 

483 

484 

485# TODO: make private? 

486class SplitResultBytes(SplitResult): 

487 __slots__ = () # prevent creation of instance dictionary 

488 

489 # RFC 3986 Appendix B 

490 _RE = re.compile( 

491 rb""" 

492 (?:([A-Za-z][A-Za-z0-9+.-]*):)? # scheme (RFC 3986 3.1) 

493 (?://([^/?#]*))? # authority 

494 ([^?#]*) # path 

495 (?:\?([^#]*))? # query 

496 (?:\#(.*))? # fragment 

497 """, 

498 flags=re.VERBOSE | re.DOTALL, 

499 ) 

500 

501 # RFC 3986 2.2 gen-delims 

502 _COLON, _SLASH, _QUEST, _HASH, _LBRACKET, _RBRACKET, _AT = ( 

503 b":", 

504 b"/", 

505 b"?", 

506 b"#", 

507 b"[", 

508 b"]", 

509 b"@", 

510 ) 

511 

512 # RFC 3986 3.3 dot-segments 

513 _DOT, _DOTDOT = b".", b".." 

514 

515 _EMPTY, _EQ = b"", b"=" 

516 

517 _DIGITS = b"0123456789" 

518 

519 

520# TODO: make private? 

521class SplitResultString(SplitResult): 

522 __slots__ = () # prevent creation of instance dictionary 

523 

524 # RFC 3986 Appendix B 

525 _RE = re.compile( 

526 r""" 

527 (?:([A-Za-z][A-Za-z0-9+.-]*):)? # scheme (RFC 3986 3.1) 

528 (?://([^/?#]*))? # authority 

529 ([^?#]*) # path 

530 (?:\?([^#]*))? # query 

531 (?:\#(.*))? # fragment 

532 """, 

533 flags=re.VERBOSE | re.DOTALL, 

534 ) 

535 

536 # RFC 3986 2.2 gen-delims 

537 _COLON, _SLASH, _QUEST, _HASH, _LBRACKET, _RBRACKET, _AT = ( 

538 ":", 

539 "/", 

540 "?", 

541 "#", 

542 "[", 

543 "]", 

544 "@", 

545 ) 

546 

547 # RFC 3986 3.3 dot-segments 

548 _DOT, _DOTDOT = ".", ".." 

549 

550 _EMPTY, _EQ = "", "=" 

551 

552 _DIGITS = "0123456789" 

553 

554 

555def uridefrag(uristring): 

556 """Remove an existing fragment component from a URI reference string.""" 

557 if isinstance(uristring, bytes): 

558 parts = uristring.partition(b"#") 

559 else: 

560 parts = uristring.partition("#") 

561 return DefragResult(parts[0], parts[2] if parts[1] else None) 

562 

563 

564def urisplit(uristring): 

565 """Split a well-formed URI reference string into a tuple with five 

566 components corresponding to a URI's general structure:: 

567 

568 <scheme>://<authority>/<path>?<query>#<fragment> 

569 

570 """ 

571 if isinstance(uristring, bytes): 

572 result = SplitResultBytes 

573 else: 

574 result = SplitResultString 

575 return result(*result._match(uristring).groups()) 

576 

577 

578def uriunsplit(parts): 

579 """Combine the elements of a five-item iterable into a URI reference's 

580 string representation. 

581 

582 """ 

583 scheme, authority, path, query, fragment = parts 

584 if isinstance(path, bytes): 

585 result = SplitResultBytes 

586 else: 

587 result = SplitResultString 

588 return result(scheme, authority, path, query, fragment).geturi() 

589 

590 

591def urijoin(base, ref, strict=False): 

592 """Convert a URI reference relative to a base URI to its target URI 

593 string. 

594 

595 """ 

596 if isinstance(base, type(ref)): 

597 return urisplit(base).transform(ref, strict).geturi() 

598 elif isinstance(base, bytes): 

599 return urisplit(base.decode()).transform(ref, strict).geturi() 

600 else: 

601 return urisplit(base).transform(ref.decode(), strict).geturi() 

602 

603 

604def isuri(uristring): 

605 """Return :const:`True` if `uristring` is a URI.""" 

606 return urisplit(uristring).isuri() 

607 

608 

609def isabsuri(uristring): 

610 """Return :const:`True` if `uristring` is an absolute URI.""" 

611 return urisplit(uristring).isabsuri() 

612 

613 

614def isnetpath(uristring): 

615 """Return :const:`True` if `uristring` is a network-path reference.""" 

616 return urisplit(uristring).isnetpath() 

617 

618 

619def isabspath(uristring): 

620 """Return :const:`True` if `uristring` is an absolute-path reference.""" 

621 return urisplit(uristring).isabspath() 

622 

623 

624def isrelpath(uristring): 

625 """Return :const:`True` if `uristring` is a relative-path reference.""" 

626 return urisplit(uristring).isrelpath() 

627 

628 

629def issamedoc(uristring): 

630 """Return :const:`True` if `uristring` is a same-document reference.""" 

631 return urisplit(uristring).issamedoc() 

632 

633 

634# RFC 3986 3.1: scheme = ALPHA *( ALPHA / DIGIT / "+" / "-" / "." ) 

635_SCHEME_RE = re.compile(b"[A-Za-z][A-Za-z0-9+.-]*") 

636 

637# safe component characters 

638_SAFE_USERINFO = SUB_DELIMS + ":" 

639_SAFE_HOST = SUB_DELIMS 

640_SAFE_PATH = SUB_DELIMS + ":@/" 

641_SAFE_QUERY = SUB_DELIMS + ":@/?" 

642_SAFE_FRAGMENT = SUB_DELIMS + ":@/?" 

643 

644 

645def _scheme(scheme): 

646 if _SCHEME_RE.fullmatch(scheme): 

647 return scheme.lower() 

648 else: 

649 raise ValueError("Invalid scheme component") 

650 

651 

652def _splitauthority(authority): 

653 # RFC 3986 3.2: authority = [ userinfo "@" ] host [ ":" port ] 

654 if isinstance(authority, bytes): 

655 userinfo, present, hostport = authority.rpartition(b"@") 

656 host, sep, port = hostport.rpartition(b":") 

657 numeric = sep and not port.lstrip(b"0123456789") 

658 else: 

659 userinfo, present, hostport = authority.rpartition("@") 

660 host, sep, port = hostport.rpartition(":") 

661 numeric = sep and not port.lstrip("0123456789") 

662 if not numeric: 

663 host, port = hostport, None 

664 return (userinfo if present else None, host, port) 

665 

666 

667def _authority(userinfo, host, port, encoding): 

668 authority = [] 

669 

670 if userinfo is not None: 

671 authority.append(uriencode(userinfo, _SAFE_USERINFO, encoding)) 

672 authority.append(b"@") 

673 

674 if isinstance(host, ipaddress.IPv6Address): 

675 authority.append(b"[" + host.compressed.encode() + b"]") 

676 elif isinstance(host, ipaddress.IPv4Address): 

677 authority.append(host.compressed.encode()) 

678 elif isinstance(host, bytes): 

679 authority.append(_host(host)) 

680 elif isinstance(host, str): 

681 authority.append(_host(host.encode("utf-8"))) 

682 elif host is not None: 

683 raise TypeError("Invalid host type: %r" % type(host)) 

684 

685 if isinstance(port, numbers.Number): 

686 authority.append(_port(str(port).encode())) 

687 elif isinstance(port, bytes): 

688 authority.append(_port(port)) 

689 elif isinstance(port, str): 

690 authority.append(_port(port.encode())) 

691 elif port is not None: 

692 raise TypeError("Invalid port type: %r" % type(port)) 

693 

694 return b"".join(authority) if authority else None 

695 

696 

697def _ip_literal(address): 

698 if address.startswith(("v", "V")): 

699 raise ValueError("Address mechanism not supported") 

700 else: 

701 return b"[" + ipaddress.IPv6Address(address).compressed.encode() + b"]" 

702 

703 

704def _host(host): 

705 # RFC 3986 3.2.3: Although host is case-insensitive, producers and 

706 # normalizers should use lowercase for registered names and 

707 # hexadecimal addresses for the sake of uniformity, while only 

708 # using uppercase letters for percent-encodings. 

709 if host.startswith(b"[") and host.endswith(b"]"): 

710 return _ip_literal(host[1:-1].decode()) 

711 # check for IPv6 addresses as returned by SplitResult.gethost() 

712 try: 

713 return _ip_literal(host.decode("utf-8")) 

714 except ValueError: 

715 return uriencode(host.lower(), _SAFE_HOST, "utf-8") 

716 

717 

718def _port(port): 

719 # RFC 3986 3.2.3: URI producers and normalizers should omit the 

720 # port component and its ":" delimiter if port is empty or if its 

721 # value would be the same as that of the scheme's default. 

722 if port.lstrip(b"0123456789"): 

723 raise ValueError("Invalid port subcomponent") 

724 elif port: 

725 return b":" + port 

726 else: 

727 return b"" 

728 

729 

730def _querylist(items, sep, encoding): 

731 terms = [] 

732 append = terms.append 

733 safe = _SAFE_QUERY.replace(sep, "") 

734 for key, value in items: 

735 name = uriencode(key, safe, encoding) 

736 if value is None: 

737 append(name) 

738 elif isinstance(value, (bytes, str)): 

739 append(name + b"=" + uriencode(value, safe, encoding)) 

740 else: 

741 append(name + b"=" + uriencode(str(value), safe, encoding)) 

742 return sep.encode("ascii").join(terms) 

743 

744 

745def _querydict(mapping, sep, encoding): 

746 items = [] 

747 for key, value in mapping.items(): 

748 if isinstance(value, (bytes, str)): 

749 items.append((key, value)) 

750 elif isinstance(value, collections.abc.Iterable): 

751 items.extend([(key, v) for v in value]) 

752 else: 

753 items.append((key, value)) 

754 return _querylist(items, sep, encoding) 

755 

756 

757def uricompose( 

758 scheme=None, 

759 authority=None, 

760 path="", 

761 query=None, 

762 fragment=None, 

763 userinfo=None, 

764 host=None, 

765 port=None, 

766 querysep="&", 

767 encoding="utf-8", 

768): 

769 """Compose a URI reference string from its individual components.""" 

770 

771 # RFC 3986 3.1: Scheme names consist of a sequence of characters 

772 # beginning with a letter and followed by any combination of 

773 # letters, digits, plus ("+"), period ("."), or hyphen ("-"). 

774 # Although schemes are case-insensitive, the canonical form is 

775 # lowercase and documents that specify schemes must do so with 

776 # lowercase letters. An implementation should accept uppercase 

777 # letters as equivalent to lowercase in scheme names (e.g., allow 

778 # "HTTP" as well as "http") for the sake of robustness but should 

779 # only produce lowercase scheme names for consistency. 

780 if isinstance(scheme, bytes): 

781 scheme = _scheme(scheme) 

782 elif isinstance(scheme, str): 

783 scheme = _scheme(scheme.encode()) 

784 elif scheme is not None: 

785 raise TypeError("Invalid scheme type: %r" % type(scheme)) 

786 

787 # authority must be string type or three-item sequence 

788 if authority is None: 

789 authority = (None, None, None) 

790 elif isinstance(authority, (bytes, str)): 

791 authority = _splitauthority(authority) 

792 elif not isinstance(authority, collections.abc.Sequence): 

793 raise TypeError("Invalid authority type") 

794 elif len(authority) != 3: 

795 raise ValueError("Invalid authority length") 

796 authority = _authority( 

797 userinfo if userinfo is not None else authority[0], 

798 host if host is not None else authority[1], 

799 port if port is not None else authority[2], 

800 encoding, 

801 ) 

802 

803 # RFC 3986 3.3: If a URI contains an authority component, then the 

804 # path component must either be empty or begin with a slash ("/") 

805 # character. If a URI does not contain an authority component, 

806 # then the path cannot begin with two slash characters ("//"). 

807 path = uriencode(path, _SAFE_PATH, encoding) 

808 if authority is not None and path and not path.startswith(b"/"): 

809 raise ValueError("Invalid path with authority component") 

810 if authority is None and path.startswith(b"//"): 

811 raise ValueError("Invalid path without authority component") 

812 

813 # RFC 3986 4.2: A path segment that contains a colon character 

814 # (e.g., "this:that") cannot be used as the first segment of a 

815 # relative-path reference, as it would be mistaken for a scheme 

816 # name. Such a segment must be preceded by a dot-segment (e.g., 

817 # "./this:that") to make a relative-path reference. 

818 if scheme is None and authority is None and not path.startswith(b"/"): 

819 if b":" in path.partition(b"/")[0]: 

820 path = b"./" + path 

821 

822 # RFC 3986 3.4: The characters slash ("/") and question mark ("?") 

823 # may represent data within the query component. Beware that some 

824 # older, erroneous implementations may not handle such data 

825 # correctly when it is used as the base URI for relative 

826 # references (Section 5.1), apparently because they fail to 

827 # distinguish query data from path data when looking for 

828 # hierarchical separators. However, as query components are often 

829 # used to carry identifying information in the form of "key=value" 

830 # pairs and one frequently used value is a reference to another 

831 # URI, it is sometimes better for usability to avoid percent- 

832 # encoding those characters. 

833 if isinstance(query, (bytes, str)): 

834 query = uriencode(query, _SAFE_QUERY, encoding) 

835 elif isinstance(query, collections.abc.Mapping): 

836 query = _querydict(query, querysep, encoding) 

837 elif isinstance(query, collections.abc.Iterable): 

838 query = _querylist(query, querysep, encoding) 

839 elif query is not None: 

840 raise TypeError("Invalid query type") 

841 

842 # RFC 3986 3.5: The characters slash ("/") and question mark ("?") 

843 # are allowed to represent data within the fragment identifier. 

844 # Beware that some older, erroneous implementations may not handle 

845 # this data correctly when it is used as the base URI for relative 

846 # references. 

847 if fragment is not None: 

848 fragment = uriencode(fragment, _SAFE_FRAGMENT, encoding) 

849 

850 # return URI reference as `str` 

851 parts = (scheme, authority, path, query, fragment) 

852 uri = uriunsplit(parts) 

853 assert isinstance(uri, bytes) 

854 return uri.decode()