requests

py 中最受欢迎的请求库。也是大多数软件测试人员做接口自动化测试的首选库。

pip install requests

# 如果使用 uv
uv add requests

为了练习方便，文末提供一份基于 Flask 创建的最小服务，会将你传递的参数原样返回给你。

基本使用

以 get 请求为例。该库提供了 get | post | put | delete | options | head 等常用方法。

调用后传递对应的参数，返回 Response 对象。

import requests

r = requests.get("https://jsonplaceholder.typicode.com/posts")
print(r.status_code)
print(r.text)

请求方法的参数

http 请求所包含的参数众多，常规情况无非就是几种 params 和 data。其中 data 包含两种：json 和 form。

然后就是 headers，cookies 等一些关于标识和权限的数据。

import requests
import json

r = requests.get(
    "http://127.0.0.1:5000/collect", # url
    params={
      "param1": "value1",
      "param2": "value2",
    },
    cookies={ "title": "cookies" },
    headers={ 
      "Authorization": "Bearer token"
    }
)
# 格式化输出
print(json.dumps(r.json(), indent=4))

# 要理解 params 和 data 以及 json 的区别
# 无论是什么请求，params 的参数都会被拼接到 url 上
# 而 data 和 json 的参数则会作为请求体
# http://127.0.0.1:5000/collect?param1=value1&param2=value2
print(r.url)

默认情况下，如果只传递 data 参数会被当做 formData 的类型传递到后端。如果需要传递 json 有两种方式。

import requests
import json

payload = {
    "name": "Alice",
    "age": 18
}

# 发送 formData 类型数据到后端
# requests 会自动携带 Content-Type: application/x-www-form-urlencoded 这样的请求头
r = requests.post(
    "http://127.0.0.1:5000/collect",
    data=payload
)
print(json.dumps(r.json(), indent=4))

# 方案一：使用 json.dumps 将参数格式化成 json，并且配合 headers 传递
r = requests.post(
    "http://127.0.0.1:5000/collect",
    headers={ 
      "Content-Type": "application/json"
    },
    data=json.dumps(payload)
)
print(json.dumps(r.json(), indent=4))

# 方案二：使用 json 参数传递数据
r = requests.post(
    "http://127.0.0.1:5000/collect",
    json=payload
)
print(json.dumps(r.json(), indent=4))

如果碰到一些比较复杂的请求，则可以设置超时时间，或者禁止重定向。

比如你给每个请求都设置超时时间，超过界限后不再等服务器响应了。

比如你去访问一些接口时，他们会将你的请求重定向，从 http 转到 https。这时你可以设置 allow_redirects=False 来禁止重定向。

import requests

r = requests.get(
    "http://127.0.0.1:5000/collect",
    allow_redirects=False,
    timeout=5 # 5s
)
print(r.status_code)

文件上传

准确来说，文件上传也是发送请求的一部分，只不过需要特殊处理，从而单独分开演示一下。上传文件时默认使用的请求头是 Content-Type: multipart/form-data;

import requests
import json

r = requests.post(
    "http://127.0.0.1:5000/collect",
    # file-field 是后端接收文件的字段名
    # 文件名取的就是读取的文件名
    files={ "file-field": open("test.txt", "rb") }
)
print(json.dumps(r.json(), indent=4))

# 可以通过传递元组的方式，指定文件名和文件头
r = requests.post(
    "http://127.0.0.1:5000/collect",
    files={ "file-field": ("xxx.txt", open("test.txt", "rb"), 'text/plain') }
)
print(json.dumps(r.json(), indent=4))

响应对象

print(r.text) # 文本内容
print(r.json()) # json 内容
print(r.headers) # 接口响应的 headers
print(r.status_code) # 接口响应状态码
print(r.cookies.get_dict()) # 服务器返回的 cookie
print(r.encoding) # 响应的编码
print(r.raw()) # bytes 类型文本内容 一般用于返回流数据的接口

print(r.request.body) # 请求体
print(r.request.headers) # 请求头
print(r.request.method) # 请求方法
print(r.request.url) # 请求 url

异常处理

requests 出现异常时，会抛出 requests.exceptions.RequestException 异常。 这是一个基类，细分下来还包括

ConnectionError  如果出现网络问题（例如 DNS 故障、连接被拒绝等）
HTTPError      如果 HTTP 响应状态码不在 200-299 范围内
Timeout        如果请求超时
TooManyRedirects 如果重定向次数超过限制
JSONDecodeError  如果 JSON 解析失败，比如后端返回 204 No Content
RequestException 所有异常的基类

也可以通过 r.raise_for_status() 来主动抛出异常。只要状态码不对就会抛出异常

二次封装

二次封装的意义就是让用户使用起来更方便（当然这不是必须的，一般都是针对当前业务的场景来做这件事情）。

从表面上看 requests 已经很方便了，但是针对异常、超时、以及传递参数还有优化空间。

import requests

class Request:

    """
    @param base_url: 基础 url, 通常项目内大多数的接口都是一个域名, 只是路径不同
    @param timeout: 超时时间, 设置统一的超时时间, 避免单个接口设置超时
    @param default_headers: 默认 headers, 可以存放一些通用的 headers, 例如 token 等
    """
    def __init__(self, base_url, timeout = 30, default_headers = {}):
        self.base_url = base_url
        self.timeout = timeout
        self.default_headers = default_headers

    # 将路径和 base_url 组装成完整 url
    def _build_url(self, url):
        return self.base_url + url
    
    # 将用户传递的 headers 和默认的 headers 合并
    def _build_headers(self, headers):
        new_headers = self.default_headers.copy()
        if (headers):
            new_headers.update(headers)
        return new_headers
            
    # 构建一个错误的对象，然后判断错误类型，并最后将这个错误返回
    def _handler_error(self, error, url, method):
        error_info = {
            "error": True,
            "url": url,
            "method": method,
            "error_type": type(error).__name__,
            "error_message": str(error),
            "error_description": ""
        }

        if isinstance(error, Timeout):
            error_info['error_description'] = '请求超时'
        elif isinstance(error, ConnectionError):
            error_info['error_description'] = '连接错误'
        elif isinstance(error, HTTPError):
            error_info['error_description'] = 'HTTP错误'
        else:
            error_info['error_description'] = '请求过程中发生错误'

        return error_info

    def request(self, method, url, **kwargs):
        full_url = self._build_url(url)
        headers = self._build_headers(kwargs.get('headers'))
        
        # 设置默认的超时时间
        if ('timeout' not in kwargs):
            kwargs['timeout'] = self.timeout

        try:
            # requests.request() 可以处理所有类型的请求
            r = requests.request(method, full_url, headers=headers, **kwargs)
            r.raise_for_status()
            return r
        except Exception as e:
            return self._handler_error(e, full_url, method)

    
    def get(self, url, params = None, **kwargs):
        return self.request('GET', url, params=params, **kwargs)

    def post(self, url, data = None, json = None, **kwargs):
        return self.request('POST', url, data=data, json=json, **kwargs)

    def put(self, url, data = None, json = None, **kwargs):
        return self.request('PUT', url, data=data, json=json, **kwargs)

    def delete(self, url, **kwargs):
        return self.request('DELETE', url, **kwargs)

    def patch(self, url, data = None, json = None, **kwargs):
        return self.request('PATCH', url, data=data, json=json, **kwargs)

    def upload_file(self, url, file_path, file_name, **kwargs):
        try:
            with open(file_path, 'rb') as f:
                files = { file_name: f }
                return self.post(url, files=files, **kwargs)
        except (IOError, OSError) as e:
            return self._handler_error(e, self._build_url(url), 'POST')
        

if __name__ == '__main__':
    client = Request('http://127.0.0.1:5000')
    
    # get
    # r = client.get('/collect', params={ 'param1': 'value1' })
    # print(r.json())
    
    # post
    # r = client.post('/collect', json={ 'param1': 'value1' })
    # print(r.json())
    
    # upload file
    r = client.upload_file('/collect', 'test.txt', 'file')
    print(r.json())

服务代码示例

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/collect', methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH'])
def collect_data():
    # 收集所有类型的数据
    result = {
        'method': request.method,
        'headers': dict(request.headers),
        'args': dict(request.args),  # 查询参数
        'form': dict(request.form),  # 表单数据
        'json': request.get_json(silent=True) or {},  # JSON数据
        'files': {key: {
            'filename': file.filename,
            'content_type': file.content_type,
            'size': len(file.read())
        } for key, file in request.files.items()},  # 文件信息
        'cookies': dict(request.cookies),  # Cookies
        'remote_addr': request.remote_addr  # 客户端IP
    }
    
    return jsonify(result)

if __name__ == '__main__':
    app.run(debug=True)