用户指南#

安装#

可以安装ULLIb3 pip

$ python -m pip install urllib3

提出请求#

首先,导入urllib3模块:

import urllib3

你需要一个 PoolManager 实例发出请求。此对象处理连接池和线程安全的所有详细信息,因此您不必:

http = urllib3.PoolManager()

要发出请求,请使用 request()

import urllib3

# Creating a PoolManager instance for sending requests.
http = urllib3.PoolManager()

# Sending a GET request and getting back response as HTTPResponse object.
resp = http.request("GET", "https://httpbin.org/robots.txt")

# Print the returned data.
print(resp.data)
# b"User-agent: *\nDisallow: /deny\n"

request() 返回A HTTPResponse 对象 响应内容 部分说明如何处理各种响应。

您可以使用 request() 要使用任何HTTP谓词发出请求,请执行以下操作:

import urllib3

http = urllib3.PoolManager()
resp = http.request(
    "POST",
    "https://httpbin.org/post",
    fields={"hello": "world"} #  Add custom form fields
)

print(resp.data)
# b"{\n "form": {\n "hello": "world"\n  }, ... }

这个 请求数据 本节介绍发送其他类型的请求数据,包括json、文件和二进制数据。

备注

对于快速脚本和实验,您还可以使用顶级 urllib3.request() 。它使用一个模块-全局 PoolManager 举个例子。正因为如此,它的副作用可以在依赖它的依赖项之间共享。若要避免副作用,请创建新的 PoolManager 实例,并改用它。此外,该方法不接受低级别的 **urlopen_kw 关键字参数。默认情况下加载系统CA证书。

响应内容#

这个 HTTPResponse 对象提供 statusdataheaders 属性:

import urllib3

# Making the request (The request function returns HTTPResponse object)
resp = urllib3.request("GET", "https://httpbin.org/ip")

print(resp.status)
# 200
print(resp.data)
# b"{\n  "origin": "104.232.115.37"\n}\n"
print(resp.headers)
# HTTPHeaderDict({"Content-Length": "32", ...})

JSON内容#

JSON内容可以通过 json() 回复方式:

import urllib3

resp = urllib3.request("GET", "https://httpbin.org/ip")

print(resp.json())
# {"origin": "127.0.0.1"}

或者,自定义JSON库,如 orjson 可用于编码数据、通过解码和反序列化 data 请求的属性:

import orjson
import urllib3

encoded_data = orjson.dumps({"attribute": "value"})
resp = urllib3.request(method="POST", url="http://httpbin.org/post", body=encoded_data)

print(orjson.loads(resp.data)["json"])
# {'attribute': 'value'}

二进制内容#

这个 data 响应的属性始终设置为表示响应内容的字节字符串:

import urllib3

resp = urllib3.request("GET", "https://httpbin.org/bytes/8")

print(resp.data)
# b"\xaa\xa5H?\x95\xe9\x9b\x11"

备注

对于更大的响应,有时最好是 stream 反应。

对响应内容使用io包装器#

有时候你想用 io.TextIOWrapper 或类似的对象,如直接使用 HTTPResponse 数据。要使这两个接口很好地结合在一起,需要使用 auto_close 通过将其设置为 False . 默认情况下,读取所有字节后关闭HTTP响应,这将禁用该行为:

import io
import urllib3

resp = urllib3.request("GET", "https://example.com", preload_content=False)
resp.auto_close = False

for line in io.TextIOWrapper(resp):
    print(line)
# <!doctype html>
# <html>
# <head>
# ....
# </body>
# </html>

请求数据#

报头#

中将标头指定为词典。 headers 中的论点 request()

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/headers",
    headers={
        "X-Something": "value"
    }
)

print(resp.json()["headers"])
# {"X-Something": "value", ...}

或者,您可以使用 HTTPHeaderDict 类以创建多值HTTP头:

import urllib3

# Create an HTTPHeaderDict and add headers
headers = urllib3.HTTPHeaderDict()
headers.add("Accept", "application/json")
headers.add("Accept", "text/plain")

# Make the request using the headers
resp = urllib3.request(
    "GET",
    "https://httpbin.org/headers",
    headers=headers
)

print(resp.json()["headers"])
# {"Accept": "application/json, text/plain", ...}

曲奇饼#

Cookie是使用 Cookie 标头中包含包含 ; 分隔的键值对:

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/cookies",
    headers={
        "Cookie": "session=f3efe9db; id=30"
    }
)

print(resp.json())
# {"cookies": {"id": "30", "session": "f3efe9db"}}

服务器提供的Cookie存储在 Set-Cookie 标题:

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/cookies/set/session/f3efe9db",
    redirect=False
)

print(resp.headers["Set-Cookie"])
# session=f3efe9db; Path=/

查询参数#

GETHEAD ,以及 DELETE 请求时,只需将参数作为字典传递到 fields 参数为 request()

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/get",
    fields={"arg": "value"}
)

print(resp.json()["args"])
# {"arg": "value"}

为了 POSTPUT 请求时,需要在URL中手动编码查询参数:

from urllib.parse import urlencode
import urllib3

# Encode the args into url grammar.
encoded_args = urlencode({"arg": "value"})

# Create a URL with args encoded.
url = "https://httpbin.org/post?" + encoded_args
resp = urllib3.request("POST", url)

print(resp.json()["args"])
# {"arg": "value"}

表单数据#

PUTPOST 请求时,urllib3将自动对 fields 参数提供给 request()

import urllib3

resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    fields={"field": "value"}
)

print(resp.json()["form"])
# {"field": "value"}

JSON#

要在请求正文中发送JSON,请在 json 参数为 request() Urllib3将使用 json 模块,带 UTF-8 编码。此外,当 json 是提供的,则 "Content-Type" 在标头中设置为 "application/json" 未另有说明的。

import urllib3

resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    json={"attribute": "value"},
    headers={"Content-Type": "application/json"}
)

print(resp.json())
# {'headers': {'Content-Type': 'application/json', ...},
#  'data': '{"attribute":"value"}', 'json': {'attribute': 'value'}, ...}

文件和二进制数据#

用于使用 multipart/form-data 编码可以使用相同的方法 表单数据 并将文件字段指定为 (file_name, file_data)

import urllib3

# Reading the text file from local storage.
with open("example.txt") as fp:
    file_data = fp.read()

# Sending the request.
resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    fields={
       "filefield": ("example.txt", file_data),
    }
)

print(resp.json()["files"])
# {"filefield": "..."}

虽然不严格要求指定文件名,但建议您使用该文件名以匹配浏览器行为。您也可以传递元组中的第三个项来显式指定文件的MIME类型:

resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    fields={
        "filefield": ("example.txt", file_data, "text/plain"),
    }
)

对于发送原始二进制数据,只需指定 body 争论。也建议设置 Content-Type 标题:

import urllib3

with open("/home/samad/example.jpg", "rb") as fp:
    binary_data = fp.read()

resp = urllib3.request(
    "POST",
    "https://httpbin.org/post",
    body=binary_data,
    headers={"Content-Type": "image/jpeg"}
)

print(resp.json()["data"])
# data:application/octet-stream;base64,...

证书验证#

备注

版本1.25中的新增功能:

现在默认情况下会验证HTTPS连接 (cert_reqs = "CERT_REQUIRED" )。

而您可以通过设置禁用证书验证 cert_reqs = "CERT_NONE" ,强烈建议您将其保持打开状态。

除非另有规定,否则urllib3将尝试加载默认的系统证书存储。最可靠的跨平台方法是使用 certifi 提供Mozilla根证书捆绑包的包:

$ python -m pip install certifi

一旦有证书,就可以创建一个 PoolManager 在发出请求时验证证书:

import certifi
import urllib3

http = urllib3.PoolManager(
    cert_reqs="CERT_REQUIRED",
    ca_certs=certifi.where()
)

这个 PoolManager 将自动处理证书验证并引发 SSLError 如果验证失败:

import certifi
import urllib3

http = urllib3.PoolManager(
    cert_reqs="CERT_REQUIRED",
    ca_certs=certifi.where()
)

http.request("GET", "https://httpbin.org/")
# (No exception)

http.request("GET", "https://expired.badssl.com")
# urllib3.exceptions.SSLError ...

备注

如果需要,可以使用OS提供的证书。只需将证书捆绑包的完整路径指定为 ca_certs 参数而不是 certifi.where() . 例如,大多数linux系统将证书存储在 /etc/ssl/certs/ca-certificates.crt . 其他操作系统可以 difficult .

使用超时#

超时允许您控制在中止请求之前允许运行多长时间(以秒为单位)。在简单的情况下,您可以将超时指定为 floatrequest()

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/delay/3",
    timeout=4.0
)

print(type(resp))
# <class "urllib3.response.HTTPResponse">

# This request will take more time to process than timeout.
urllib3.request(
    "GET",
    "https://httpbin.org/delay/3",
    timeout=2.5
)
# MaxRetryError caused by ReadTimeoutError

对于更细粒度的控件,可以使用 Timeout 实例,允许您指定单独的连接和读取超时:

import urllib3

resp = urllib3.request(
    "GET",
    "https://httpbin.org/delay/3",
    timeout=urllib3.Timeout(connect=1.0)
)

print(type(resp))
# <urllib3.response.HTTPResponse>

urllib3.request(
    "GET",
    "https://httpbin.org/delay/3",
    timeout=urllib3.Timeout(connect=1.0, read=2.0)
)
# MaxRetryError caused by ReadTimeoutError

如果您希望所有请求都受到相同的超时,则可以在 PoolManager 水平:

import urllib3

http = urllib3.PoolManager(timeout=3.0)

http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=1.0, read=2.0)
)

您仍然可以通过指定以下内容来覆盖此池级别超时 timeoutrequest()

正在重试请求#

Urllib3可以自动重试幂等请求。同样的机制也可以处理重定向。您可以使用 retries 参数设置为 request() 。默认情况下,urllib3会重试3次请求,最多跟踪3次重定向。

要更改重试次数,只需指定一个整数:

import urllib3

urllib3.request("GET", "https://httpbin.org/ip", retries=10)

要禁用所有重试和重定向逻辑,请指定 retries=False

import urllib3

urllib3.request(
    "GET",
    "https://nxdomain.example.com",
    retries=False
)
# NewConnectionError

resp = urllib3.request(
    "GET",
    "https://httpbin.org/redirect/1",
    retries=False
)

print(resp.status)
# 302

要禁用重定向但保留重试逻辑,请指定 redirect=False

resp = urllib3.request(
    "GET",
    "https://httpbin.org/redirect/1",
    redirect=False
)

print(resp.status)
# 302

对于更细粒度的控件,可以使用 Retry 实例。这个类允许您更大程度地控制请求的重试方式。

例如,要执行总共3次重试,但仅限于2次重定向:

urllib3.request(
    "GET",
    "https://httpbin.org/redirect/3",
    retries=urllib3.Retry(3, redirect=2)
)
# MaxRetryError

您也可以禁用太多重定向的异常,只返回 302 回应:

resp = urllib3.request(
    "GET",
    "https://httpbin.org/redirect/3",
    retries=urllib3.Retry(
        redirect=2,
        raise_on_redirect=False
    )
)

print(resp.status)
# 302

如果希望所有请求都受同一重试策略的约束,则可以在 PoolManager 水平:

import urllib3

http = urllib3.PoolManager(retries=False)

http = urllib3.PoolManager(
    retries=urllib3.Retry(5, redirect=2)
)

您仍然可以通过指定以下内容来覆盖此池级别重试策略 retriesrequest()

错误和异常#

urllib3包装较低级别的异常,例如:

import urllib3

try:
    urllib3.request("GET","https://nx.example.com", retries=False)

except urllib3.exceptions.NewConnectionError:
    print("Connection failed.")
# Connection failed.

exceptions 所有异常的完整列表。

登录中#

如果您使用的是标准库 logging urllib3模块将发出多个日志。在某些情况下,这可能是不可取的。您可以使用标准记录器界面更改urllib3的记录器的日志级别:

logging.getLogger("urllib3").setLevel(logging.WARNING)