Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/starlette/datastructures.py: 31%

1from __future__ import annotations 

2 

3import ipaddress 

4from collections.abc import ItemsView, Iterable, Iterator, KeysView, Mapping, MutableMapping, Sequence, ValuesView 

5from operator import itemgetter 

6from shlex import shlex 

7from typing import Any, BinaryIO, Literal, NamedTuple, TypeVar, cast 

8from urllib.parse import SplitResult, parse_qsl, urlencode, urlsplit 

9 

10from starlette._utils import parse_host_header 

11from starlette.concurrency import run_in_threadpool 

12from starlette.types import Scope 

13 

14 

15class Address(NamedTuple): 

16 host: str 

17 port: int 

18 

19 

20_KeyType = TypeVar("_KeyType") 

21# Mapping keys are invariant but their values are covariant since 

22# you can only read them 

23# that is, you can't do `Mapping[str, Animal]()["fido"] = Dog()` 

24_CovariantValueType = TypeVar("_CovariantValueType", covariant=True) 

25 

26 

27class URL: 

28 def __init__( 

29 self, 

30 url: str = "", 

31 scope: Scope | None = None, 

32 **components: Any, 

33 ) -> None: 

34 if scope is not None: 

35 assert not url, 'Cannot set both "url" and "scope".' 

36 assert not components, 'Cannot set both "scope" and "**components".' 

37 scheme = scope.get("scheme", "http") 

38 server = scope.get("server", None) 

39 path = scope["path"].replace("?", "%3F").replace("#", "%23").replace("\\", "%5C") 

40 query_string = scope.get("query_string", b"") 

41 

42 host_header = None 

43 for key, value in scope["headers"]: 

44 if key == b"host": 

45 host_header = value.decode("latin-1") 

46 break 

47 

48 parsed_host = parse_host_header(host_header) 

49 if parsed_host is not None and parsed_host.is_valid_port: 

50 netloc = parsed_host.authority 

51 elif server is not None: 

52 host, port = server 

53 if ":" in host and not host.startswith("["): 

54 host = f"[{host}]" 

55 default_port = {"http": 80, "https": 443, "ws": 80, "wss": 443}[scheme] 

56 netloc = host if port == default_port else f"{host}:{port}" 

57 else: 

58 netloc = None 

59 

60 query = query_string.decode() 

61 if netloc is not None: 

62 url = SplitResult(scheme=scheme, netloc=netloc, path=path, query=query, fragment="").geturl() 

63 else: 

64 if path.startswith("//"): 

65 # Without a host, a leading `//` would be parsed as the authority. 

66 path = "/%2F" + path[2:] 

67 url = f"{path}?{query}" if query else path 

68 elif components: 

69 assert not url, 'Cannot set both "url" and "**components".' 

70 url = URL("").replace(**components).components.geturl() 

71 

72 self._url = url 

73 

74 @property 

75 def components(self) -> SplitResult: 

76 if not hasattr(self, "_components"): 

77 self._components = urlsplit(self._url) 

78 return self._components 

79 

80 @property 

81 def scheme(self) -> str: 

82 return self.components.scheme 

83 

84 @property 

85 def netloc(self) -> str: 

86 return self.components.netloc 

87 

88 @property 

89 def path(self) -> str: 

90 return self.components.path 

91 

92 @property 

93 def query(self) -> str: 

94 return self.components.query 

95 

96 @property 

97 def fragment(self) -> str: 

98 return self.components.fragment 

99 

100 @property 

101 def username(self) -> None | str: 

102 return self.components.username 

103 

104 @property 

105 def password(self) -> None | str: 

106 return self.components.password 

107 

108 @property 

109 def hostname(self) -> None | str: 

110 return self.components.hostname 

111 

112 @property 

113 def port(self) -> int | None: 

114 return self.components.port 

115 

116 @property 

117 def is_secure(self) -> bool: 

118 return self.scheme in ("https", "wss") 

119 

120 def replace(self, **kwargs: Any) -> URL: 

121 if "username" in kwargs or "password" in kwargs or "hostname" in kwargs or "port" in kwargs: 

122 hostname = kwargs.pop("hostname", None) 

123 port = kwargs.pop("port", self.port) 

124 username = kwargs.pop("username", self.username) 

125 password = kwargs.pop("password", self.password) 

126 

127 if hostname is None: 

128 netloc = self.netloc 

129 _, _, hostname = netloc.rpartition("@") 

130 

131 if hostname and hostname[-1] != "]": 

132 hostname = hostname.rsplit(":", 1)[0] 

133 elif ":" in hostname and not hostname.startswith("["): 

134 try: 

135 ipaddress.IPv6Address(hostname) 

136 except ValueError: 

137 pass 

138 else: 

139 hostname = f"[{hostname}]" 

140 

141 netloc = hostname 

142 if port is not None: 

143 netloc += f":{port}" 

144 if username is not None: 

145 userpass = username 

146 if password is not None: 

147 userpass += f":{password}" 

148 netloc = f"{userpass}@{netloc}" 

149 

150 kwargs["netloc"] = netloc 

151 

152 components = self.components._replace(**kwargs) 

153 return self.__class__(components.geturl()) 

154 

155 def include_query_params(self, **kwargs: Any) -> URL: 

156 params = MultiDict(parse_qsl(self.query, keep_blank_values=True)) 

157 params.update({str(key): str(value) for key, value in kwargs.items()}) 

158 query = urlencode(params.multi_items()) 

159 return self.replace(query=query) 

160 

161 def replace_query_params(self, **kwargs: Any) -> URL: 

162 query = urlencode([(str(key), str(value)) for key, value in kwargs.items()]) 

163 return self.replace(query=query) 

164 

165 def remove_query_params(self, keys: str | Sequence[str]) -> URL: 

166 if isinstance(keys, str): 

167 keys = [keys] 

168 params = MultiDict(parse_qsl(self.query, keep_blank_values=True)) 

169 for key in keys: 

170 params.pop(key, None) 

171 query = urlencode(params.multi_items()) 

172 return self.replace(query=query) 

173 

174 def __eq__(self, other: Any) -> bool: 

175 return str(self) == str(other) 

176 

177 def __str__(self) -> str: 

178 return self._url 

179 

180 def __repr__(self) -> str: 

181 url = str(self) 

182 if self.password: 

183 url = str(self.replace(password="********")) 

184 return f"{self.__class__.__name__}({repr(url)})" 

185 

186 

187class URLPath(str): 

188 """ 

189 A URL path string that may also hold an associated protocol and/or host. 

190 Used by the routing to return `url_path_for` matches. 

191 """ 

192 

193 def __new__(cls, path: str, protocol: Literal["http", "websocket", ""] = "", host: str = "") -> URLPath: 

194 assert protocol in ("http", "websocket", "") 

195 path = path.replace("\\", "%5C") 

196 if path.startswith("//"): 

197 # A leading `//` would be parsed as the authority. 

198 path = "/%2F" + path[2:] 

199 return str.__new__(cls, path) 

200 

201 def __init__(self, path: str, protocol: Literal["http", "websocket", ""] = "", host: str = "") -> None: 

202 self.protocol = protocol 

203 self.host = host 

204 

205 def make_absolute_url(self, base_url: str | URL) -> URL: 

206 if isinstance(base_url, str): 

207 base_url = URL(base_url) 

208 if self.protocol: 

209 scheme = { 

210 "http": {True: "https", False: "http"}, 

211 "websocket": {True: "wss", False: "ws"}, 

212 }[self.protocol][base_url.is_secure] 

213 else: 

214 scheme = base_url.scheme 

215 

216 netloc = self.host or base_url.netloc 

217 path = base_url.path.rstrip("/") + str(self) 

218 return URL(scheme=scheme, netloc=netloc, path=path) 

219 

220 

221class Secret: 

222 """ 

223 Holds a string value that should not be revealed in tracebacks etc. 

224 You should cast the value to `str` at the point it is required. 

225 """ 

226 

227 def __init__(self, value: str): 

228 self._value = value 

229 

230 def __repr__(self) -> str: 

231 class_name = self.__class__.__name__ 

232 return f"{class_name}('**********')" 

233 

234 def __str__(self) -> str: 

235 return self._value 

236 

237 def __bool__(self) -> bool: 

238 return bool(self._value) 

239 

240 

241class CommaSeparatedStrings(Sequence[str]): 

242 def __init__(self, value: str | Sequence[str]): 

243 if isinstance(value, str): 

244 splitter = shlex(value, posix=True) 

245 splitter.whitespace = "," 

246 splitter.whitespace_split = True 

247 self._items = [item.strip() for item in splitter] 

248 else: 

249 self._items = list(value) 

250 

251 def __len__(self) -> int: 

252 return len(self._items) 

253 

254 def __getitem__(self, index: int | slice) -> Any: 

255 return self._items[index] 

256 

257 def __iter__(self) -> Iterator[str]: 

258 return iter(self._items) 

259 

260 def __repr__(self) -> str: 

261 class_name = self.__class__.__name__ 

262 items = [item for item in self] 

263 return f"{class_name}({items!r})" 

264 

265 def __str__(self) -> str: 

266 return ", ".join(repr(item) for item in self) 

267 

268 

269class ImmutableMultiDict(Mapping[_KeyType, _CovariantValueType]): 

270 _dict: dict[_KeyType, _CovariantValueType] 

271 

272 def __init__( 

273 self, 

274 *args: ImmutableMultiDict[_KeyType, _CovariantValueType] 

275 | Mapping[_KeyType, _CovariantValueType] 

276 | Iterable[tuple[_KeyType, _CovariantValueType]], 

277 **kwargs: Any, 

278 ) -> None: 

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

280 

281 value: Any = args[0] if args else [] 

282 if kwargs: 

283 value = ImmutableMultiDict(value).multi_items() + ImmutableMultiDict(kwargs).multi_items() 

284 

285 if not value: 

286 _items: list[tuple[Any, Any]] = [] 

287 elif hasattr(value, "multi_items"): 

288 value = cast(ImmutableMultiDict[_KeyType, _CovariantValueType], value) 

289 _items = list(value.multi_items()) 

290 elif hasattr(value, "items"): 

291 value = cast(Mapping[_KeyType, _CovariantValueType], value) 

292 _items = list(value.items()) 

293 else: 

294 value = cast("list[tuple[Any, Any]]", value) 

295 _items = list(value) 

296 

297 self._dict = {k: v for k, v in _items} 

298 self._list = _items 

299 

300 def getlist(self, key: Any) -> list[_CovariantValueType]: 

301 return [item_value for item_key, item_value in self._list if item_key == key] 

302 

303 def keys(self) -> KeysView[_KeyType]: 

304 return self._dict.keys() 

305 

306 def values(self) -> ValuesView[_CovariantValueType]: 

307 return self._dict.values() 

308 

309 def items(self) -> ItemsView[_KeyType, _CovariantValueType]: 

310 return self._dict.items() 

311 

312 def multi_items(self) -> list[tuple[_KeyType, _CovariantValueType]]: 

313 return list(self._list) 

314 

315 def __getitem__(self, key: _KeyType) -> _CovariantValueType: 

316 return self._dict[key] 

317 

318 def __contains__(self, key: Any) -> bool: 

319 return key in self._dict 

320 

321 def __iter__(self) -> Iterator[_KeyType]: 

322 return iter(self.keys()) 

323 

324 def __len__(self) -> int: 

325 return len(self._dict) 

326 

327 def __eq__(self, other: Any) -> bool: 

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

329 return False 

330 return sorted(self._list, key=itemgetter(0)) == sorted(other._list, key=itemgetter(0)) 

331 

332 def __repr__(self) -> str: 

333 class_name = self.__class__.__name__ 

334 items = self.multi_items() 

335 return f"{class_name}({items!r})" 

336 

337 

338class MultiDict(ImmutableMultiDict[Any, Any]): 

339 def __setitem__(self, key: Any, value: Any) -> None: 

340 self.setlist(key, [value]) 

341 

342 def __delitem__(self, key: Any) -> None: 

343 self._list = [(k, v) for k, v in self._list if k != key] 

344 del self._dict[key] 

345 

346 def pop(self, key: Any, default: Any = None) -> Any: 

347 self._list = [(k, v) for k, v in self._list if k != key] 

348 return self._dict.pop(key, default) 

349 

350 def popitem(self) -> tuple[Any, Any]: 

351 key, value = self._dict.popitem() 

352 self._list = [(k, v) for k, v in self._list if k != key] 

353 return key, value 

354 

355 def poplist(self, key: Any) -> list[Any]: 

356 values = [v for k, v in self._list if k == key] 

357 self.pop(key) 

358 return values 

359 

360 def clear(self) -> None: 

361 self._dict.clear() 

362 self._list.clear() 

363 

364 def setdefault(self, key: Any, default: Any = None) -> Any: 

365 if key not in self: 

366 self._dict[key] = default 

367 self._list.append((key, default)) 

368 

369 return self[key] 

370 

371 def setlist(self, key: Any, values: list[Any]) -> None: 

372 if not values: 

373 self.pop(key, None) 

374 else: 

375 existing_items = [(k, v) for (k, v) in self._list if k != key] 

376 self._list = existing_items + [(key, value) for value in values] 

377 self._dict[key] = values[-1] 

378 

379 def append(self, key: Any, value: Any) -> None: 

380 self._list.append((key, value)) 

381 self._dict[key] = value 

382 

383 def update( 

384 self, 

385 *args: MultiDict | Mapping[Any, Any] | list[tuple[Any, Any]], 

386 **kwargs: Any, 

387 ) -> None: 

388 value = MultiDict(*args, **kwargs) 

389 existing_items = [(k, v) for (k, v) in self._list if k not in value.keys()] 

390 self._list = existing_items + value.multi_items() 

391 self._dict.update(value) 

392 

393 

394class QueryParams(ImmutableMultiDict[str, str]): 

395 """ 

396 An immutable multidict. 

397 """ 

398 

399 def __init__( 

400 self, 

401 *args: ImmutableMultiDict[Any, Any] | Mapping[Any, Any] | list[tuple[Any, Any]] | str | bytes, 

402 **kwargs: Any, 

403 ) -> None: 

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

405 

406 value = args[0] if args else [] 

407 

408 if isinstance(value, str): 

409 super().__init__(parse_qsl(value, keep_blank_values=True), **kwargs) 

410 elif isinstance(value, bytes): 

411 super().__init__(parse_qsl(value.decode("latin-1"), keep_blank_values=True), **kwargs) 

412 else: 

413 super().__init__(*args, **kwargs) # type: ignore[arg-type] 

414 self._list = [(str(k), str(v)) for k, v in self._list] 

415 self._dict = {str(k): str(v) for k, v in self._dict.items()} 

416 

417 def __str__(self) -> str: 

418 return urlencode(self._list) 

419 

420 def __repr__(self) -> str: 

421 class_name = self.__class__.__name__ 

422 query_string = str(self) 

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

424 

425 

426class UploadFile: 

427 """ 

428 An uploaded file included as part of the request data. 

429 """ 

430 

431 def __init__( 

432 self, 

433 file: BinaryIO, 

434 *, 

435 size: int | None = None, 

436 filename: str | None = None, 

437 headers: Headers | None = None, 

438 ) -> None: 

439 self.filename = filename 

440 self.file = file 

441 self.size = size 

442 self.headers = headers or Headers() 

443 

444 # Capture max size from SpooledTemporaryFile if one is provided. This slightly speeds up future checks. 

445 # Note 0 means unlimited mirroring SpooledTemporaryFile's __init__ 

446 self._max_mem_size = getattr(self.file, "_max_size", 0) 

447 

448 @property 

449 def content_type(self) -> str | None: 

450 return self.headers.get("content-type", None) 

451 

452 @property 

453 def _in_memory(self) -> bool: 

454 # check for SpooledTemporaryFile._rolled 

455 rolled_to_disk = getattr(self.file, "_rolled", True) 

456 return not rolled_to_disk 

457 

458 def _will_roll(self, size_to_add: int) -> bool: 

459 # If we're not in_memory then we will always roll 

460 if not self._in_memory: 

461 return True 

462 

463 # Check for SpooledTemporaryFile._max_size 

464 future_size = self.file.tell() + size_to_add 

465 return bool(future_size > self._max_mem_size) if self._max_mem_size else False 

466 

467 async def write(self, data: bytes) -> None: 

468 new_data_len = len(data) 

469 if self.size is not None: 

470 self.size += new_data_len 

471 

472 if self._will_roll(new_data_len): 

473 await run_in_threadpool(self.file.write, data) 

474 else: 

475 self.file.write(data) 

476 

477 async def read(self, size: int = -1) -> bytes: 

478 if self._in_memory: 

479 return self.file.read(size) 

480 return await run_in_threadpool(self.file.read, size) 

481 

482 async def seek(self, offset: int) -> None: 

483 if self._in_memory: 

484 self.file.seek(offset) 

485 else: 

486 await run_in_threadpool(self.file.seek, offset) 

487 

488 async def close(self) -> None: 

489 if self._in_memory: 

490 self.file.close() 

491 else: 

492 await run_in_threadpool(self.file.close) 

493 

494 def __repr__(self) -> str: 

495 return f"{self.__class__.__name__}(filename={self.filename!r}, size={self.size!r}, headers={self.headers!r})" 

496 

497 

498class FormData(ImmutableMultiDict[str, UploadFile | str]): 

499 """ 

500 An immutable multidict, containing both file uploads and text input. 

501 """ 

502 

503 def __init__( 

504 self, 

505 *args: FormData | Mapping[str, str | UploadFile] | list[tuple[str, str | UploadFile]], 

506 **kwargs: str | UploadFile, 

507 ) -> None: 

508 super().__init__(*args, **kwargs) 

509 

510 async def close(self) -> None: 

511 for key, value in self.multi_items(): 

512 if isinstance(value, UploadFile): 

513 await value.close() 

514 

515 

516class Headers(Mapping[str, str]): 

517 """ 

518 An immutable, case-insensitive multidict. 

519 """ 

520 

521 def __init__( 

522 self, 

523 headers: Mapping[str, str] | None = None, 

524 raw: list[tuple[bytes, bytes]] | None = None, 

525 scope: MutableMapping[str, Any] | None = None, 

526 ) -> None: 

527 self._list: list[tuple[bytes, bytes]] = [] 

528 if headers is not None: 

529 assert raw is None, 'Cannot set both "headers" and "raw".' 

530 assert scope is None, 'Cannot set both "headers" and "scope".' 

531 self._list = [(key.lower().encode("latin-1"), value.encode("latin-1")) for key, value in headers.items()] 

532 elif raw is not None: 

533 assert scope is None, 'Cannot set both "raw" and "scope".' 

534 self._list = raw 

535 elif scope is not None: 

536 # scope["headers"] isn't necessarily a list 

537 # it might be a tuple or other iterable 

538 self._list = scope["headers"] = list(scope["headers"]) 

539 

540 @property 

541 def raw(self) -> list[tuple[bytes, bytes]]: 

542 return list(self._list) 

543 

544 def keys(self) -> list[str]: # type: ignore[override] 

545 return [key.decode("latin-1") for key, value in self._list] 

546 

547 def values(self) -> list[str]: # type: ignore[override] 

548 return [value.decode("latin-1") for key, value in self._list] 

549 

550 def items(self) -> list[tuple[str, str]]: # type: ignore[override] 

551 return [(key.decode("latin-1"), value.decode("latin-1")) for key, value in self._list] 

552 

553 def getlist(self, key: str) -> list[str]: 

554 get_header_key = key.lower().encode("latin-1") 

555 return [item_value.decode("latin-1") for item_key, item_value in self._list if item_key == get_header_key] 

556 

557 def mutablecopy(self) -> MutableHeaders: 

558 return MutableHeaders(raw=self._list[:]) 

559 

560 def __getitem__(self, key: str) -> str: 

561 get_header_key = key.lower().encode("latin-1") 

562 for header_key, header_value in self._list: 

563 if header_key == get_header_key: 

564 return header_value.decode("latin-1") 

565 raise KeyError(key) 

566 

567 def __contains__(self, key: Any) -> bool: 

568 get_header_key = key.lower().encode("latin-1") 

569 for header_key, header_value in self._list: 

570 if header_key == get_header_key: 

571 return True 

572 return False 

573 

574 def __iter__(self) -> Iterator[Any]: 

575 return iter(self.keys()) 

576 

577 def __len__(self) -> int: 

578 return len(self._list) 

579 

580 def __eq__(self, other: Any) -> bool: 

581 if not isinstance(other, Headers): 

582 return False 

583 return sorted(self._list) == sorted(other._list) 

584 

585 def __repr__(self) -> str: 

586 class_name = self.__class__.__name__ 

587 as_dict = dict(self.items()) 

588 if len(as_dict) == len(self): 

589 return f"{class_name}({as_dict!r})" 

590 return f"{class_name}(raw={self.raw!r})" 

591 

592 

593class MutableHeaders(Headers): 

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

595 """ 

596 Set the header `key` to `value`, removing any duplicate entries. 

597 Retains insertion order. 

598 """ 

599 set_key = key.lower().encode("latin-1") 

600 set_value = value.encode("latin-1") 

601 

602 found_indexes: list[int] = [] 

603 for idx, (item_key, item_value) in enumerate(self._list): 

604 if item_key == set_key: 

605 found_indexes.append(idx) 

606 

607 for idx in reversed(found_indexes[1:]): 

608 del self._list[idx] 

609 

610 if found_indexes: 

611 idx = found_indexes[0] 

612 self._list[idx] = (set_key, set_value) 

613 else: 

614 self._list.append((set_key, set_value)) 

615 

616 def __delitem__(self, key: str) -> None: 

617 """ 

618 Remove the header `key`. 

619 """ 

620 del_key = key.lower().encode("latin-1") 

621 

622 pop_indexes: list[int] = [] 

623 for idx, (item_key, item_value) in enumerate(self._list): 

624 if item_key == del_key: 

625 pop_indexes.append(idx) 

626 

627 for idx in reversed(pop_indexes): 

628 del self._list[idx] 

629 

630 def __ior__(self, other: Mapping[str, str]) -> MutableHeaders: 

631 if not isinstance(other, Mapping): 

632 raise TypeError(f"Expected a mapping but got {other.__class__.__name__}") 

633 self.update(other) 

634 return self 

635 

636 def __or__(self, other: Mapping[str, str]) -> MutableHeaders: 

637 if not isinstance(other, Mapping): 

638 raise TypeError(f"Expected a mapping but got {other.__class__.__name__}") 

639 new = self.mutablecopy() 

640 new.update(other) 

641 return new 

642 

643 @property 

644 def raw(self) -> list[tuple[bytes, bytes]]: 

645 return self._list 

646 

647 def setdefault(self, key: str, value: str) -> str: 

648 """ 

649 If the header `key` does not exist, then set it to `value`. 

650 Returns the header value. 

651 """ 

652 set_key = key.lower().encode("latin-1") 

653 set_value = value.encode("latin-1") 

654 

655 for idx, (item_key, item_value) in enumerate(self._list): 

656 if item_key == set_key: 

657 return item_value.decode("latin-1") 

658 self._list.append((set_key, set_value)) 

659 return value 

660 

661 def update(self, other: Mapping[str, str]) -> None: 

662 for key, val in other.items(): 

663 self[key] = val 

664 

665 def append(self, key: str, value: str) -> None: 

666 """ 

667 Append a header, preserving any duplicate entries. 

668 """ 

669 append_key = key.lower().encode("latin-1") 

670 append_value = value.encode("latin-1") 

671 self._list.append((append_key, append_value)) 

672 

673 def add_vary_header(self, vary: str) -> None: 

674 self["vary"] = ", ".join([*self.getlist("vary"), vary]) 

675 

676 

677class State: 

678 """ 

679 An object that can be used to store arbitrary state. 

680 

681 Used for `request.state` and `app.state`. 

682 """ 

683 

684 _state: dict[str, Any] 

685 

686 def __init__(self, state: dict[str, Any] | None = None): 

687 if state is None: 

688 state = {} 

689 super().__setattr__("_state", state) 

690 

691 def __setattr__(self, key: Any, value: Any) -> None: 

692 self._state[key] = value 

693 

694 def __getattr__(self, key: Any) -> Any: 

695 try: 

696 return self._state[key] 

697 except KeyError: 

698 message = "'{}' object has no attribute '{}'" 

699 raise AttributeError(message.format(self.__class__.__name__, key))