1. 程式人生 > 其它 >httpx的兩個坑(httpx.ReadTimeout; SSL: CERTIFICATE_VERIFY_FAILED)

httpx的兩個坑(httpx.ReadTimeout; SSL: CERTIFICATE_VERIFY_FAILED)

關於python非同步網路請求庫httpx的兩個坑

其一:httpx.ReadTimeout

  • 實測發現,網路不穩定的情況下,極其容易出現該錯誤。
  • 相對於requests庫, httpx庫是有預設的超時時間的。
  • 參考方案: 初始化時將timeout 賦值為 None

例1: client = httpx.AsyncClient(timeout=None)
例2: httpx.get(url=url, timeout=None)

# 預設配置
# location: /site-packages/httpx/_config.py:358
DEFAULT_TIMEOUT_CONFIG = Timeout(timeout=5.0)


# 附 Timeout 類
class Timeout:
    """
    Timeout configuration.

    **Usage**:

    Timeout(None)               # No timeouts.
    Timeout(5.0)                # 5s timeout on all operations.
    Timeout(None, connect=5.0)  # 5s timeout on connect, no other timeouts.
    Timeout(5.0, connect=10.0)  # 10s timeout on connect. 5s timeout elsewhere.
    Timeout(5.0, pool=None)     # No timeout on acquiring connection from pool.
                                # 5s timeout elsewhere.
    """

    def __init__(
        self,
        timeout: typing.Union[TimeoutTypes, UnsetType] = UNSET,
        *,
        connect: typing.Union[None, float, UnsetType] = UNSET,
        read: typing.Union[None, float, UnsetType] = UNSET,
        write: typing.Union[None, float, UnsetType] = UNSET,
        pool: typing.Union[None, float, UnsetType] = UNSET,
    ):
        if isinstance(timeout, Timeout):
            # Passed as a single explicit Timeout.
            assert connect is UNSET
            assert read is UNSET
            assert write is UNSET
            assert pool is UNSET
            self.connect = timeout.connect  # type: typing.Optional[float]
            self.read = timeout.read  # type: typing.Optional[float]
            self.write = timeout.write  # type: typing.Optional[float]
            self.pool = timeout.pool  # type: typing.Optional[float]
        elif isinstance(timeout, tuple):
            # Passed as a tuple.
            self.connect = timeout[0]
            self.read = timeout[1]
            self.write = None if len(timeout) < 3 else timeout[2]
            self.pool = None if len(timeout) < 4 else timeout[3]
        elif not (
            isinstance(connect, UnsetType)
            or isinstance(read, UnsetType)
            or isinstance(write, UnsetType)
            or isinstance(pool, UnsetType)
        ):
            self.connect = connect
            self.read = read
            self.write = write
            self.pool = pool
        else:
            if isinstance(timeout, UnsetType):
                raise ValueError(
                    "httpx.Timeout must either include a default, or set all "
                    "four parameters explicitly."
                )
            self.connect = timeout if isinstance(connect, UnsetType) else connect
            self.read = timeout if isinstance(read, UnsetType) else read
            self.write = timeout if isinstance(write, UnsetType) else write
            self.pool = timeout if isinstance(pool, UnsetType) else pool

其二:httpx.ConnectError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: Hostname mismatch, certificate is not valid for '*'. (_ssl.c:1108)

  • 在多域名或者萬用字元域名證書的情況下,可能會出現此類錯誤。
  • 參考方案: 初始化時將verify 賦值為 False

例1: client = httpx.AsyncClient(verify=False)
例2: httpx.get(url=url, verify=False)

# 預設配置
# `SSLConfig` 類中, `verify`預設值為 `True`

class SSLConfig:
    """
    SSL Configuration.
    """

    DEFAULT_CA_BUNDLE_PATH = Path(certifi.where())

    def __init__(
        self,
        *,
        cert: CertTypes = None,
        verify: VerifyTypes = True,
        trust_env: bool = True,
        http2: bool = False,
    ):
    	pass

綜上:

在引入 httpx 包 初始化時,將以上兩個引數先行傳入。


class NewClass:
    def __init__(self):
        self.client = httpx.AsyncClient(verify=False, timeout=None)
        _header = {
            "User-Agent": "Mozilla/5.0 (M1 Mac OS X 12) Safari/666.66"
        }
        self.client.headers.update(_header)