接口测试

本章列举了一些直接发起请求的模块，这些模块在旧时代 HTTP1.0 时期，非 Js 渲染时代是我们获取网页数据的利器。

现在是HTTP2.0（简称 h2）以及以 VUE 为代表的 Js 渲染页面时代。因此这些模块现在主要用于获取直接的数据，譬如：图片、视频、纯静态的 HTML 网页或 XHR(XMLHttpRequest)类型的请求。

请求的方法与 HTTP 协议对资源的操作对应

(1) GET：请求获取 URL 位置的资源 (2) HEAD：请求获取 URL 位置资源的响应消息报告，即获得该资源的头部信息 (3) POST：请求向 URL 位置的资源后附加新的数据 (4) PUT：请求向 URL 位置存储一个资源，覆盖原 URL 位置的资源 (5) PATCH：请求局部更新 URL 位置的资源，即更改该处资源的部分内容 (6) DELETE：请求删除 URL 位置存储的资源 (特殊)options ：向服务器获取一些服务器和客户端能够打交道的参数，但与获取资源并不直接相关

urllib 模块：仅支持 HTTP1.0 仅同步，但是解码和解析功能是真的很好用

request 模块：仅支持 HTTP1.0 仅同步是基于 urllib3 的再次封装，因此语法更简单

aiohttp 模块：支持 HTTP2.0 仅异步

httpx 模块：支持 HTTP1.1/2.0 同步异步，双卡双待，是直接发起请求的模块中最全能的模块。

urllib

urllib 是一个收集了多个涉及 URL 的模块的自带包：可以打开和读取 URL、抛出异常、解析 URL、解析 robots.txt 文件是最底层的模块。

urllib 模块代码文档

urllib 发送请求

import urllib.request

url = 'https://www.python.org'
# 方式一
response = urllib.request.urlopen(url)
print(type(response))  # <class 'http.client.HTTPResponse'>
# 方式二
request = urllib.request.Request(url)
res = urllib.request.urlopen(url)
print(type(res))  # <class 'http.client.HTTPResponse'>
print(response.read())                  # 获取响应体 二进制字符串
print(response.getheaders())
## 结果为
[('Connection', 'close'), ('Content-Length', '50064'), ('Server', 'nginx'), ('Content-Type', 'text/html; charset=utf-8'), ('X-Frame-Options', 'DENY'), ('Via', '1.1 vegur, 1.1 varnish, 1.1 varnish'), ('Accept-Ranges', 'bytes'), ('Date', 'Tue, 17 Jan 2023 14:37:33 GMT'), ('Age', '1938'), ('X-Served-By', 'cache-iad-kiad7000025-IAD, cache-nrt-rjtf7700057-NRT'), ('X-Cache', 'HIT, HIT'), ('X-Cache-Hits', '263, 1190'), ('X-Timer', 'S1673966254.566369,VS0,VE0'), ('Vary', 'Cookie'), ('Strict-Transport-Security', 'max-age=63072000; includeSubDomains')]

urllib 异常处理

URLError 是 OSError 的一个子类，所有请求问题都会被捕获。

HTTPError 是 URLError 的一个子类，服务器上 HTTP 的响应会返回一个状态码，根据这个 HTTP 状态码来决定是否捕获，比如常见的 404 错误等。

from urllib import request
from urllib import error

if __name__ == "__main__":
    url = "http://www.iloveyou.com/"#一个不存在的连接
    req = request.Request(url)
    try:
        response = request.urlopen(req)
        print(response.read())
    except error.URLError as e:
        print(e) # <urlopen error [Errno 11002] getaddrinfo failed>

urllib 解析 URL

你肯定经历过复制网址出现乱码，这是因为网址必须以通用码的形式传送，而且还要避免几个特殊字符，因此网址要经编码，汉字经过编码后自然就是不可辨认的乱码了。

那么浏览器的地址栏中，网址为什么看起来是中文呢？这大概是浏览器的“人性化”处理，将编码好的中文网址还原出来“暂时”显示在地址栏中。

知道原理就能清楚的解码啦，你可以通过 encode 和 decode 方法进行操作解码和转码，只不过要考虑汉字中有%等特殊字符和/x 与%互转的情况，所以，直接用 quote 函数吧，别重复造轮子。

from urllib.parse import unquote
from urllib.parse import quote

url = 'https://www.baidu.com/s?ie=UTF-8&wd=%E7%A7%91%E6%8A%80&%E6%8A%80%E6%9C%AF'
print(unquote(url))
# 结果为https://www.baidu.com/s?ie=UTF-8&wd=科技&技术


print( 'https://www.baidu.com/s?ie=UTF-8&wd='+quote('科技&技术'))
# 结果为'https://www.baidu.com/s?ie=UTF-8&wd=%E7%A7%91%E6%8A%80&%E6%8A%80%E6%9C%AF'

urllib 解析 robots.txt 文件

import urllib.robotparser
rp = urllib.robotparser.RobotFileParser()
rp.set_url("http://www.musi-cal.com/robots.txt")
rp.read()

print(rp.can_fetch("*", "http://www.musi-cal.com/")) #判断网页是否可以抓取，'*'表示适用于所有爬虫
# True

requests

requests 是基于 urllib3 开发的一个优雅而简单的 PythonHTTP 库，为人类构建。

requests 模块官方文档

requests 发送请求

import requests
r = requests.get('https://api.github.com/events')
r = requests.post('https://httpbin.org/post', data={'key': 'value'})
r = requests.put('https://httpbin.org/put', data={'key': 'value'})
r = requests.delete('https://httpbin.org/delete')
r = requests.head('https://httpbin.org/get')
r = requests.patch('http://httpbin.org/patch',data ={'key': 'value'})
r = requests.options('https://httpbin.org/get')

requests 异常处理

Requests 库通过 raise_for_status()引发支持六种常用的连接异常

(1) requests.ConnectionError：网络连接错误异常，如 DNS 查询失败、拒绝连接等 (2) requests.HTTPError：HTTP 错误异常 (3) requests.URLRequired：URL 缺失异常 (4) requests.TooManyRedirects：超过最大重定向次数，产生重定向异常 (5) requests.ConnectTimeout：连接远程服务器超时异常 (6) requests.Timeout：请求 URL 超时，产生超时异常（发出请求到获得内容的整个过程的超时异常）

import requests
try:
    r=requests.get("https://www.baidu.com/error",timeout=30) #请求一个url链接
    r.raise_for_status() #如果状态不是200，引发HTTPError异常
except requests.HTTPError as e:
    print(e)# 404 Client Error: Not Found for url: https://www.baidu.com/error

aiohttp

aiohttp 模块官方文档

aiohttp 发起请求

import aiohttp
import asyncio

async def main():
    async with aiohttp.ClientSession('http://httpbin.org') as session:
        async with session.post('/post', data=b'data'):
            pass
        async with session.put('/put', data=b'data'):
            pass
        async with session.get('/get') as resp:
            print(resp.status)
            print(await resp.text())

asyncio.run(main())

httpx

HTTPX 是 Python 3 的全功能 HTTP 客户端，它提供同步和异步 API，并支持 HTTP / 1.1 和 HTTP / 2。配置代理和请求头也很容易。httpx 还有 web 测试模式、精细化配置代理 IP、Event Hooks、SSL 证书等语法，但是因为 VUE 时代的到来，这些实现都有了新的方案，所以不再过多介绍。

httpx 代码文档

httpxH2 检测

import httpx
import asyncio

'''
检查 HTTP 版本
在客户端上启用 HTTP/2 支持并不一定意味着您的 请求和响应将通过 HTTP/2 传输，因为客户端和服务器都需要支持 HTTP/2。如果连接到仅 支持 HTTP/1.1 客户端将使用标准的 HTTP/1.1 连接。

您可以通过检查来确定使用了哪个版本的 HTTP 协议 响应上的属性。
'''
async def main():
        client = httpx.AsyncClient(http2=True)
        response = await client.get('https://www.python-httpx.org/')
        print(response.http_version)  # "HTTP/1.0", "HTTP/1.1", or "HTTP/2".
asyncio.run(main())

httpx 发起请求

import httpx

## HTTP / 1.1 同步请求模式A
r = httpx.post('https://httpbin.org/post', data={'key': 'value'})
r = httpx.put('https://httpbin.org/put', data={'key': 'value'})
r = httpx.delete('https://httpbin.org/delete')
r = httpx.head('https://httpbin.org/get')
r = httpx.options('https://httpbin.org/get')
r = httpx.patch('https://httpbin.org/get')
r = httpx.get('https://www.example.org/')

r # <Response [200 OK]>
r.status_code # 200
r.headers['content-type'] # r.headers['content-type']
r.text #'<!doctype html>...</html>'

## HTTP / 1.1 同步请求模式B
client = httpx.Client()
r = client.post('https://httpbin.org/post', data={'key': 'value'})
r = client.put('https://httpbin.org/put', data={'key': 'value'})
r = client.delete('https://httpbin.org/delete')
r = client.head('https://httpbin.org/get')
r = client.options('https://httpbin.org/get')
r = client.patch('https://httpbin.org/get')
r = client.get('https://www.example.org/')
client.close()


## 异步请求模式A
async def maina():
    async with httpx.AsyncClient() as client:
        r = await client.post('https://httpbin.org/post', data={'key': 'value'})
        r = await client.put('https://httpbin.org/put', data={'key': 'value'})
        r = await client.delete('https://httpbin.org/delete')
        r = await client.head('https://httpbin.org/get')
        r = await client.options('https://httpbin.org/get')
        r = await client.patch('https://httpbin.org/get')
        r = await client.get('https://www.example.com/')
    print(r)  #<Response [200 OK]>
asyncio.run(maina())

## 异步请求模式B
async def mainb():
    client = httpx.AsyncClient()
    r = await client.post('https://httpbin.org/post', data={'key': 'value'})
    r = await client.put('https://httpbin.org/put', data={'key': 'value'})
    r = await client.delete('https://httpbin.org/delete')
    r = await client.head('https://httpbin.org/get')
    r = await client.options('https://httpbin.org/get')
    r = await client.patch('https://httpbin.org/get')
    r = await client.get('https://www.example.com/')
    await client.aclose()
    print(r)  #<Response [200 OK]>
asyncio.run(mainb())

## H2模式下同步与异步请求
client = httpx.Client(http2=True)
client = httpx.AsyncClient(http2=True)

## 配置合并
headers = {'X-Auth': 'from-client'}
params = {'client_id': 'client1'}
with httpx.Client(headers=headers, params=params,http2=True,proxies="http://localhost:8030") as client:
    headers = {'X-Custom': 'from-request'}
    params = {'request_id': 'request1'}
    r = client.get('https://example.com', headers=headers, params=params)

r.request.url # URL('https://example.com?client_id=client1&request_id=request1')
r.request.headers['X-Auth'] # 'from-client'
r.request.headers['X-Custom'] # 'from-request'

httpx 高级用法

如果需要监视大型响应的下载进度，可以使用响应流式处理并检查属性。response.num_bytes_downloaded

import tempfile
import httpx
import rich.progress

with tempfile.NamedTemporaryFile() as download_file:
    url = "https://speed.hetzner.de/100MB.bin"
    with httpx.stream("GET", url) as response:
        total = int(response.headers["Content-Length"])

        with rich.progress.Progress(
            "[progress.percentage]{task.percentage:>3.0f}%",
            rich.progress.BarColumn(bar_width=None),
            rich.progress.DownloadColumn(),
            rich.progress.TransferSpeedColumn(),
        ) as progress:
            download_task = progress.add_task("Download", total=total)
            for chunk in response.iter_bytes():
                download_file.write(chunk)
                progress.update(download_task, completed=response.num_bytes_downloaded)

接口测试​

urllib​

urllib 发送请求​

urllib 异常处理​

urllib 解析 URL​

urllib 解析 robots.txt 文件​

requests​

requests 发送请求​

requests 异常处理​

aiohttp​

aiohttp 发起请求​

httpx​

httpxH2 检测​

httpx 发起请求​

httpx 高级用法​