URLLib3 傳輸實(shí)現(xiàn)（從 HTTPX 中提?。?/h1>

urllib3-transport

使用 urllib3作為 HTTP 網(wǎng)絡(luò)后端的  HTTPCore傳輸。（這最初是隨 HTTPX一起提供的）。

當(dāng)與 HTTPX 一起使用時(shí)，此傳輸通過保持相同的基礎(chǔ) HTTP 網(wǎng)絡(luò)層，可以更輕松地從請(qǐng)求過渡到 HTTPX。

兼容：HTTPX 0.15.x，0.16.x（即HTTPCore 0.11.x和HTTPCore 0.12.x）。

注意：此處不支持所有urllib3 池管理器選項(xiàng) - 請(qǐng)隨時(shí)根據(jù)特定需求調(diào)整此要點(diǎn)。

用法

使用HTTPX:

import httpx
from urllib3_transport import URLLib3Transport

with httpx.Client(transport=URLLib3Transport()) as client:
    response = client.get("https://example.org")
    print(response)

如果要使用與 HTTPX(verify, cert, trust_env) 相同的選項(xiàng)傳遞自定義?ssl_context?, 請(qǐng)使用httpx.create_ssl_context() 幫助程序:

import httpx
from urllib3_transport import URLLib3Transport

ssl_context = httpx.create_ssl_context(verify="/tmp/client.pem")

with httpx.Client(transport=URLLib3Transport(ssl_context=ssl_context)) as client:
    response = client.get("https://example.org")
    print(response)

另請(qǐng)參閱更改 HTTPX 文檔中的驗(yàn)證默認(rèn)值。

urllib3_transport.py：

import socket
import ssl
from typing import Dict, Iterator, List, Optional, Tuple

import httpcore
import urllib3


class URLLib3ByteStream(httpcore.SyncByteStream):
    def __init__(self, response: urllib3.HTTPResponse) -> None:
        self._response = response

    def __iter__(self) -> Iterator[bytes]:
        try:
            for chunk in self._response.stream(4096, decode_content=False):
                yield chunk
        except socket.error as exc:
            raise httpcore.NetworkError(exc)

    def close(self) -> None:
        self._response.release_conn()


class URLLib3Transport(httpcore.SyncHTTPTransport):
    def __init__(
        self,
        *,
        ssl_context: ssl.SSLContext = None,
        pool_connections: int = 10,
        pool_maxsize: int = 10,
        pool_block: bool = False,
    ) -> None:
        self._pool = urllib3.PoolManager(
            ssl_context=ssl_context,
            num_pools=pool_connections,
            maxsize=pool_maxsize,
            block=pool_block,
        )

    def request(
        self,
        method: bytes,
        url: Tuple[bytes, bytes, Optional[int], bytes],
        headers: List[Tuple[bytes, bytes]] = None,
        stream: httpcore.SyncByteStream = None,
        ext: dict = None,
    ) -> Tuple[int, List[Tuple[bytes, bytes]], httpcore.SyncByteStream, dict]:
        headers = [] if headers is None else headers
        stream = httpcore.PlainByteStream(b"") if stream is None else stream
        ext = {} if ext is None else ext
        timeout: Dict[str, float] = ext["timeout"]

        urllib3_timeout = urllib3.util.Timeout(
            connect=timeout.get("connect"), read=timeout.get("read")
        )

        chunked = False
        content_length = 0
        for header_key, header_value in headers:
            header_key = header_key.lower()
            if header_key == b"transfer-encoding":
                chunked = header_value == b"chunked"
            if header_key == b"content-length":
                content_length = int(header_value.decode("ascii"))
        body = stream if chunked or content_length else None

        scheme, host, port, path = url
        default_port = {b"http": 80, "https": 443}.get(scheme)
        if port is None or port == default_port:
            url_str = "%s://%s%s" % (
                scheme.decode("ascii"),
                host.decode("ascii"),
                path.decode("ascii"),
            )
        else:
            url_str = "%s://%s:%d%s" % (
                scheme.decode("ascii"),
                host.decode("ascii"),
                port,
                path.decode("ascii"),
            )

        try:
            response = self._pool.urlopen(
                method=method.decode(),
                url=url_str,
                headers={
                    key.decode("ascii"): value.decode("ascii") for key, value in headers
                },
                body=body,
                redirect=False,
                assert_same_host=False,
                retries=0,
                preload_content=False,
                chunked=chunked,
                timeout=urllib3_timeout,
                pool_timeout=timeout.get("pool"),
            )
        except (urllib3.exceptions.SSLError, socket.error) as exc:
            raise httpcore.NetworkError(exc)

        status_code = response.status
        reason_phrase = response.reason
        headers = list(response.headers.items())
        stream = URLLib3ByteStream(response)
        ext = {"reason": reason_phrase, "http_version": "HTTP/1.1"}

        return (status_code, headers, stream, ext)

    def close(self) -> None:
        self._pool.clear()


class URLLib3ProxyTransport(URLLib3Transport):
    def __init__(
        self,
        *,
        proxy_url: str,
        proxy_headers: dict = None,
        ssl_context: ssl.SSLContext = None,
        pool_connections: int = 10,
        pool_maxsize: int = 10,
        pool_block: bool = False,
    ) -> None:
        self._pool = urllib3.ProxyManager(
            proxy_url=proxy_url,
            proxy_headers=proxy_headers,
            ssl_context=ssl_context,
            num_pools=pool_connections,
            maxsize=pool_maxsize,
            block=pool_block,
        )

許可證

MIT 許可

Copyright (c) 2020 Florimond Manca

特此免費(fèi)授予獲得本軟件和相關(guān)文檔文件（“軟件”）副本的任何人不受限制地處理本軟件的許可，包括但不限于使用、復(fù)制、修改、合并、發(fā)布、分發(fā)、再許可和/或出售本軟件副本的權(quán)利，并允許向其提供本軟件的人員這樣做， 須符合以下條件：

上述版權(quán)聲明和本許可聲明應(yīng)包含在本軟件的所有副本或大部分內(nèi)容中。

本軟件按“原樣”提供，不作任何明示或暗示的保證，包括但不限于適銷性、特定用途適用性和無侵權(quán)的保證。在任何情況下，作者或版權(quán)所有者均不對(duì)任何索賠、損害或其他責(zé)任負(fù)責(zé)，無論是在合同訴訟、侵權(quán)行為或其他訴訟中，由本軟件或本軟件的使用或其他交易引起、由本軟件引起或與之相關(guān)。

本文地址：https://gist.github.com/florimondmanca/d56764d78d748eb9f73165da388e546e