Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/httpcore/

1from typing import (

2 Any,

3 AsyncIterable,

4 AsyncIterator,

5 Iterable,

6 Iterator,

7 List,

8 Mapping,

9 MutableMapping,

10 Optional,

11 Sequence,

12 Tuple,

13 Union,

14)

15from urllib.parse import urlparse

17# Functions for typechecking...

20HeadersAsSequence = Sequence[Tuple[Union[bytes, str], Union[bytes, str]]]

21HeadersAsMapping = Mapping[Union[bytes, str], Union[bytes, str]]

22HeaderTypes = Union[HeadersAsSequence, HeadersAsMapping, None]

24Extensions = MutableMapping[str, Any]

27def enforce_bytes(value: Union[bytes, str], *, name: str) -> bytes:

28 """

29 Any arguments that are ultimately represented as bytes can be specified

30 either as bytes or as strings.

32 However we enforce that any string arguments must only contain characters in

33 the plain ASCII range. chr(0)...chr(127). If you need to use characters

34 outside that range then be precise, and use a byte-wise argument.

35 """

36 if isinstance(value, str):

37 try:

38 return value.encode("ascii")

39 except UnicodeEncodeError:

40 raise TypeError(f"{name} strings may not include unicode characters.")

41 elif isinstance(value, bytes):

42 return value

44 seen_type = type(value).__name__

45 raise TypeError(f"{name} must be bytes or str, but got {seen_type}.")

48def enforce_url(value: Union["URL", bytes, str], *, name: str) -> "URL":

49 """

50 Type check for URL parameters.

51 """

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

53 return URL(value)

54 elif isinstance(value, URL):

55 return value

57 seen_type = type(value).__name__

58 raise TypeError(f"{name} must be a URL, bytes, or str, but got {seen_type}.")

61def enforce_headers(

62 value: Union[HeadersAsMapping, HeadersAsSequence, None] = None, *, name: str

63) -> List[Tuple[bytes, bytes]]:

64 """

65 Convienence function that ensure all items in request or response headers

66 are either bytes or strings in the plain ASCII range.

67 """

68 if value is None:

69 return []

70 elif isinstance(value, Mapping):

71 return [

72 (

73 enforce_bytes(k, name="header name"),

74 enforce_bytes(v, name="header value"),

75 )

76 for k, v in value.items()

77 ]

78 elif isinstance(value, Sequence):

79 return [

80 (

81 enforce_bytes(k, name="header name"),

82 enforce_bytes(v, name="header value"),

83 )

84 for k, v in value

85 ]

87 seen_type = type(value).__name__

88 raise TypeError(

89 f"{name} must be a mapping or sequence of two-tuples, but got {seen_type}."

90 )

93def enforce_stream(

94 value: Union[bytes, Iterable[bytes], AsyncIterable[bytes], None], *, name: str

95) -> Union[Iterable[bytes], AsyncIterable[bytes]]:

96 if value is None:

97 return ByteStream(b"")

98 elif isinstance(value, bytes):

99 return ByteStream(value)

100 return value

101

102

103# * https://tools.ietf.org/html/rfc3986#section-3.2.3

104# * https://url.spec.whatwg.org/#url-miscellaneous

105# * https://url.spec.whatwg.org/#scheme-state

106DEFAULT_PORTS = {

107 b"ftp": 21,

108 b"http": 80,

109 b"https": 443,

110 b"ws": 80,

111 b"wss": 443,

112}

113

114

115def include_request_headers(

116 headers: List[Tuple[bytes, bytes]],

117 *,

118 url: "URL",

119 content: Union[None, bytes, Iterable[bytes], AsyncIterable[bytes]],

120) -> List[Tuple[bytes, bytes]]:

121 headers_set = set(k.lower() for k, v in headers)

122

123 if b"host" not in headers_set:

124 default_port = DEFAULT_PORTS.get(url.scheme)

125 if url.port is None or url.port == default_port:

126 header_value = url.host

127 else:

128 header_value = b"%b:%d" % (url.host, url.port)

129 headers = [(b"Host", header_value)] + headers

130

131 if (

132 content is not None

133 and b"content-length" not in headers_set

134 and b"transfer-encoding" not in headers_set

135 ):

136 if isinstance(content, bytes):

137 content_length = str(len(content)).encode("ascii")

138 headers += [(b"Content-Length", content_length)]

139 else:

140 headers += [(b"Transfer-Encoding", b"chunked")] # pragma: nocover

141

142 return headers

143

144

145# Interfaces for byte streams...

146

147

148class ByteStream:

149 """

150 A container for non-streaming content, and that supports both sync and async

151 stream iteration.

152 """

153

154 def __init__(self, content: bytes) -> None:

155 self._content = content

156

157 def __iter__(self) -> Iterator[bytes]:

158 yield self._content

159

160 async def __aiter__(self) -> AsyncIterator[bytes]:

161 yield self._content

162

163 def __repr__(self) -> str:

164 return f"<{self.__class__.__name__} [{len(self._content)} bytes]>"

165

166

167class Origin:

168 def __init__(self, scheme: bytes, host: bytes, port: int) -> None:

169 self.scheme = scheme

170 self.host = host

171 self.port = port

172

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

174 return (

175 isinstance(other, Origin)

176 and self.scheme == other.scheme

177 and self.host == other.host

178 and self.port == other.port

179 )

180

181 def __str__(self) -> str:

182 scheme = self.scheme.decode("ascii")

183 host = self.host.decode("ascii")

184 port = str(self.port)

185 return f"{scheme}://{host}:{port}"

186

187

188class URL:

189 """

190 Represents the URL against which an HTTP request may be made.

191

192 The URL may either be specified as a plain string, for convienence:

193

194 ```python

195 url = httpcore.URL("https://www.example.com/")

196 ```

197

198 Or be constructed with explicitily pre-parsed components:

199

200 ```python

201 url = httpcore.URL(scheme=b'https', host=b'www.example.com', port=None, target=b'/')

202 ```

203

204 Using this second more explicit style allows integrations that are using

205 `httpcore` to pass through URLs that have already been parsed in order to use

206 libraries such as `rfc-3986` rather than relying on the stdlib. It also ensures

207 that URL parsing is treated identically at both the networking level and at any

208 higher layers of abstraction.

209

210 The four components are important here, as they allow the URL to be precisely

211 specified in a pre-parsed format. They also allow certain types of request to

212 be created that could not otherwise be expressed.

213

214 For example, an HTTP request to `http://www.example.com/` forwarded via a proxy

215 at `http://localhost:8080`...

216

217 ```python

218 # Constructs an HTTP request with a complete URL as the target:

219 # GET https://www.example.com/ HTTP/1.1

220 url = httpcore.URL(

221 scheme=b'http',

222 host=b'localhost',

223 port=8080,

224 target=b'https://www.example.com/'

225 )

226 request = httpcore.Request(

227 method="GET",

228 url=url

229 )

230 ```

231

232 Another example is constructing an `OPTIONS *` request...

233

234 ```python

235 # Constructs an 'OPTIONS *' HTTP request:

236 # OPTIONS * HTTP/1.1

237 url = httpcore.URL(scheme=b'https', host=b'www.example.com', target=b'*')

238 request = httpcore.Request(method="OPTIONS", url=url)

239 ```

240

241 This kind of request is not possible to formulate with a URL string,

242 because the `/` delimiter is always used to demark the target from the

243 host/port portion of the URL.

244

245 For convenience, string-like arguments may be specified either as strings or

246 as bytes. However, once a request is being issue over-the-wire, the URL

247 components are always ultimately required to be a bytewise representation.

248

249 In order to avoid any ambiguity over character encodings, when strings are used

250 as arguments, they must be strictly limited to the ASCII range `chr(0)`-`chr(127)`.

251 If you require a bytewise representation that is outside this range you must

252 handle the character encoding directly, and pass a bytes instance.

253 """

254

255 def __init__(

256 self,

257 url: Union[bytes, str] = "",

258 *,

259 scheme: Union[bytes, str] = b"",

260 host: Union[bytes, str] = b"",

261 port: Optional[int] = None,

262 target: Union[bytes, str] = b"",

263 ) -> None:

264 """

265 Parameters:

266 url: The complete URL as a string or bytes.

267 scheme: The URL scheme as a string or bytes.

268 Typically either `"http"` or `"https"`.

269 host: The URL host as a string or bytes. Such as `"www.example.com"`.

270 port: The port to connect to. Either an integer or `None`.

271 target: The target of the HTTP request. Such as `"/items?search=red"`.

272 """

273 if url:

274 parsed = urlparse(enforce_bytes(url, name="url"))

275 self.scheme = parsed.scheme

276 self.host = parsed.hostname or b""

277 self.port = parsed.port

278 self.target = (parsed.path or b"/") + (

279 b"?" + parsed.query if parsed.query else b""

280 )

281 else:

282 self.scheme = enforce_bytes(scheme, name="scheme")

283 self.host = enforce_bytes(host, name="host")

284 self.port = port

285 self.target = enforce_bytes(target, name="target")

286

287 @property

288 def origin(self) -> Origin:

289 default_port = {

290 b"http": 80,

291 b"https": 443,

292 b"ws": 80,

293 b"wss": 443,

294 b"socks5": 1080,

295 }[self.scheme]

296 return Origin(

297 scheme=self.scheme, host=self.host, port=self.port or default_port

298 )

299

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

301 return (

302 isinstance(other, URL)

303 and other.scheme == self.scheme

304 and other.host == self.host

305 and other.port == self.port

306 and other.target == self.target

307 )

308

309 def __bytes__(self) -> bytes:

310 if self.port is None:

311 return b"%b://%b%b" % (self.scheme, self.host, self.target)

312 return b"%b://%b:%d%b" % (self.scheme, self.host, self.port, self.target)

313

314 def __repr__(self) -> str:

315 return (

316 f"{self.__class__.__name__}(scheme={self.scheme!r}, "

317 f"host={self.host!r}, port={self.port!r}, target={self.target!r})"

318 )

319

320

321class Request:

322 """

323 An HTTP request.

324 """

325

326 def __init__(

327 self,

328 method: Union[bytes, str],

329 url: Union[URL, bytes, str],

330 *,

331 headers: HeaderTypes = None,

332 content: Union[bytes, Iterable[bytes], AsyncIterable[bytes], None] = None,

333 extensions: Optional[Extensions] = None,

334 ) -> None:

335 """

336 Parameters:

337 method: The HTTP request method, either as a string or bytes.

338 For example: `GET`.

339 url: The request URL, either as a `URL` instance, or as a string or bytes.

340 For example: `"https://www.example.com".`

341 headers: The HTTP request headers.

342 content: The content of the request body.

343 extensions: A dictionary of optional extra information included on

344 the request. Possible keys include `"timeout"`, and `"trace"`.

345 """

346 self.method: bytes = enforce_bytes(method, name="method")

347 self.url: URL = enforce_url(url, name="url")

348 self.headers: List[Tuple[bytes, bytes]] = enforce_headers(

349 headers, name="headers"

350 )

351 self.stream: Union[Iterable[bytes], AsyncIterable[bytes]] = enforce_stream(

352 content, name="content"

353 )

354 self.extensions = {} if extensions is None else extensions

355

356 if "target" in self.extensions:

357 self.url = URL(

358 scheme=self.url.scheme,

359 host=self.url.host,

360 port=self.url.port,

361 target=self.extensions["target"],

362 )

363

364 def __repr__(self) -> str:

365 return f"<{self.__class__.__name__} [{self.method!r}]>"

366

367

368class Response:

369 """

370 An HTTP response.

371 """

372

373 def __init__(

374 self,

375 status: int,

376 *,

377 headers: HeaderTypes = None,

378 content: Union[bytes, Iterable[bytes], AsyncIterable[bytes], None] = None,

379 extensions: Optional[Extensions] = None,

380 ) -> None:

381 """

382 Parameters:

383 status: The HTTP status code of the response. For example `200`.

384 headers: The HTTP response headers.

385 content: The content of the response body.

386 extensions: A dictionary of optional extra information included on

387 the responseself.Possible keys include `"http_version"`,

388 `"reason_phrase"`, and `"network_stream"`.

389 """

390 self.status: int = status

391 self.headers: List[Tuple[bytes, bytes]] = enforce_headers(

392 headers, name="headers"

393 )

394 self.stream: Union[Iterable[bytes], AsyncIterable[bytes]] = enforce_stream(

395 content, name="content"

396 )

397 self.extensions = {} if extensions is None else extensions

398

399 self._stream_consumed = False

400

401 @property

402 def content(self) -> bytes:

403 if not hasattr(self, "_content"):

404 if isinstance(self.stream, Iterable):

405 raise RuntimeError(

406 "Attempted to access 'response.content' on a streaming response. "

407 "Call 'response.read()' first."

408 )

409 else:

410 raise RuntimeError(

411 "Attempted to access 'response.content' on a streaming response. "

412 "Call 'await response.aread()' first."

413 )

414 return self._content

415

416 def __repr__(self) -> str:

417 return f"<{self.__class__.__name__} [{self.status}]>"

418

419 # Sync interface...

420

421 def read(self) -> bytes:

422 if not isinstance(self.stream, Iterable): # pragma: nocover

423 raise RuntimeError(

424 "Attempted to read an asynchronous response using 'response.read()'. "

425 "You should use 'await response.aread()' instead."

426 )

427 if not hasattr(self, "_content"):

428 self._content = b"".join([part for part in self.iter_stream()])

429 return self._content

430

431 def iter_stream(self) -> Iterator[bytes]:

432 if not isinstance(self.stream, Iterable): # pragma: nocover

433 raise RuntimeError(

434 "Attempted to stream an asynchronous response using 'for ... in "

435 "response.iter_stream()'. "

436 "You should use 'async for ... in response.aiter_stream()' instead."

437 )

438 if self._stream_consumed:

439 raise RuntimeError(

440 "Attempted to call 'for ... in response.iter_stream()' more than once."

441 )

442 self._stream_consumed = True

443 for chunk in self.stream:

444 yield chunk

445

446 def close(self) -> None:

447 if not isinstance(self.stream, Iterable): # pragma: nocover

448 raise RuntimeError(

449 "Attempted to close an asynchronous response using 'response.close()'. "

450 "You should use 'await response.aclose()' instead."

451 )

452 if hasattr(self.stream, "close"):

453 self.stream.close()

454

455 # Async interface...

456

457 async def aread(self) -> bytes:

458 if not isinstance(self.stream, AsyncIterable): # pragma: nocover

459 raise RuntimeError(

460 "Attempted to read an synchronous response using "

461 "'await response.aread()'. "

462 "You should use 'response.read()' instead."

463 )

464 if not hasattr(self, "_content"):

465 self._content = b"".join([part async for part in self.aiter_stream()])

466 return self._content

467

468 async def aiter_stream(self) -> AsyncIterator[bytes]:

469 if not isinstance(self.stream, AsyncIterable): # pragma: nocover

470 raise RuntimeError(

471 "Attempted to stream an synchronous response using 'async for ... in "

472 "response.aiter_stream()'. "

473 "You should use 'for ... in response.iter_stream()' instead."

474 )

475 if self._stream_consumed:

476 raise RuntimeError(

477 "Attempted to call 'async for ... in response.aiter_stream()' "

478 "more than once."

479 )

480 self._stream_consumed = True

481 async for chunk in self.stream:

482 yield chunk

483

484 async def aclose(self) -> None:

485 if not isinstance(self.stream, AsyncIterable): # pragma: nocover

486 raise RuntimeError(

487 "Attempted to close a synchronous response using "

488 "'await response.aclose()'. "

489 "You should use 'response.close()' instead."

490 )

491 if hasattr(self.stream, "aclose"):

492 await self.stream.aclose()

Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/httpcore/_models.py: 28%

150 statements