Python接口调用实战指南：requests库处理HTTP RESTful API详解

软件开发中，应用程序经常需要与外部服务进行数据交互，例如获取天气信息、调用第三方API或与后端服务器通信。Python凭借其简洁的语法和丰富的库生态，成为实现此类任务的理想选择，尤其适合快速开发和后端服务集成。

本文将详细介绍如何使用Python调用远程接口，涵盖基本方法、适用场景及具体代码示例。

Requests模块

Python中调用远程接口最常用、最强大的库是requests。它提供了简洁的API来处理HTTP请求，支持GET、POST等多种方法，并能轻松处理参数、头部信息、Cookies等。

安装Requests模块

在开始之前，需要确保已安装requests库。可以通过Python的包管理工具pip进行安装：

pip install requests

常见应用场景

1、数据获取与聚合：从多个公开API（如天气、新闻、金融数据）获取信息，整合到自己的应用中。

2、微服务通信：在微服务架构中，服务之间通过HTTP API进行通信。

3、第三方服务集成：集成如支付网关（支付宝、微信支付）、短信服务、地图服务等。

4、自动化运维与监控：通过调用监控系统的API获取服务器状态，或自动创建工单。

5、Web爬虫与数据采集：以程序化方式访问网页或API来收集数据。

GET方法

GET方法通常用于从服务器获取数据。它通过URL传递参数，适合查询操作。

关键参数

• url: 远程接口的地址。
• params: 一个字典，包含要发送的查询参数。
• headers: 一个字典，包含HTTP请求头信息，如认证令牌、内容类型等。

代码示例

以下示例展示了如何使用GET方法调用一个假设的用户信息查询接口：

# -*- coding: utf-8 -*-
import requests
import ast

# 接口地址
url = 'https://api.example.com/user/info'

# GET传参
params = {
    'user_id': '12345'
}

# headers信息，通常用于认证和指定内容格式
headers = {
    'Content-Type': 'application/x-www-form-urlencoded',
    'Authorization': 'Bearer your_access_token_here'
}

# 发送GET请求
response = requests.get(url, params=params, headers=headers)

# 检查HTTP状态码，200表示成功
print(f"状态码: {response.status_code}")

# 获取返回的原始文本内容
content_text = response.text
print(f"原始文本响应: {content_text}")

# 如果返回的是JSON字符串，可以将其转换为Python字典
# 方法1: 使用ast.literal_eval（适用于安全的字符串）
content_dict = ast.literal_eval(content_text)
print(f"转换后的字典: {content_dict}")

# 方法2: 直接使用response.json()（更推荐，前提是响应是JSON格式）
content_json = response.json()
print(f"JSON响应: {content_json}")

POST方法

POST方法通常用于向服务器提交数据，例如创建新资源或提交表单。数据通常放在请求体中。

关键参数

• url: 远程接口的地址。
• data: 一个字典，包含要发送的表单数据。
• json: 一个字典，如果要发送JSON格式的数据，使用此参数。
• headers: 请求头信息。

代码示例

以下示例展示了如何使用POST方法向一个用户注册接口提交数据：

# -*- coding: utf-8 -*-
import requests
import ast

# 接口地址
url = 'https://api.example.com/user/register'

# headers信息
headers = {
    'Content-Type': 'application/x-www-form-urlencoded',
    'Authorization': 'Bearer your_access_token_here'
}

# POST传参（表单数据）
form_data = {
    'nickname': 'PythonUser',
    'gender': 1,
    'city': 'Beijing',
    'avatar': 'https://avatar.example.com/default.jpg'
}

# 发送POST请求
response = requests.post(url, data=form_data, headers=headers)

# 处理响应
print(f"状态码: {response.status_code}")
content_text = response.text
print(f"原始文本响应: {content_text}")

# 转换响应内容
try:
    content_dict = ast.literal_eval(content_text)
    print(f"转换后的字典: {content_dict}")
except:
    print("响应内容无法转换为字典")

# 如果接口返回JSON，直接使用.json()
if response.headers.get('Content-Type', '').startswith('application/json'):
    print(f"JSON响应: {response.json()}")

PUT方法

PUT方法通常用于更新或替换服务器上的现有资源。

# -*- coding: utf-8 -*-
import requests

# 接口地址 - 更新用户信息
url = 'https://api.example.com/user/12345'

# 要更新的数据
update_data = {
    'nickname': 'UpdatedUser',
    'city': 'Shanghai',
    'age': 28
}

# 请求头
headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer your_access_token_here'
}

# 发送PUT请求
response = requests.put(url, json=update_data, headers=headers)

print(f"状态码: {response.status_code}")
if response.status_code == 200:
    print("更新成功")
    print(f"响应内容: {response.json()}")

DELETE方法

DELETE方法用于删除服务器上的资源。

# -*- coding: utf-8 -*-
import requests

# 要删除的资源地址
url = 'https://api.example.com/user/12345'

headers = {
    'Authorization': 'Bearer your_access_token_here'
}

# 发送DELETE请求
response = requests.delete(url, headers=headers)

print(f"状态码: {response.status_code}")
if response.status_code == 204:  # 204 No Content是常见的删除成功状态码
    print("删除成功")

PATCH方法

PATCH方法用于对资源进行部分更新，而不是像PUT那样完全替换。

# -*- coding: utf-8 -*-
import requests

url = 'https://api.example.com/user/12345'

# 只更新部分字段
partial_update = {
    'city': 'Guangzhou'
}

headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer your_access_token_here'
}

response = requests.patch(url, json=partial_update, headers=headers)

print(f"状态码: {response.status_code}")
if response.status_code == 200:
    print("部分更新成功")
    print(f"响应内容: {response.json()}")

其他场景

文件上传

使用POST方法上传文件到服务器。

# -*- coding: utf-8 -*-
import requests

url = 'https://api.example.com/upload'

# 要上传的文件
files = {
    'file': open('example.jpg', 'rb'),
    'description': (None, '这是一个示例图片')
}

# 其他表单数据
data = {
    'category': 'avatar',
    'user_id': '12345'
}

response = requests.post(url, files=files, data=data)

print(f"状态码: {response.status_code}")
if response.status_code == 200:
    print("文件上传成功")
    print(f"响应: {response.json()}")

流式请求

处理大文件或流式数据。

# -*- coding: utf-8 -*-
import requests

url = 'https://api.example.com/large-data'

# 流式下载大文件
response = requests.get(url, stream=True)

# 分块保存文件
with open('large_file.dat', 'wb') as f:
    for chunk in response.iter_content(chunk_size=8192):
        if chunk:
            f.write(chunk)

print("文件下载完成")

处理Cookies和会话

# -*- coding: utf-8 -*-
import requests

# 创建会话对象
session = requests.Session()

# 登录获取cookies
login_url = 'https://api.example.com/login'
login_data = {
    'username': 'user@example.com',
    'password': 'your_password'
}
login_response = session.post(login_url, data=login_data)
print(f"登录状态码: {login_response.status_code}")

# 使用同一个会话访问需要认证的接口
profile_url = 'https://api.example.com/profile'
profile_response = session.get(profile_url)
print(f"获取个人信息状态码: {profile_response.status_code}")
if profile_response.status_code == 200:
    print(f"个人信息: {profile_response.json()}")

异步请求处理

对于需要同时调用多个接口或处理大量请求的场景，可以使用异步方式提高效率。

使用aiohttp库（异步）

# -*- coding: utf-8 -*-
import aiohttp
import asyncio

async def fetch_data(session, url):
    async with session.get(url) as response:
        return await response.json()

async def main():
    urls = [
        'https://api.example.com/data1',
        'https://api.example.com/data2',
        'https://api.example.com/data3'
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_data(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

        for i, result in enumerate(results):
            print(f"接口{i+1}返回: {result}")

# 运行异步函数
# asyncio.run(main())

使用concurrent.futures（并发）

# -*- coding: utf-8 -*-
import requests
import concurrent.futures

def call_api(url):
    response = requests.get(url)
    return response.json()

urls = [
    'https://api.example.com/data1',
    'https://api.example.com/data2',
    'https://api.example.com/data3',
    'https://api.example.com/data4'
]

# 使用线程池并发调用
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(call_api, url): url for url in urls}

    for future in concurrent.futures.as_completed(futures):
        url = futures[future]
        try:
            data = future.result()
            print(f"{url} 返回数据: {data}")
        except Exception as e:
            print(f"{url} 调用失败: {e}")

自定义请求适配器

# -*- coding: utf-8 -*-
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 创建会话
session = requests.Session()

# 配置重试策略
retry_strategy = Retry(
    total=3,  # 总重试次数
    backoff_factor=1,  # 重试等待时间因子
    status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的状态码
    allowed_methods=["GET", "POST"]  # 允许重试的方法
)

# 创建适配器并应用到会话
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)

# 使用配置好的会话发送请求
response = session.get('https://api.example.com/data')

使用代理

# -*- coding: utf-8 -*-
import requests

proxies = {
    'http': 'http://10.10.1.10:3128',
    'https': 'http://10.10.1.10:1080',
}

# 通过代理发送请求
response = requests.get('https://api.example.com/data', proxies=proxies, timeout=10)

注意事项

1、错误处理：始终检查response.status_code，并使用try-except块处理网络异常。

try:
    response = requests.get(url, timeout=5)
    response.raise_for_status() # 如果状态码不是200，将抛出HTTPError异常
except requests.exceptions.RequestException as e:
    print(f"请求发生错误: {e}")

2、超时设置：为请求设置超时，避免程序长时间等待。

response = requests.get(url, timeout=(3.05, 27)) # (连接超时, 读取超时)

3、会话保持：如果需要多次调用同一接口并保持Cookies或会话，使用requests.Session()。

session = requests.Session()
session.headers.update({'User-Agent': 'MyApp/1.0'})
response = session.get(url)

4、安全考虑：

• 不要在代码中硬编码敏感信息（如API密钥）。使用环境变量或配置文件。
• 对于HTTPS接口， requests 默认会验证SSL证书。在生产环境中不应禁用验证。

5、处理JSON：当接口返回JSON时，优先使用response.json()方法进行解析，它更安全高效。

6、性能优化

（1）连接复用：使用requests.Session()复用TCP连接，减少连接建立开销。
（2）连接池配置：调整连接池大小以适应并发需求。
（3）响应压缩：如果服务器支持，使用gzip压缩减少传输数据量。
（4）缓存策略：对于不常变化的数据，实现客户端缓存。
（5）批量请求：如果API支持，尽量使用批量接口减少请求次数。

Python调用远程接口的方法从基本的GET/POST到复杂的异步并发处理，requests库及其生态提供了完整的解决方案。掌握这些方法后，您将能够应对各种远程接口调用场景，构建高效可靠的应用程序。在实际开发中，应根据具体需求选择合适的方法：

• 简单数据获取：使用GET方法
• 数据提交：使用POST方法
• 资源更新：使用PUT或PATCH方法
• 资源删除：使用DELETE方法
• 高性能场景：考虑异步或并发处理
• 复杂网络环境：配置重试机制和代理